Implementing FastAPI Services – Abstraction and Separation of Concerns

Implementing FastAPI Services – Abstraction and Separation of Concerns

FastAPI application and service structure for a more maintainable codebase

Published on August 11, 2020

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.

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.

Request-Response Flow across Application Layers

Request-Response Flow across Application Layers

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.

Directory Structure Overview

Directory Structure Overview

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.

  main.py
1# ...
2from fastapi import FastAPI
3from routers import foo

4app = FastAPI()

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

8app.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.

  routers ›foo.py
1router = APIRouter(prefix="/foo")

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

6@router.get("/item/{item_id}", response_model=FooItem)
7async def get_item(item_id: int, db: get_db = Depends()):
8    result = FooService(db).get_item(item_id)
9    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).

  utils ›service_result.py
1class ServiceResult:
2    def __init__(self, arg):
3        if isinstance(arg, AppExceptionCase):
4            self.success = False
5            self.exception_case = arg.exception_case
6            self.status_code = arg.status_code
7        else:
8            self.success = True
9            self.exception_case = None
10            self.status_code = None
11        self.value = arg

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

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

20    def __enter__(self):
21        return self.value

22    def __exit__(self, *kwargs):
23        pass


24def handle_result(result: ServiceResult) -> :
25    if not result.success:
26        with result as exception:
27            raise exception
28    with result as result:
29        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 mixin classes may be added via multiple inheritance in order to extend available attributes).

  services ›main.py
1class DBSessionMixin:
2    def __init__(self, db: Session):
3        self.db = db

4class AppService(DBSessionMixin):
5    pass

6class AppCRUD(DBSessionMixin):
7    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.

  services ›foo.py
1class FooService(AppService):
2    def create_item(self, item: FooItemCreate) -> ServiceResult:
3        foo_item = FooCRUD(self.db).create_item(item)
4        if not foo_item:
5            return ServiceResult(AppException.FooCreateItem())
6        return ServiceResult(foo_item)

7    def get_item(self, item_id: int) -> ServiceResult:
8        foo_item = FooCRUD(self.db).get_item(item_id)
9        if not foo_item:
10            return ServiceResult(AppException.FooGetItem({"item_id": item_id}))
11        if not foo_item.public:
12            return ServiceResult(AppException.FooItemRequiresAuth())
13        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.

  services ›foo.py
1class FooCRUD(AppCRUD):
2    def create_item(self, item: FooItemCreate) -> FooItem:
3        foo_item = FooItem(description=item.description, public=item.public)
4        self.db.add(foo_item)
5        self.db.commit()
6        self.db.refresh(foo_item)
7        return foo_item

8    def get_item(self, item_id: int) -> FooItem:
9        foo_item = self.db.query(FooItem).filter(FooItem.id == item_id).first()
10        if foo_item:
11            return foo_item
12        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).

  schemas ›foo.py
1class FooItemBase(BaseModel):
2    description: str

3class FooItemCreate(FooItemBase):
4    public: bool

5class FooItem(FooItemBase):
6    id: int

7    class Config:
8        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.

  models ›foo.py
1class FooItem(Base):
2    __tablename__ = "foo_items"

3    id = Column(Integer, primary_key=True, index=True)
4    description = Column(String)
5    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.

  utils ›app_exceptions.py
1class AppExceptionCase(Exception):
2    def __init__(self, status_code: int, context: dict):
3        self.exception_case = self.__class__.__name__
4        self.status_code = status_code
5        self.context = context

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


11async def app_exception_handler(request: Request, exc: AppExceptionCase):
12    return JSONResponse(
13        status_code=exc.status_code,
14        content={
15            "app_exception": exc.exception_case,
16            "context": exc.context,
17        },
18    )

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.

  utils ›app_exceptions.py
1class AppException:
2    class FooCreateItem(AppExceptionCase):
3        def __init__(self, context: dict = None):
4            """
5            Item creation failed
6            """
7            status_code = 500
8            AppExceptionCase.__init__(self, status_code, context)

9    class FooGetItem(AppExceptionCase):
10        def __init__(self, context: dict = None):
11            """
12            Item not found
13            """
14            status_code = 404
15            AppExceptionCase.__init__(self, status_code, context)

16    class FooItemRequiresAuth(AppExceptionCase):
17        def __init__(self, context: dict = None):
18            """
19            Item is not public and requires auth
20            """
21            status_code = 401
22            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:

  utils ›get_exceptions_for_frontend.py
1from utils.app_exceptions import AppException

2print([e for e in dir(AppException) if "__" not in e])
3# ['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.

Let's stay in touch!

If you'd like to get notified about updates, feel free to

Enter your email address below to receive an email whenever I write a new post.

Note: You can unsubscribe by clicking on the unsubscribe link included at the bottom of every email.