pass db session for testcase

with depend db session, it will be override from conftest

Author: mostafa6765

app/api/v1/user.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
-from fastapi import APIRouter, Request, status
+from fastapi import APIRouter, Request, status, Depends
 from fastapi.encoders import jsonable_encoder
 from fastapi.responses import JSONResponse
 from fastapi_pundra.rest.helpers import the_query
 from fastapi_pundra.rest.validation import dto
-
+from sqlalchemy.orm import Session
+from app.database.database import get_db_session
 from app.schemas.user_schema import UserCreateSchema, UserUpdateSchema
 from app.services.user_service import UserService
 
@@ -19,47 +20,53 @@
 # Registration route
 @router.post("/users/registration")
 @dto(UserCreateSchema)
-async def registration(request: Request) -> JSONResponse:
+async def registration(request: Request, db: Session = Depends(get_db_session)) -> JSONResponse:
     """Register a new user."""
     # Retrieve data from the request
     request_data = await the_query(request)
     data = UserCreateSchema(**request_data)
 
-    output = await user_service.s_registration(request, data)
+    output = await user_service.s_registration(request, db, data)
     return JSONResponse(content=output, status_code=status.HTTP_201_CREATED)
 
 
 @router.post("/users/login")
-async def login(request: Request) -> JSONResponse:
+async def login(request: Request, db: Session = Depends(get_db_session)) -> JSONResponse:
     """Login a user."""
-    data = await user_service.s_login(request)
+    data = await user_service.s_login(request, db)
     return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)
 
 
 @router.get("/users")
-async def get_users(request: Request) -> JSONResponse:
+async def get_users(request: Request, db: Session = Depends(get_db_session)) -> JSONResponse:
     """Get all users."""
-    data = await user_service.s_get_users(request)
+    data = await user_service.s_get_users(request, db)
     return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)
 
 
 @router.get("/users/{user_id}")
-async def get_user(request: Request, user_id: int | str) -> JSONResponse:
+async def get_user(
+    request: Request, user_id: int | str, db: Session = Depends(get_db_session)
+) -> JSONResponse:
     """Get a user by id."""
-    data = await user_service.s_get_user_by_id(request, user_id=user_id)
+    data = await user_service.s_get_user_by_id(request, db, user_id=user_id)
     return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)
 
 
 @router.put("/users/{user_id}/update")
 @dto(UserUpdateSchema)
-async def update_user(request: Request, user_id: int | str) -> JSONResponse:
+async def update_user(
+    request: Request, user_id: int | str, db: Session = Depends(get_db_session)
+) -> JSONResponse:
     """Update a user by id."""
-    data = await user_service.s_update_user(request, user_id=user_id)
+    data = await user_service.s_update_user(request, db, user_id=user_id)
     return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)
 
 
 @router.delete("/users/{user_id}/delete")
-async def delete_user(request: Request, user_id: int | str) -> JSONResponse:
+async def delete_user(
+    request: Request, user_id: int | str, db: Session = Depends(get_db_session)
+) -> JSONResponse:
     """Delete a user by id."""
-    data = await user_service.s_delete_user(request, user_id=user_id)
+    data = await user_service.s_delete_user(request, db, user_id=user_id)
     return JSONResponse(content=jsonable_encoder(data), status_code=status.HTTP_200_OK)

app/database/__init__.py

@@ -1,3 +1,3 @@
 """Database module."""
 
-from app.database.database import Base, engine, get_db
+from app.database.database import Base, engine, get_db_session

app/database/database.py

@@ -1,5 +1,5 @@
 import os
-
+from collections.abc import Generator
 from dotenv import load_dotenv
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, declarative_base, sessionmaker
@@ -29,6 +29,10 @@
 
 
 # Dependency function to get a database session
-def get_db() -> Session:
+def get_db_session() -> Generator[Session, None, None]:
     """Get a database session."""
-    return SessionLocal()
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/services/user_service.py

@@ -10,7 +10,6 @@
 from fastapi_pundra.rest.paginate import paginate
 from sqlalchemy.orm import Session
 
-from app.database import get_db
 from app.models.users import User
 from app.schemas.user_schema import UserCreateSchema
 from app.serializers.user_serializer import UserLoginSerializer, UserSerializer
@@ -19,13 +18,9 @@
 class UserService:
     """User service."""
 
-    def __init__(self, db: Session | None = None) -> None:
-        """Initialize the user service."""
-        self.db = db or get_db()
-
-    async def s_registration(self, request: Request, data: UserCreateSchema) -> dict:
+    async def s_registration(self, request: Request, db: Session, data: UserCreateSchema) -> dict:
         """Register a new user."""
-        db_user = self.db.query(User).filter(User.email == data.email).first()
+        db_user = db.query(User).filter(User.email == data.email).first()
 
         if db_user:
             raise BaseAPIException(message="Email already registered", status_code=400)
@@ -36,9 +31,9 @@ async def s_registration(self, request: Request, data: UserCreateSchema) -> dict
         new_user.name = data.name
         new_user.status = "active"
 
-        self.db.add(new_user)
-        self.db.commit()
-        self.db.refresh(new_user)
+        db.add(new_user)
+        db.commit()
+        db.refresh(new_user)
 
         user_data = UserSerializer(**new_user.as_dict())
 
@@ -59,9 +54,9 @@ async def s_registration(self, request: Request, data: UserCreateSchema) -> dict
 
         return {"message": "Registration successful", "user": user_data.model_dump()}
 
-    async def s_get_users(self, request: Request) -> dict:
+    async def s_get_users(self, request: Request, db: Session) -> dict:
         """Get users."""
-        query = self.db.query(User)
+        query = db.query(User)
 
         # TODO: add logic here if you want to filter users
 
@@ -80,22 +75,22 @@ def additional_data(data: list) -> dict:
             additional_data=additional_data,
         )
 
-    async def s_get_user_by_id(self, request: Request, user_id: str) -> User:
+    async def s_get_user_by_id(self, request: Request, db: Session, user_id: str) -> User:
         """Get user by id."""
-        user = self.db.query(User).filter(User.id == user_id).first()
+        user = db.query(User).filter(User.id == user_id).first()
         if user is None:
             raise ItemNotFoundException(message="User not found")
         return user
 
-    async def s_login(self, request: Request) -> dict:
+    async def s_login(self, request: Request, db: Session) -> dict:
         """Login a user."""
         # Get data from request
         the_data = await the_query(request)
         email = the_data.get("email")
         password = the_data.get("password")
 
         # Find user by email
-        user = self.db.query(User).filter(User.email == email).first()
+        user = db.query(User).filter(User.email == email).first()
         if not user:
             raise UnauthorizedException(message="Invalid credentials")
 
@@ -125,10 +120,10 @@ async def s_login(self, request: Request) -> dict:
             "refresh_token": refresh_token,
         }
 
-    async def s_update_user(self, request: Request, user_id: str) -> dict:
+    async def s_update_user(self, request: Request, db: Session, user_id: str) -> dict:
         """Update a user."""
         the_data = await the_query(request)
-        user = self.db.query(User).filter(User.id == user_id).first()
+        user = db.query(User).filter(User.id == user_id).first()
 
         if not user:
             raise ItemNotFoundException(message="User not found")
@@ -139,17 +134,17 @@ async def s_update_user(self, request: Request, user_id: str) -> dict:
             user.email = the_data.get("email")
         if the_data.get("password"):
             user.password = generate_password_hash(the_data.get("password"))
-        self.db.commit()
-        self.db.refresh(user)
+        db.commit()
+        db.refresh(user)
 
         user_data = UserSerializer(**user.as_dict())
         return {"message": "User updated successfully", "user": user_data.model_dump()}
 
-    async def s_delete_user(self, request: Request, user_id: str) -> dict:
+    async def s_delete_user(self, request: Request, db: Session, user_id: str) -> dict:
         """Delete a user."""
-        user = self.db.query(User).filter(User.id == user_id).first()
+        user = db.query(User).filter(User.id == user_id).first()
         if not user:
             raise ItemNotFoundException(message="User not found")
-        self.db.delete(user)
-        self.db.commit()
+        db.delete(user)
+        db.commit()
         return {"message": "User deleted successfully"}

ruff.toml

@@ -52,7 +52,9 @@ ignore = [
     "FIX002",  # Line contains TODO, consider resolving the issue
     "COM812",  # Missing trailing comma in Python 3.6+
     "N805",    # Instance method first argument name should be 'self'
-    "ERA001",
+    "ERA001",  # Found commented-out code
+    "FAST002", # FastAPI dependency injection error annotation
+    "B008",    # Do not perform function call in argument defaults
 ]
 
 # Allow unused variables when underscore-prefixed