added validation dto decorator

Author: mostafa6765

fastapi_project/api/v1/user.py

@@ -4,13 +4,14 @@
 from fastapi_project.utils.base import the_query
 from fastapi_project.services.user_service import UserService
 from fastapi_project.schemas.user_schema import UserCreateSchema
-
+from fastapi_project.utils.validation import dto
 router = APIRouter()
 
 user_service = UserService()
 
 @router.post("/users/create")
-async def create_order(request: Request, the_data: UserCreateSchema):
+@dto(UserCreateSchema)
+async def create_order(request: Request):
     # Retrieve data from the request
     request_data = await the_query(request)
     data = UserCreateSchema(**request_data)

fastapi_project/main.py

@@ -3,6 +3,7 @@
 from dotenv import load_dotenv
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi_project.api import health, root_index
+from fastapi_project.utils.validation import setup_validation_exception_handler
 
 # Load .env file
 load_dotenv()
@@ -13,6 +14,7 @@
 
 def create_application():
     application = FastAPI()
+    setup_validation_exception_handler(application)
 
     # Include the root index and health router
     application.include_router(root_index.router)

fastapi_project/schemas/user_schema.py

@@ -1,15 +1,15 @@
-from pydantic import BaseModel, EmailStr
+from pydantic import BaseModel, EmailStr, Field
 from typing import Optional
 
 # Pydantic model for creating a new user
 class UserCreateSchema(BaseModel):
     name: str
     email: EmailStr
-    password: str
+    password: str = Field(..., min_length=8)
 
 
 # Pydantic model for updating user data
 class UserUpdateSchema(BaseModel):
     name: Optional[str] = None
     email: Optional[EmailStr] = None
-    password: Optional[str] = None
+    password: Optional[str] = Field(None, min_length=8)

fastapi_project/utils/base.py

@@ -1,8 +1,10 @@
-from fastapi import Request
+from fastapi import Request, HTTPException
 from typing import Any, Dict, Optional, Union
 from pydantic import BaseModel, ValidationError
 from sqlalchemy import desc
 from urllib.parse import urlparse
+from fastapi.responses import JSONResponse
+from fastapi import FastAPI
 
 async def the_query(request: Request, name = None) -> Dict[str, str]:
     data = {}
@@ -21,17 +23,22 @@ async def the_query(request: Request, name = None) -> Dict[str, str]:
 
 
 async def validate_data(data: Dict[str, Any], model: BaseModel) -> Dict[str, Union[str, Dict[str, Any]]]:
-    output = {'status': 'valid'}
-    
     try:
         instance = model(**data)
-        output['data'] = instance.dict()
+        return {'success': True, 'data': instance.dict()}
     except ValidationError as e:
-        # If validation fails, return status as invalid and the validation errors
-        output['status'] = 'invalid'
-        output['errors'] = e.errors()
-        
-    return output
+        errors = {}
+        for error in e.errors():
+            field = error['loc'][0]
+            message = field + " " + error['msg'] 
+            if field not in errors:
+                errors[field] = []
+            errors[field].append(message)
+            
+        return {
+            'success': False,
+            'errors': errors["details"]
+        }
 
 
 def the_sorting(request, query):
@@ -96,4 +103,8 @@ def paginate(request: Request, query, serilizer, the_page: int = 1, the_per_page
         'from': offset + 1 if data else None,
         'to': offset + len(data) if data else None,
         wrap: data
-    }
+    }
+
+
+# Update the __all__ list
+__all__ = []

fastapi_project/utils/validation.py

@@ -0,0 +1,51 @@
+from fastapi import Request, HTTPException
+from fastapi.responses import JSONResponse
+from fastapi import FastAPI
+from pydantic import BaseModel, ValidationError
+from functools import wraps
+from fastapi_project.utils.base import the_query
+
+# Add the ValidationException class
+class ValidationException(Exception):
+    def __init__(self, errors: dict):
+        self.errors = errors
+
+# Add the setup_validation_exception_handler function
+def setup_validation_exception_handler(app: FastAPI):
+    @app.exception_handler(ValidationException)
+    async def validation_exception_handler(request: Request, exc: ValidationException):
+        return JSONResponse(
+            status_code=422,
+            content=exc.errors
+        )
+
+# Add the dto decorator
+def dto(schema: BaseModel):
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(request: Request, *args, **kwargs):
+            try:
+                request_data = await the_query(request)
+                validated_data = schema(**request_data)
+                return await func(validated_data, *args, **kwargs)
+            except ValidationError as e:
+                errors = {}
+                for error in e.errors():
+                    field = error['loc'][0]
+                    message = field + " " + error['msg']
+                    if field not in errors:
+                        errors[field] = []
+                    errors[field].append(message)
+                    
+                raise ValidationException({
+                    'success': False,
+                    'errors': errors
+                })
+            except ValueError:
+                raise HTTPException(status_code=400, detail="Invalid JSON")
+            
+        return wrapper
+    return decorator
+
+# Update the __all__ list
+__all__ = ['dto', 'ValidationException', 'setup_validation_exception_handler']