FastAPI 0.100.0 and pydantic v2 compatible API migration

.gitignore

@@ -11,3 +11,4 @@
 __pycache__
 .vercel
 tempCodeRunnerFile.py
+.pytest_cache

API/__init__.py

@@ -11,11 +11,11 @@
 """
 
 from fastapi import FastAPI
-from API.utils.DBConnection import DBConnection
 
-app = FastAPI(title="Connecting Villages API",version="V0.1.0",description="API for Connecting Villages")
+app = FastAPI(title="Connecting Villages API",version="V0.2.0",description="API for Connecting Villages")
 
 from API import fwapp
+from API.utils.DBConnection import DBConnection
 
 # inits
 try:

API/core/ConfigEnv.py

@@ -2,7 +2,7 @@
 """
 from functools import lru_cache
 
-from pydantic import BaseSettings
+from pydantic_settings import BaseSettings
 
 
 class Settings(BaseSettings):
@@ -11,6 +11,14 @@ class Settings(BaseSettings):
     JWT_SECRET_KEY: str
     JWT_REFRESH_SECRET_KEY: str
 
+    ADMIN_ID: str
+    ADMIN_PWD: str
+    ADMIN_VILLAGE_NAME: str
+    ADMIN_ROLE: str
+    USER_ROLE: str
+    OWNER_ROLE: str
+
+
     class Config:
         env_file = ".env"
 

API/core/ExceptionHandlers.py

@@ -1,12 +1,13 @@
 """Exception Handlers to handle custom built exceptions.
 """
-from API import app
-from .Exceptions import *
-
 from fastapi.responses import JSONResponse
 from fastapi.requests import Request
 from fastapi import status
 
+from API import app
+
+from .Exceptions import *
+
 
 @app.exception_handler(VillageNotFoundException)
 async def handle_village_not_found(request: Request, exec: VillageNotFoundException):

API/core/Exceptions.py

@@ -18,8 +18,7 @@
 """
 import json
 
-from ..models.FrontendResponseSchema import FrontendResponseModel
-from ..models.AuthSchema import TokenSchema
+from API.models import TokenSchema, FrontendResponseModel
 
 
 class VillageNotFoundException(Exception):

API/fwapp.py

@@ -1,26 +1,27 @@
 """This module contains the routes for the API.It contains the functions 
 that are used to create the endpoints."""
-
 from API import app
-from API.services.DBManipulation import *
-from API.services.AuthServices import *
-from .models.RequestBodySchema import FormData
-from .models.AuthSchema import UserAuth, TokenSchema, UserOut, UseRefreshToken, BulkSignup
-from .utils.JWTBearer import JWTBearer
-from .utils import scopes
-from .core.ExceptionHandlers import *
-from .core.Exceptions import *
-
+from API.services.db import *
+from API.services.auth import *
+from API.services.auth.utils import JWTBearer
+from API.utils import scopes
+from API.core.ExceptionHandlers import *
+from API.core.Exceptions import *
+from API.models import (UserAuth, UserOut, UseRefreshToken,
+                        BulkSignup, FormData, TokenSchema, FrontendResponseModel)
+
+from fastapi import Depends, Request
 from fastapi.templating import Jinja2Templates
-from fastapi import Request, Depends
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
 
+import datetime
+
 # template and static files setup
 templates = Jinja2Templates(directory="API/templates/")
 app.mount("/static", StaticFiles(directory="API/static"), name="static")
 
-#Middleware to handle CORS (cross origin  resource sharing) error in the browser
+# Middleware to handle CORS (cross origin  resource sharing) error in the browser
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -29,8 +30,9 @@
     allow_headers=["*"],
 )
 
+
 @app.get("/", tags=["Home"])
-def home(request: Request):
+def _home(request: Request):
     return templates.TemplateResponse("home.html", context={"request": request})
 
 
@@ -49,6 +51,7 @@ def api_response_check():
             db_msg = "Connection failed to db"
 
         response_result["message"].append(db_msg)
+        response_result["data"]['timestamp'] = f"{datetime.datetime.now()}"
 
     except Exception as e:
         print("Exception :", e)
@@ -150,38 +153,39 @@ def scoped_checks(user_creds: UserOut):
     return response_result
 
 
-@app.post('/auth/signup', summary="Create new user", response_model=FrontendResponseModel, tags=["Authorization Server"],dependencies=[Depends(JWTBearer())])
-async def create_user(data: Union[UserAuth,BulkSignup],user_credentials: str = Depends(JWTBearer())):
+@app.post('/auth/signup', summary="Create new user", response_model=FrontendResponseModel,
+          tags=["Authorization Server"], dependencies=[Depends(JWTBearer())])
+async def auth_signup(data: Union[UserAuth, BulkSignup], user_credentials: str = Depends(JWTBearer())):
     response_result = {
         "status": "not_allowed",
         "message": ["Not authenticated"],
         "data": {},
     }
 
     user_creds = get_current_user_credentials(user_credentials)
-
 
     @scopes.init_checks(authorized_roles=['admin', 'GOVTOff'],
                         village_name=data.village_name, response_result=response_result)
     def scoped_checks(user_creds: UserOut):
-        if isinstance(data,UserAuth):
+        if isinstance(data, UserAuth):
             if data.role not in ['admin', 'user']:
                 raise AuthorizationFailedException(response_result, "not authorized")
-            
+
             if data.role == 'admin' and user_creds.role == 'admin':
-                    raise AuthorizationFailedException(response_result, "not authorized")
-        
+                raise AuthorizationFailedException(response_result, "not authorized")
+
         if user_creds.role == "admin" and data.village_name != user_creds.village_name:
             raise AuthorizationFailedException(response_result, "not authorized")
-        
+
     scoped_checks(user_creds)
-    
+
     signup(response_result, data)
     return response_result
 
 
-@app.post('/auth/login', summary="Log-in to the user account", response_model=TokenSchema, tags=["Authorization Server"])
-async def login(form_data: UserAuth = Depends()):
+@app.post('/auth/login', summary="Log-in to the user account", response_model=TokenSchema,
+          tags=["Authorization Server"])
+async def auth_login(form_data: UserAuth = Depends()):
     tokens = {
         "status": "Internal Server Error 500",
         "access_token": "",
@@ -197,9 +201,10 @@ async def login(form_data: UserAuth = Depends()):
 async def auth_use_refresh_token(existing_tokens: UseRefreshToken):
     return handle_refresh_token_access(existing_tokens.refresh_access_token)
 
-@app.get("/ops/get_village_list", summary="Get the list of village names", response_model=FrontendResponseModel, tags=["Sensitive ops"], dependencies=[Depends(JWTBearer())])
-async def get_village_list(user_credentials: str = Depends(JWTBearer())):
 
+@app.get("/ops/get_village_list", summary="Get the list of village names", response_model=FrontendResponseModel,
+         tags=["Sensitive ops"], dependencies=[Depends(JWTBearer())])
+async def get_village_list(user_credentials: str = Depends(JWTBearer())):
     response_result = {
         "status": "not_allowed",
         "message": ["Not authenticated"],
@@ -208,19 +213,21 @@ async def get_village_list(user_credentials: str = Depends(JWTBearer())):
 
     user_creds = get_current_user_credentials(user_credentials)
 
-    @scopes.init_checks(authorized_roles=['GOVTOff'],response_result=response_result)
+    @scopes.init_checks(authorized_roles=['GOVTOff'], response_result=response_result)
     def scoped_checks(user_creds: UserOut):
         pass
 
     scoped_checks(user_creds)
 
-    village_list=get_available_villages(response_result)
+    village_list = get_available_villages(response_result)
     response_result['data']["village_names"] = village_list
     return response_result
 
-#delete database route
-@app.delete('/ops/delete_database',summary="Delete the database", tags=["Sensitive ops"], dependencies=[Depends(JWTBearer())])
-async def delete_database(dbname:str,user_credentials: str = Depends(JWTBearer())):
+
+# delete database route
+@app.delete('/ops/delete_database', summary="Delete the database", tags=["Sensitive ops"],
+            dependencies=[Depends(JWTBearer())])
+async def ops_delete_database(dbname: str, user_credentials: str = Depends(JWTBearer())):
     response_result = {
         "status": "not_allowed",
         "message": ["Not authenticated"],
@@ -229,18 +236,19 @@ async def delete_database(dbname:str,user_credentials: str = Depends(JWTBearer()
 
     user_creds = get_current_user_credentials(user_credentials)
 
-    @scopes.init_checks(authorized_roles=['GOVTOff'],response_result=response_result,village_name=dbname)
+    @scopes.init_checks(authorized_roles=['GOVTOff'], response_result=response_result, village_name=dbname)
     def scoped_checks(user_creds):
         pass
 
     scoped_checks(user_creds)
 
-    delete_village_data(dbname,response_result)
+    delete_village_data(dbname, response_result)
     return response_result
 
 
-@app.put('/ops/update_village_list',summary="Update the village list", tags=["Sensitive ops"], dependencies=[Depends(JWTBearer())])
-async def update_village_list(dbname:str,user_credentials: str = Depends(JWTBearer())):
+@app.put('/ops/update_village_list', summary="Update the village list", tags=["Sensitive ops"],
+         dependencies=[Depends(JWTBearer())])
+async def ops_update_village_list(dbname: str, user_credentials: str = Depends(JWTBearer())):
     response_result = {
         "status": "not_allowed",
         "message": ["Not authenticated"],
@@ -249,13 +257,13 @@ async def update_village_list(dbname:str,user_credentials: str = Depends(JWTBear
 
     user_creds = get_current_user_credentials(user_credentials)
 
-    @scopes.init_checks(authorized_roles=['GOVTOff'],response_result=response_result)
+    @scopes.init_checks(authorized_roles=['GOVTOff'], response_result=response_result)
     def scoped_checks(user_creds):
         pass
 
     scoped_checks(user_creds)
 
-    create_new_village(dbname,user_creds,response_result)
+    create_new_village(dbname, user_creds, response_result)
     return response_result
 
 # @app.get('/auth/me', summary='Get details of currently logged in user', response_model=UserOut, tags=["SessionInfo"])

API/models/FrontendResponseSchema.py

@@ -1,6 +1,8 @@
 from pydantic import BaseModel
 
 from typing import List
+
+
 class FrontendResponseModel(BaseModel):
     status:str
     message: List[str]

API/models/__init__.py

@@ -0,0 +1,4 @@
+from .AuthSchema import *
+from .FrontendResponseSchema import *
+from .RequestBodySchema import *
+from .EDAResponseSchema import *

API/services/AuthServices.py → API/services/auth/AuthServices.py

@@ -7,11 +7,12 @@
 from jose import jwt
 from typing import Union
 
-from API.utils.Auth import Auth
-from ..core.ConfigEnv import settings
-from ..core.Exceptions import *
-from ..models.AuthSchema import UserOut, UserAuth, TokenPayload, BulkSignup
-from ..utils.DBQueries import DBQueries
+from API.core.ConfigEnv import settings
+from API.core.Exceptions import *
+from API.models import UserOut, UserAuth, TokenPayload, BulkSignup
+from API.services.db.utils import DBQueries
+
+from .utils import Auth
 
 
 def signup(response_result: FrontendResponseModel, data: Union[UserAuth,BulkSignup]):

API/services/auth/__init__.py

@@ -0,0 +1 @@
+from .AuthServices import *

API/utils/Auth.py → API/services/auth/utils/Auth.py

@@ -8,9 +8,9 @@
 from passlib.context import CryptContext
 from pydantic import ValidationError
 
-from ..core.ConfigEnv import settings
-from..core.Exceptions import *
-from ..models.AuthSchema import TokenPayload, TokenSchema
+from API.core.ConfigEnv import settings
+from API.core.Exceptions import *
+from API.models import TokenPayload, TokenSchema
 
 from fastapi.exceptions import HTTPException
 

API/utils/JWTBearer.py → API/services/auth/utils/JWTBearer.py

@@ -8,16 +8,16 @@
 This utility class validates generated JWTs and grants scoped access to users
 according to their roles.
 """
-from ..core.ConfigEnv import settings
-from ..models.AuthSchema import TokenPayload
-
 from datetime import datetime
 
 from fastapi import Request, HTTPException
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from pydantic import ValidationError
 from jose import jwt
 
+from API.core.ConfigEnv import settings
+from API.models import TokenPayload
+
 
 class JWTBearer(HTTPBearer):
     """Custom bearer to validate access tokens.

API/services/auth/utils/__init__.py

@@ -0,0 +1,2 @@
+from .Auth import *
+from .JWTBearer import *