[Python] Generalizing FastAPI with Dependency Injection

I want to have a BaseController class and a sample UserController class that inherits from BaseController. In BaseController, I have add, get, get_all, update, and delete apis and in UserController I have additional apis like get_by_username api. I want to implement dependency injection to my project. DI works for my UserController and it looks like this

dependencies.py:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated
from src.application.contracts.user_service import IUserService
from src.application.services.user_service import UserService
from src.infrastructure.database import get_db_session

DBSessionDep = Annotated[AsyncSession, Depends(get_db_session)]


def get_user_service(db: DBSessionDep) -> IUserService:
return UserService(db)


user_controller.py:

from typing import Annotated
from fastapi import Depends, APIRouter
from src.api.controllers.base_controller import BaseController
from src.application.contracts.user_service import IUserService
from src.application.dependencies import get_user_service, DBSessionDep
from src.application.dtos.user.user import UserResponse


class UserController(BaseController[UserResponse, IUserService]):
def __init__(self):
super().__init__(get_user_service, APIRouter(prefix="/api/user", tags= ["user"]))

self.router.get("/get_by_username/{username}", response_model=UserResponse)(self.get_by_username)

async def get_by_username(self, username: str, user_service: Annotated[IUserService, Depends(get_user_service)]) -> UserResponse:
return await user_service.get_by_username(username)


The solution is that I should pass an argument like this user_service: Annotated[IUserService, Depends(get_user_service)] for my apis to work, so I pass it to BaseController's init function like above. But the main problem is that when I receive the get_user_service function, I cannot use it as a part of my method arguments like this

from fastapi import Depends, status, APIRouter
from typing import Generic, TypeVar, List, Annotated, Callable
from src.application.contracts.user_service import IUserService
from src.application.services.user_service import UserService

TEntity = TypeVar("TEntity")
TService = TypeVar("TService", bound="BaseService")


class BaseController(Generic[TEntity, TService]):
def __init__(self, service_dep, router: APIRouter):
self.service = service_dep
self.router = router

self.router.post("/", response_model=TEntity)(self.add)
self.router.get("/{id}", response_model=TEntity)(self.get)
self.router.get("/", response_model=List[TEntity])(self.get_all)
self.router.put("/{id}", response_model=TEntity)(self.update)
self.router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)(self.delete)

async def add(self, entity: TEntity, service: Annotated[TService, Depends(self.service)]) -> TEntity:
return await service.add(entity)

async def get(self, entity_id: int, service: Annotated[TService, Depends(self.service)]) -> TEntity:
return await service.get(entity_id)

async def get_all(self, service: Annotated[TService, Depends(self.service)]) -> List[TEntity]:
return await service.get_all()

async def update(self, entity: TEntity, service: Annotated[TService, Depends(self.service)]) -> TEntity:
return await service.update(entity)

async def delete(self, entity_id: int, service: Annotated[TService, Depends(self.service)]) -> None:
await service.delete(entity_id)


and I get error on using self as a part of the arguments. Does anybody has a solution for that?

Continue reading...