Building Scalable and Maintainable FastAPI Applications: Part-2 Dependancy Injection

In my previous blog post, we explored a well-organized directory structure for FastAPI projects. Now, let’s delve into the power of FastAPI’s dependency injection system to create clean and maintainable applications.

Throughout this series, we’ll construct a simple logger project that saves input messages to a database. However, this article will solely concentrate on dependency injection. We’ll cover repository implementation and business logic in subsequent installments.

A Cornerstone for Clean Code

Dependency injection (DI) is a software design pattern that promotes loose coupling between components. Instead of objects creating their own dependencies, they receive them from an external source. This approach offers several advantages:

• Improved Separation of Concerns — DI allows you to focus on the core functionality of your code (the business logic) without getting bogged down in instantiating and managing dependencies like database connections or authentication services.
• Enhanced Testability — With DI, it’s much easier to create mock or fake dependencies during testing, isolating the component under test and ensuring reliable tests.
• Increased Maintainability — By centralizing dependency creation, DI promotes code modularity and simplifies maintenance. Changes needed to a dependency can be made in one place, eliminating the need to modify every function that uses it.

Continuing our exploration of FastAPI’s dependency injection system, let’s now define the APIs for our logger application:

• Log Data (POST /logs): This endpoint allows users to submit log messages. It should accept data in a suitable format (e.g., JSON) and persist it to the database.
• Get Log by ID (GET /logs/{log_id}): This endpoint retrieves a specific log entry based on its unique identifier. The response should return the log details in a user-friendly format.
• Get All Logs with Date Filter (GET /logs?from={from_date}&to={to_date}): This endpoint facilitates fetching a subset of logs within a date range. Users can provide optional query parameters from_date and to_date to control the filtering. The response should return a list of filtered log entries.

Directory Structure (Recap)

Directory structure

As you can see, just like my previous blog, I organized my code into different directories. Here’s a breakdown of the key components:

• service Package: This directory encapsulates service implementations and any meta classes used for handling business logic or validation operations.
• repository Package: This directory houses classes and implementations related to the data access layer (DAL). These components handle interacting with the database for data persistence and retrieval.
• routes Package: This directory contains route definitions for our API endpoints. Essentially, it acts as the roadmap for handling incoming requests.
• controllers Package: This directory holds controller classes. Controllers are responsible for routing logic and interacting with services (for business logic) and repositories (for data access) to fulfill API requests.

This structure promotes separation of concerns, making the codebase easier to navigate, understand, and modify in the future.

FastAPI’s Dependency Injection in Action

FastAPI offers a powerful dependency injection mechanism that simplifies building applications. Let’s see how it works in our logger example, specifically for the API endpoint that retrieves logs filtered by date.

from datetime import datetime
from typing import Annotated
from fastapi import Depends, FastAPI, Query

app = FastAPI()
_db = {}  # this is a fake implementation of database class

class FilterParams:
   def __init__(self, _date: Annotated[datetime, Query(title="Date time filter")]):
       self.date = _date

class QueryParams:
   def __init__(self,
                _skip: Annotated[int, Query(title="skip record limit")],
                _limit: Annotated[int, Query(title="Limit of records to take from database")], ):
       self.skip = _skip
       self.limit = _limit

@app.get("/logs/all")
async def get_all(_filters: Annotated[FilterParams, Depends(FilterParams)],
                 _query: Annotated[QueryParams, Depends(QueryParams)]):
   response = {}
   items = _db.get_logs(limit=_query.limit, skip=_query.skip, filter_date=_filters.date)
   response.update({"data": items})
   return response

Our get_all endpoint utilizes two dependencies:

• FilterParams: This dependency captures the date filter criteria provided by the user as a query parameter (_date).
• QueryParams: This dependency handles pagination parameters, allowing users to control the number of records retrieved _skip for offset and _limit for maximum number of retrieving records.

We use the Depends function from FastAPI to inject both dependencies into the get_all function. This ensures FastAPI handles creating and passing the necessary objects based on the defined parameter types.

The FastAPI documentation provides extensive resources on dependency injection and various use cases.

Fast API dependency injection Implementation for layered architecture

We can now utilize dependency injection to provide services and repositories to our classes.

The first step involves establishing a well-defined class structure to facilitate dependency injection. We’ll begin by creating an abstract class for the logger repository, followed by a concrete implementation of that repository.

LogRepositoryMeta class

from abc import abstractmethod
from typing import List

from model.log.log_model import LogModel
from model.log.log_view_model import LogViewModel

class LogRepositoryMeta:
   @abstractmethod
   def add(self, model: LogModel) -> LogViewModel:
       pass

   @abstractmethod
   def get_by_id(self, _id: str) -> LogViewModel:
       pass

   @abstractmethod
   def edit(self, model: LogModel, _id: str) -> LogViewModel:
       pass

   @abstractmethod
   def get_all(self, _id: str) -> List[LogViewModel]:
       pass

LogRepository class implementation

from typing import List

from database.repository.meta.log_repository_meta import LogRepositoryMeta
from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel
from model.log.log_view_model import LogViewModel


class LogRepository(LogRepositoryMeta):

    def add(self, model: LogModel) -> LogViewModel:
        # map log model to data class
        # log to storage implementation
        return LogViewModel('Log created')

    def get_by_id(self, _id: str) -> LogViewModel:
        # Getting log from storage implementation

        return LogViewModel('Log get by id')

    def edit(self, model: LogModel, _id: str) -> LogViewModel:
        # map log model to dataclass
        # Edit log implementation
        return LogViewModel('Log has edited')

    def get_all(self,
                _filters: LogFilterParamModel,
                _query: QueryParamModel
                ) -> List[LogViewModel]:
        # Implementation get all log record from storage
        return [LogViewModel('Log record 1'),
                LogViewModel('Log record 2'),
                LogViewModel('Log record 3')]

Let’s create our log service meta class and implement log service class

LogServiceMeta Class

from abc import abstractmethod
from typing import List

from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel
from model.log.log_view_model import LogViewModel


class LogServiceMeta:
    @abstractmethod
    def add(self, model: LogModel) -> LogViewModel:
        pass

    @abstractmethod
    def get_by_id(self, _id: str) -> LogViewModel:
        pass

    @abstractmethod
    def edit(self, model: LogModel, _id: str) -> LogViewModel:
        pass

    @abstractmethod
    def get_all(self, _filters: LogFilterParamModel,
                _query: QueryParamModel) -> List[LogViewModel]:
        pass

Our service menta class contain same four methods to store, update and read log records

LogService Class implementation

from typing import List

from fastapi import Depends

from database.repository.impl.log_repository import LogRepository
from database.repository.meta.log_repository_meta import LogRepositoryMeta
from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel
from model.log.log_view_model import LogViewModel
from service.meta.log_service_meta import LogServiceMeta


class LogService(LogServiceMeta):

    def add(self, model: LogModel) -> LogViewModel:
        # implementation of business logic to add log records to database
        # add log record to data storage
        pass

    def get_by_id(self, _id: str) -> LogViewModel:
        # implementation of business logic to retrieve
        # and return log record from data storage
        # retrieve log record from data storage
        pass

    def edit(self, model: LogModel, _id: str) -> LogViewModel:
        # business logic implementation to edit log record in data storage
        # edit and retrieve log record in data storage
        pass

    def get_all(self, _filters: LogFilterParamModel,
                _query: QueryParamModel) -> List[LogViewModel]:
        # business logic to fiter and retrieve all record from data storage
        # retrieve all log record from storage
        pass

To facilitate log creation, update, and retrieval operations requiring storage interactions, we’ll inject a log repository instance into the log service class using FastAPI’s dependency injection framework.

from typing import List

from fastapi import Depends

from database.repository.impl.log_repository import LogRepository
from database.repository.meta.log_repository_meta import LogRepositoryMeta
from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel
from model.log.log_view_model import LogViewModel
from service.meta.log_service_meta import LogServiceMeta


class LogService(LogServiceMeta):
    _log_repository: LogRepositoryMeta

    def __init__(self, log_repository: LogRepositoryMeta = Depends(LogRepository)):
        self._log_repository = log_repository

    def add(self, model: LogModel) -> LogViewModel:
        # implementation of business logic to add log records to database

        # add log record to data storage
        return self._log_repository.add(model)

    def get_by_id(self, _id: str) -> LogViewModel:
        # implementation of business logic to retrieve
        # and return log record from data storage

        # retrieve log record from data storage
        return self._log_repository.get_by_id(_id=_id)

    def edit(self, model: LogModel, _id: str) -> LogViewModel:
        # business logic implementation to edit log record in data storage
        # edit and retrieve log record in data storage
        return self._log_repository.edit(model=model, _id=_id)

    def get_all(self, _filters: LogFilterParamModel,
                _query: QueryParamModel) -> List[LogViewModel]:
        # business logic to fiter and retrieve all record from data storage
        # retrieve all log record from storage
        return self._log_repository.get_all(_filters=_filters, _query=_query)

We can leverage the same dependency injection approach used for the service class to implement our controllers. Before creating individual controllers, we’ll define a base controller class. This base class will encapsulate common methods that any controller might need. Specific controllers will then inherit from this base class, gaining access to shared functionality and potential dependency injection for common services.

BaseController

from fastapi import HTTPException
from starlette import status

class BaseController:
    def ise(self, e: Exception):
        # log exception into file
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                            detail="Something went wrong in our end !")

# add all common controller methods here

LogController

from fastapi import Depends

from controllers.base_controller import BaseController
from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel
from service.impl.log_service import LogService
from service.meta.log_service_meta import LogServiceMeta


class LogController(BaseController):
    _log_service: LogServiceMeta

    def __init__(self, log_service: LogServiceMeta = Depends(LogService)):
        self._log_service = log_service

    def add(self, model: LogModel):
        try:
            return self._log_service.add(model=model)
        except Exception as e:
            return self.ise(e)

    def get_by_id(self, _id: str):
        try:
            return self._log_service.get_by_id(_id=_id)
        except Exception as e:
            return self.ise(e)

    def get_all(self,
                _filters: LogFilterParamModel,
                _query: QueryParamModel):
        try:
            return self._log_service.get_all(_filters=_filters, _query=_query)
        except Exception as e:
            return self.ise(e)

Here i had used fast api dependency injection in controller classes as well. Then each route we can inject our log controller as bellow

log_routes.py

from fastapi import APIRouter, Depends

from controllers import LogController
from model.common.query_param_model import QueryParamModel
from model.log.log_filter_param_model import LogFilterParamModel
from model.log.log_model import LogModel

loger_router = APIRouter(prefix='/log', tags=['Log'])


@loger_router.get("/get/{_id}")
async def get_log(_id: str, controller: LogController = Depends(LogController)):
    return controller.get_by_id(_id=_id)


@loger_router.post("/create")
async def create_log(model: LogModel, controller: LogController = Depends(LogController)):
    return controller.add(model)


@loger_router.post("/get")
async def create_log(_filters: LogFilterParamModel = Depends(LogFilterParamModel),
                     _query: QueryParamModel = Depends(QueryParamModel),
                     controller: LogController = Depends(LogController)):
    return controller.get_all(_filters=_filters, _query=_query)

The the logger route should include to main routers

main.py

from fastapi import FastAPI

from routes import loger_router

app = FastAPI(
    openapi_tags=[],
    title="Logger Application",
    description="",
    version="0.0.1",
    contact={
        "name": "Danushka Dissanayaka",
        "email": "dsjayamal@gmail.com",
    },)

# register logger routes
app.include_router(loger_router)

⚠️ Conclusion

FastAPI’s dependency injection system goes beyond simply injecting parameters into routes. It empowers you to inject dependencies directly into class constructors, enabling a layered application structure that promotes maintainability and scalability.

While this approach achieves a degree of loose coupling, it’s not the optimal strategy. We’re still directly referencing class implementations during dependency injection.

</> The source code can be found in Here.

👉 What Next

In the next blog post, we’ll delve into Inversion of Control (IoC) using meta classes to achieve true loose coupling by injecting dependencies without explicit class references.

Until then, happy coding! 🚀