Implementing FastAPI Services – Abstraction and Separation of Concerns
Tags: fastapi, python, abstraction, concerns
Published: 2020-12-29 13:29:36

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](https://fastapi.tiangolo.com/) is a fast, highly intuitive and well-documented API framework based on [Starlette](https://www.starlette.io/). Despite being relatively new, it's gaining [strong adoption](https://star-history.t9t.io/#tiangolo/fastapi) by the developer community – it is even already being adopted in production by corporates like [Netflix](https://netflixtechblog.com/introducing-dispatch-da4b8a2a8072) and [Microsoft](https://github.com/tiangolo/fastapi/pull/26).

Following the [UNIX philosophy](https://en.wikipedia.org/wiki/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](https://dev-to-uploads.s3.amazonaws.com/i/zktlx0pz0jggtwisu3xz.jpg)

**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](https://en.wikipedia.org/wiki/Don%27t_repeat_yourself).

![Directory Structure Overview](https://dev-to-uploads.s3.amazonaws.com/i/o1xvdrbs1holuvk2wtsv.jpg)
 
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):

```bash
./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.

```python
# ...
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.

```python
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).

```python
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).

```python
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.

```python
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.

```python
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).

```python
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.

```python
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.

```python
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.

```python
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`:

```python
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](https://github.com/visini/abstracting-fastapi-services) to this article, or reach out to me directly.