feat: upgrade pydantic versin with python version and made changes for pydantic syntax

Author: Jhanvi-vc

README.md

@@ -58,7 +58,7 @@ To get a local copy up and running follow these simple steps.
 Requirement of Project
 * Install Python 
   ```sh
-  Python-Version : 3.10.13
+  Python-Version : 3.11.0
   ```
 * Create python virtual environment
   ```sh
@@ -73,7 +73,7 @@ Requirement of Project
 
 1. Clone the repo
    ```sh
-   git clone https://github.com/viitoradmin/python-fastapi-boilerplate
+   git clone https://github.com/viitoradmin/python-fastapi-boilerplate/tree/feature/fastapi
    ```
 2. Upgrade pip version
     ```sh
@@ -92,7 +92,7 @@ Requirement of Project
 
 1. Create python virtual environment
    ```
-   conda create --name venv python=3.10.12
+   conda create --name venv python=3.11
    ```
 
 2. Activate the python virtual environment
@@ -160,8 +160,13 @@ Run the server
    python asgi.py
    ```
 Browse Swagger API Doc at: http://localhost:8000/docs
+
 Browse  Redoc at: http://localhost:8000/redoc
 
+Browse Swagger API Doc for version v1 at: http://localhost:8000/v1/docs
+
+Browse Swagger API Doc for version v2 at: http://localhost:8000/v2/docs
+
 ## Release History
 
 * 0.1

apps/__init__.py

@@ -3,7 +3,7 @@
 from config import database
 from apps.constant import constant
 from fastapi_versioning import VersionedFastAPI
-from apps.api.auth.view import defaultrouter, router
+from apps.api.auth.view import router_v1, router_v2
 from apps.api.auth.models import Base as authbase
 
 # Bind with the database, whenever new models find it's create it.
@@ -15,13 +15,14 @@
 # define router for different version
 # router for version 1
 app.include_router(
-    defaultrouter, 
+    router_v1, 
     prefix=constant.API_V1, tags=["/v1"]
     )
 # router for version 2
 app.include_router(
-    router, prefix=constant.API_V2, tags=["/v2"]
-    ) 
+    router_v2, 
+    prefix=constant.API_V2, tags=["/v2"]
+    )
 
 # Define version to specify version related API's.
-app = VersionedFastAPI(app, version_format="{major}", prefix_format="/v{major}", enable_latest=True)
+app = VersionedFastAPI(app, version_format="{major}", prefix_format="/v{major}")

apps/api/auth/method.py

@@ -0,0 +1,20 @@
+from sqlalchemy.orm import Session
+from apps.constant import constant
+
+class UserAuthMethod():
+    """This class defines methods to authenticate users."""
+    
+    def __init__(self, model) -> constant.STATUS_NULL:
+        self.model = model
+        
+    def find_by_email(self, db: Session, email: str):
+        """This funtion will return the email object"""
+        return db.query(self.model).filter(
+            self.model.email == email
+            ).first()
+    
+    def find_by_username(self, db: Session, username: str):
+        """This function will return the username object"""
+        return db.query(self.model).filter(
+            self.model.username == username
+            ).first()

apps/api/auth/schema.py

@@ -1,24 +1,43 @@
 """This module is for swager and request parameter schema"""
-from pydantic import BaseModel, Extra
+from pydantic import BaseModel, Extra, validator, Field
+from apps.api.core import validation
 
 
 class UserAuth(BaseModel):
-    first_name: str
-    last_name: str
-    email: str
-    password: str
-    username: str
+    first_name: str 
+    last_name: str 
+    email: str 
+    password: str 
+    username: str 
 
     class Config:
-        extra = Extra.forbid
-        orm_mode = True
-        extra = Extra.allow
-        schema_extra = {
+        from_attributes = True
+        json_schema_extra = {
             "example": {
                     "first_name": "John",
                     "last_name": "Smith",
                     "email": "jhohnsmith@example.com",
                     "password": "Abc@123",
                     "username": "Jhon123"
                 }
-        }
+        }
+    
+    @validator('first_name', pre=True)
+    def first_name_must_be_required(cls, v):
+        return validation.ValidationMethods().not_null_validator(v, 'first_name')
+    
+    @validator('last_name', allow_reuse=True)
+    def last_name_must_be_required(cls, v):
+        return validation.ValidationMethods().not_null_validator(v, 'last_name')
+    
+    @validator('email', allow_reuse=True)
+    def email_must_be_required(cls, v):
+        return validation.ValidationMethods().not_null_validator(v, 'email')
+    
+    @validator('username', allow_reuse=True)
+    def username_must_be_required(cls, v):
+        return validation.ValidationMethods().not_null_validator(v, 'username')
+    
+    @validator('email', allow_reuse=True)
+    def email_field_validator(cls, v):
+        return validation.ValidationMethods().email_validator(v)

apps/api/auth/service.py

@@ -1,15 +1,102 @@
-from apps.utils.standard_response import StandardResponse
+import uuid
 from fastapi import status
 from apps.constant import constant
+from sqlalchemy.orm import Session
+from apps.api.auth.models import Users
+from apps.api.core import db_methods
+from fastapi.encoders import jsonable_encoder
+from apps.api.auth.method import UserAuthMethod
+from apps.api.core.validation import ValidationMethods
+from apps.utils.message import ErrorMessage, InfoMessage
+from apps.utils.standard_response import StandardResponse
+
 
-def get_str_name(name: str):
-    if name is not None:
+class UserAuthService:
+    """This class represents the user creation service"""
+    def create_user_service(self, db: Session, body: dict):
+        """This function is used to create user
+        Args:
+            db (Session): database connection
+            body (dict): dictionary to user information data
+
+        Returns:
+            response (dict): user object representing the user
+        """
+        username = body['username']
+        email = body['email'] or None
+        password = body['password']
+        
+        # check Email exists in body or not
+        if "email" not in body or not email:
+            return StandardResponse(
+                False, status.HTTP_400_BAD_REQUEST, None, ErrorMessage.emailRequired
+                )
+            
+        # check Email exists in Db or not
+        if (user_object := UserAuthMethod(Users).find_by_email(db, email)):
+            return StandardResponse(
+                False, status.HTTP_400_BAD_REQUEST, 
+                constant.STATUS_NULL, ErrorMessage.emailAlreadyExist
+                ).make
+        
+        # For password validation
+        if not ValidationMethods().password_validator(password):
+            return StandardResponse(
+                False, status.HTTP_400_BAD_REQUEST,
+                constant.STATUS_NULL, ErrorMessage.invalidPasswordFormat
+            ).make
+            
+        user_object = Users(
+            uuid = uuid.uuid4(),
+            first_name=body['first_name'],
+            last_name=body['last_name'],
+            username=username,
+            email=email,
+            password=password
+        )
+        
+        # Store user object in database
+        if not(user_save := db_methods.BaseMethods(Users).save(user_object, db)):
+            return StandardResponse(
+                False, status.HTTP_400_BAD_REQUEST, 
+                constant.STATUS_NULL, ErrorMessage.userNotSaved
+            )
+        
+        # check email exists or not
+        if user_object := UserAuthMethod(Users).find_by_email(db, email):
+            data = {
+                "username": user_object.username,
+                "first_name": user_object.first_name,
+                "last_name": user_object.last_name,
+                "email": user_object.email,
+                "password": user_object.password,
+            }
+        
+        else: 
+            return StandardResponse(
+                False, status.HTTP_400_BAD_REQUEST, 
+                constant.STATUS_NULL, ErrorMessage.userInvalid
+            ).make
+            
         return StandardResponse(
-            True, status.HTTP_200_OK, {"success": "Welcome"}, 
-            constant.STATUS_SUCCESS
+            True, status.HTTP_200_OK, data, InfoMessage.userCreated
+        ).make
+        
+    def get_user_service(self, db):
+        """This function returns the user service list."""
+        if not (user_object := db_methods.BaseMethods(Users).find_by_uuid(db)):
+            return StandardResponse(
+                False,
+                status.HTTP_400_BAD_REQUEST,
+                None,
+                ErrorMessage.userNotFound
             )
-    else:
+        # convert the object data into json
+        user_data = jsonable_encoder(user_object)
+        
         return StandardResponse(
-            True, status.HTTP_400_BAD_REQUEST, {"success": "Error"}, 
-            constant.STATUS_ERROR
-            )
+            True, 
+            status.HTTP_200_OK,
+            user_data,
+            InfoMessage.retriveInfoSuccessfully
+        )

apps/api/auth/view.py

@@ -2,59 +2,61 @@
 from config import database
 from fastapi import Depends
 from sqlalchemy.orm import Session
-from fastapi_utils.cbv import cbv
 from apps.api.auth import schema
 from fastapi import APIRouter
 from apps.constant import constant
 from fastapi_versioning import version
-from fastapi_utils.inferring_router import InferringRouter
+from apps.api.auth.service import UserAuthService
 from apps.utils.standard_response import StandardResponse
 
 ## Load API's
-defaultrouter = APIRouter()
 router = APIRouter()
+router_v1 = APIRouter()
 getdb = database.get_db
 
 ## Define verison 1 API's here
-@cbv(defaultrouter)
 class UserCrudApi():
     """This class is for user's CRUD operation with version 1 API's"""
 
-    @defaultrouter.get('/list/user')
+    @router_v1.get('/users')
     @version(1)
-    async def list_user(self):
+    async def list_user(db: Session = Depends(getdb)):
         """This API is for list user.
         Args: None
         Returns: 
-            response: will return list."""
+            response: will return users list."""
         try:
-            data = {"List:" : "Hello there, welcome to fastapi bolierplate"}
-            return StandardResponse(True, status.HTTP_200_OK, data, constant.STATUS_SUCCESS)
+            response = UserAuthService().get_user_service(db)
+            return response
         except Exception as e:
             return StandardResponse(False, status.HTTP_400_BAD_REQUEST, None, constant.ERROR_MSG)
 
-    @defaultrouter.post('/create/user')
+    @router_v1.post('/create/user')
     @version(1)
-    async def create_user(self, body: schema.UserAuth,
+    async def create_user(body: schema.UserAuth,
                           db: Session = Depends(getdb)):
         """This API is for create user.
         Args: 
             body(dict) : user's data
         Returns:
             response: will return the user's data"""
         try:
-            data = body.dict()
-            return StandardResponse(True, status.HTTP_200_OK, data, constant.STATUS_SUCCESS)
+            # as per pydantic version 2.
+            body = body.model_dump()
+            response = UserAuthService().create_user_service(db, body)
+            return response
         except Exception as e:
             return StandardResponse(False, status.HTTP_400_BAD_REQUEST, None, constant.ERROR_MSG)
 
 
 ## Define version 2 API's here
-@cbv(router)
+router_v2 = APIRouter()
+
 class UserVersionApi():
-    @router.get("/list")
+    """This class provides version 2 API's for users"""
+    @router_v2.get("/list")
     @version(2)
-    async def get_list(self):
+    async def get_list():
         """ This API will list version 2 Api's
         Args: None
         Returns:

apps/api/core/db_methods.py

@@ -0,0 +1,36 @@
+from fastapi import Depends
+from config import database
+from apps.constant import constant
+from sqlalchemy.orm import Session
+
+getdb = database.get_db
+
+class BaseMethods():
+    """This class provides basic DB methods"""
+    
+    def __init__(self, model):
+        self.model = model
+    
+    def save(self, validate_data, db: Session = Depends(getdb)):
+        """this function saves the object to the database for the given model
+        Args:
+            validate_data (dict): validate data
+            db (Session): database session.
+        Returns:
+            returns the created object
+        """
+        try:
+            db.add(validate_data)
+            db.commit()
+            db.refresh(validate_data)
+            return constant.STATUS_TRUE
+        except Exception as err:
+            print(err)
+            db.rollback()
+            db.close()
+            return constant.STATUS_FALSE
+    
+    def find_by_uuid(self, db: Session = Depends(getdb)):
+        """This function is used to find users."""
+        return db.query(self.model).filter(self.model.deleted_at == None).all()
+