Implementing FastAPI Services – Abstraction and Separation of Concerns

This article introduces an approach to structure FastAPI applications with multiple services in mind. The proposed structure decomposes the individual services into packages and modules, following principles of abstraction and separation of concerns.

The code discussed below can also be studied in its entirety in a dedicated companion GitHub repository: https://github.com/visini/abstracting-fastapi-services

FastAPI – Building High-performing Python APIs

FastAPI is a fast, highly intuitive and well-documented API framework based on Starlette. Despite being relatively new, it's gaining strong adoption by the developer community – it is even already being adopted in production by corporates like Netflix and Microsoft.

Following the UNIX philosophy of "doing one thing, and doing it well", separating parts of the application according to their task improves code readability and maintainability; ultimately reducing complexity. The main benefits of structuring applications in this way:

Separation of concerns – Decomposing the application into modules performing a single job. This allows accepting requests (top down: controller > service > data access) and returning responses (bottom up: data access > service > controller) with a clear separation of what particular functionality should be implemented in which particular module, reducing cognitive load for development.

Abstraction — Components of the application are designed in a reusable way. For instance, ServiceResult is implemented as a generic outcome of a service operation (which may be successful and return a response, or unsuccessful and raise an exception) able to be used by all services of the app, keeping code DRY.

The granular nature of namespacing allow to distinguish parts of the application, e.g., routes versus business logic belonging to a particular service – grouping similar tasks together, while keeping distinct parts separated. Four principal packages are needed. For each service, one module is added to these four packages. For instance, a service called "Foo" requires the following modules (discussed in detail below):

./routers/  foo.py # Router instance and routes
./services/ foo.py # Business logic (including CRUD helpers)
./schemas/  foo.py # Data "schemas" (e.g., Pydantic models)
./models/   foo.py # Database models (e.g., SQLAlchemy models)

The four principal packages are complemented by two generic packages which contain application-specific (and not service-specific) functionality, such as configuration or utility functions.

Controller Layer – Routes

Within main, the application is instantiated and all routers are included. Additionally, middlewares and/or exception handlers are implemented. The example application discussed below is based on a service called Foo, which requires a number of routes. To handle custom exceptions occurring at the service layer, as instances of class AppExceptionCase, a respective exception handler is added to the application.

# ...
from fastapi import FastAPI
from routers import foo

app = FastAPI()

@app.exception_handler(AppExceptionCase)
async def custom_app_exception_handler(request, e):
    return await app_exception_handler(request, e)

app.include_router(foo.router)

Routers and their routes are defined in modules within the routers package. Each route instantiates the respective service and passes on the database session from the request dependency. Handled by handle_result(), the service result (either the requested data as the result of a successful operation, or an exception) is returned. In case of an exception, instead of (and before) returning any response, the app exception handler in main picks up handling the exception and returns a response.

router = APIRouter(prefix="/foo")

@router.post("/item/", response_model=FooItem)
async def create_item(item: FooItemCreate, db: get_db = Depends()):
    result = FooService(db).create_item(item)
    return handle_result(result)

@router.get("/item/{item_id}", response_model=FooItem)
async def get_item(item_id: int, db: get_db = Depends()):
    result = FooService(db).get_item(item_id)
    return handle_result(result)

Service Result Class – Success & Exceptions

The ServiceResult class defines a generic outcome of a service operation. In case the operation is successful, the outcome (or "value") is returned contained within the value attribute of the instance. In case of a custom app exception, the service result instance contains information about the raised exception (e.g., which status code should be returned to the client).

class ServiceResult:
    def __init__(self, arg):
        if isinstance(arg, AppExceptionCase):
            self.success = False
            self.exception_case = arg.exception_case
            self.status_code = arg.status_code
        else:
            self.success = True
            self.exception_case = None
            self.status_code = None
        self.value = arg

    def __str__(self):
        if self.success:
            return "[Success]"
        return f'[Exception] "{self.exception_case}"'

    def __repr__(self):
        if self.success:
            return "<ServiceResult Success>"
        return f"<ServiceResult AppException {self.exception_case}>"

    def __enter__(self):
        return self.value

    def __exit__(self, *kwargs):
        pass


def handle_result(result: ServiceResult) -> :
    if not result.success:
        with result as exception:
            raise exception
    with result as result:
        return result

Service Layer – Business Logic

Services are defined in the services package. Each service is a subclass of AppService. The database session is passed down from the request dependency via an "interface-like" mixin utility class (other mixing classes may be added via multiple inheritance in order to extend available attributes).

class DBSessionMixin:
    def __init__(self, db: Session):
        self.db = db

class AppService(DBSessionMixin):
    pass

class AppCRUD(DBSessionMixin):
    pass

Routes belonging to service Foo are connected to methods of FooService, which encapsulates all business logic of the service. The return value are of type ServiceResult: Containing a value attribute with returnable data or, in case of an exception, an AppException. In both cases, the respective result is returned back "upwards" to the controller layer.

class FooService(AppService):
    def create_item(self, item: FooItemCreate) -> ServiceResult:
        foo_item = FooCRUD(self.db).create_item(item)
        if not foo_item:
            return ServiceResult(AppException.FooCreateItem())
        return ServiceResult(foo_item)

    def get_item(self, item_id: int) -> ServiceResult:
        foo_item = FooCRUD(self.db).get_item(item_id)
        if not foo_item:
            return ServiceResult(AppException.FooGetItem({"item_id": item_id}))
        if not foo_item.public:
            return ServiceResult(AppException.FooItemRequiresAuth())
        return ServiceResult(foo_item)

Data Access Layer – Database Operations

CRUD helper methods perform operations on the database and are subclassing AppCRUD. The database session is passed down from the AppService instance. These methods are atomic and only concerned with operating on the database. They do not contain any business logic.

class FooCRUD(AppCRUD):
    def create_item(self, item: FooItemCreate) -> FooItem:
        foo_item = FooItem(description=item.description, public=item.public)
        self.db.add(foo_item)
        self.db.commit()
        self.db.refresh(foo_item)
        return foo_item

    def get_item(self, item_id: int) -> FooItem:
        foo_item = self.db.query(FooItem).filter(FooItem.id == item_id).first()
        if foo_item:
            return foo_item
        return None

Pydantic "schemas" or models are defined in the schemas package. They contain mainly two different kinds of data models. First, those expected from clients as request data (route parameters in route method definitions). Second, those expected to be returned to clients as response data (defined in response_model parameter of route definitions).

Additionally, it's possible to model any kind of data being passed inbetween the layers of the app (controller, service and data-access).

class FooItemBase(BaseModel):
    description: str

class FooItemCreate(FooItemBase):
    public: bool

class FooItem(FooItemBase):
    id: int

    class Config:
        orm_mode = True

SQLAlchemy models are defined in the models package. They define how data is stored within the relational database. They are referenced from AppCRUD. If needed, make sure to differentiate between FooItem models (SQLAlchemy) and FooItem schemas (Pydantic) by appropriate import namespacing.

class FooItem(Base):
    __tablename__ = "foo_items"

    id = Column(Integer, primary_key=True, index=True)
    description = Column(String)
    public = Column(Boolean, default=False)

Exception Handling

App exceptions are implemented in the utils package. First, AppExceptionCase is subclassed from base Exception and includes various attributes for defining custom app exception scenarios. The exception handler with the task of handling custom app exceptions (added to main, see above) is defined with a response containing information about the app exception.

class AppExceptionCase(Exception):
    def __init__(self, status_code: int, context: dict):
        self.exception_case = self.__class__.__name__
        self.status_code = status_code
        self.context = context

    def __str__(self):
        return (
            f"<AppException {self.exception_case} - "
            + f"status_code={self.status_code} - context={self.context}>"
        )


async def app_exception_handler(request: Request, exc: AppExceptionCase):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "app_exception": exc.exception_case,
            "context": exc.context,
        },
    )

Second, defining and documenting custom exception scenarios happens in the same module and requires subclassing from AppExceptionCase. Each app exception includes a description within the docstring and defines the status code to be returned to the client.

The class names are reported back to inform clients of the particular exception scenario - see AppExceptionCase initalization, and JSONResponse within the app exception handler.

class AppException:
    class FooCreateItem(AppExceptionCase):
        def __init__(self, context: dict = None):
            """
            Item creation failed
            """
            status_code = 500
            AppExceptionCase.__init__(self, status_code, context)

    class FooGetItem(AppExceptionCase):
        def __init__(self, context: dict = None):
            """
            Item not found
            """
            status_code = 404
            AppExceptionCase.__init__(self, status_code, context)

    class FooItemRequiresAuth(AppExceptionCase):
        def __init__(self, context: dict = None):
            """
            Item is not public and requires auth
            """
            status_code = 401
            AppExceptionCase.__init__(self, status_code, context)

It's possible to add context to an exception AppException.FooCreateItem({"id": 123}) in order to inform the client of an exception's context, if needed.

To compile a list of all exceptions (e.g., in order to localize app exception messages displayed to the client), access all attributes within the scope of AppException:

from utils.app_exceptions import AppException

print([e for e in dir(AppException) if "__" not in e])
# ['FooCreateItem', 'FooGetItem', 'FooItemRequiresAuth']

Conclusion

The proposed approach allows designing FastAPI applications which are highly structured to accomodate multiple services, allowing the implementation of any number of services in a unified pattern.

I hope you enjoyed this article! I'm curious to hear your thoughts regarding this proposed structure and look forward to learn from alternative approaches - feel free to start a discussion by opening an issue on GitHub in the companion repository to this article, or reach out to me directly.