Clean Architecture in Python with FastAPI, FastStream, and Dishka

One business logic layer, two entry points: HTTP and RabbitMQ. Here is the backend template I built with FastAPI, FastStream, and Dishka. This walkthrough shows how I kept the business logic independent of both interfaces and used dependency injection to connect them to PostgreSQL.

This is an English translation of my article originally published in Russian on Habr: Building a General-Purpose Python Backend Prototype.

Hi, DEV! My name is Sergey. When I wrote the original article, I was a tech lead on the PT BlackBox team. My colleagues and I were developing a product that detected vulnerabilities in applications through black-box testing. We were also building web applications ourselves. This article draws on that experience.

In this article, I look back at the backend development decisions I made at the time and share the prototype template I built. My goal was to cover the vast majority of backend development needs.

I used FastAPI, FastStream, and dishka to build the prototype. This article is particularly useful if you use Tornado, Django, Flask, or AIOHTTP and want to move to more up-to-date technologies for further development of your projects.

Jump to a section:

• FastAPI
• FastStream
• Clean Architecture (SOLID)
• Dependency Injection (dishka)
• Implementing the Prototype
• Conclusion

Using FastAPI, FastStream, and dishka lets us follow Clean Architecture principles and almost avoid leaky abstractions. The prototype also used Pydantic and SQLAlchemy. I do not focus on them because these technologies are already popular, and the community has plenty of guides on their use.

FastAPI

FastAPI is a modern, high-performance web framework for building APIs in Python, based on standard type annotations. It is built on established tools: the Starlette ASGI framework and the Pydantic data validation library. It is a minimalist framework with no unnecessary features, which lets developers focus on business logic rather than built-in utilities.

FastStream

FastStream is an event-driven framework that simplifies and unifies work with message brokers, removing the need to reinvent the wheel each time. It offers validation, automatic AsyncAPI generation, and a ChatGPT-based CLI tool that generates ready-to-use FastStream services from your description. The framework supports Kafka, RabbitMQ, NATS, and Redis Streams.

Clean Architecture (SOLID)

As mentioned above, the prototype is built on Clean Architecture principles. I followed SOLID and divided the application into layers.

I prepared this section for readers who are unfamiliar with Clean Architecture or want to refresh their knowledge. This is not a definitive guide to Clean Architecture, and some things may be deliberately simplified. Here is Robert Martin's illustration of Clean Architecture:

Source

Each circle represents a different part of the application. The closer a layer is to the center, the less it is subject to change. Therefore, inner layers must not depend on outer layers.

Frameworks and Drivers

This layer contains external components, such as a specific database or an interface to your application. In my prototype, I did not tie the business logic to this layer.

Interface Adapters

This layer contains concrete implementations that communicate with external services, such as a database. You can picture it like this:

class BooksGateway: def get_books(self) -> list[Books]: query = 'SELECT * FROM books' rows = self.psql.execute(query) return [Books(**r) for r in rows] 

Application Business Rules

This layer contains the code that implements user scenarios. It holds components that describe a specific use case and bring together the dependencies it needs. Schematically, this looks like:

def interactor(self) -> bool: """ User buy books """ books = self.get_books_from_card() full_price = sum([b.price for b in books]) self.write_off_money(full_price) self.send_books_to_user(books) 

Enterprise Business Rules

This is the application's domain layer, which contains its foundation.

SOLID

Clean Architecture is about following SOLID principles, rather than simply having layers. Layers are a consequence of SOLID and a couple of other principles.

These principles apply to building any application. They are universal and independent of the programming language. You could call them principles derived from OOP, although that would not be entirely accurate: they apply both to microservice architecture and to individual applications. Only one of the five principles is specific to OOP.

S — Single Responsibility Principle

A module should have one and only one reason to change.

O — Open–Closed Principle

It should be possible to add new components without changing existing ones.

L — Liskov Substitution Principle

Barbara Liskov's substitution principle states that a subclass must conform to the behavior of its parent class. This is the principle that applies specifically to OOP.

I — Interface Segregation Principle

Components should not depend on methods they do not use.

D — Dependency Inversion Principle

Components should depend on abstract interfaces rather than concrete implementations.

Dependency Injection (dishka)

Dependency injection is a principle that reduces coupling and increases cohesion between components in an application. If component A needs B to work, A should receive B from outside instead of creating it itself.

To make this easier to understand, consider an analogy: coupling is the superglue that joins parts of our application, while cohesion is more like a construction set. You can learn more about this principle in this article.

Here is an example of tightly coupled code:

import os class ApiClient: def __init__(self) -> None: self.api_key = os.getenv("API_KEY") # <-- dependency  self.timeout = int(os.getenv("TIMEOUT")) # <-- dependency  class Service: def __init__(self) -> None: self.api_client = ApiClient() # <-- dependency  def main() -> None: service = Service() # <-- dependency  ... if __name__ == "__main__": main() 

Here is the same code using dependency injection:

import os class ApiClient: def __init__(self, api_key: str, timeout: int) -> None: self.api_key = api_key # <-- dependency is injected  self.timeout = timeout # <-- dependency is injected  class Service: def __init__(self, api_client: ApiClient) -> None: self.api_client = api_client # <-- dependency is injected  def main(service: Service) -> None: # <-- dependency is injected  ... if __name__ == "__main__": main( service=Service( api_client=ApiClient( api_key=os.getenv("API_KEY"), timeout=int(os.getenv("TIMEOUT")), ), ), ) 

I followed the dependency injection principle throughout the implementation of the prototype.

Dishka is an IoC container: a framework that manages dependencies in a project. Although FastAPI and FastStream each have their own solutions for this, using an external tool lets us run both within a single codebase.

You can read more about the principle here. The same material also explains what an IoC container is.

Implementing the Prototype

For this article, I built an application prototype that stores information about books. It listens to a queue for incoming events and writes the information from those events to PostgreSQL.

I do not go into detail about writing Docker Compose configuration or setting up migrations. The complete implementation is available on GitHub, with all of that already in place.

. ├── docker-compose.yaml — start RabbitMQ and PostgreSQL here ├── .env — define environment variables here └── book-club ├── domain — domain entities ├── application — application business logic ├── infrastructure — adapters and migrations ├── handlers — HTTP and AMQP controllers and their DTOs ├── ioc.py — dependency container ├── config.py — project configuration ├── main.py — application entry point 

The first thing I would like to start with is the domain. The domain is the foundation of any application: the reason it exists. For example, a pizza delivery application is built around pizza, and a car service application is built around cars.

I am deliberately simplifying this for clarity. In reality, an application's domain is much broader. It can include customers who buy pizza, employees who service cars, and supporting domains for selling drinks or car air fresheners. For the book club, I limited the domain to a single entity: BookDM.

# book_club/domain/entities.py from dataclasses import dataclass @dataclass(slots=True) class BookDM: # DM - Domain model  uuid: str title: str pages: int is_read: bool 

Next, I defined the application's business logic: what the application does. It can save information about a book and return it on request. Let us look at the DTO needed for the business logic layer.

# book_club/application/dto.py from dataclasses import dataclass @dataclass(slots=True) class NewBookDTO: title: str pages: int is_read: bool 

DTO is a pattern for exchanging information between layers. An inner layer must not depend on an outer layer, so it cannot work with objects from outer layers, such as a Pydantic model or a request object. NewBookDTO receives book information from the application's outer layers.

Based on the business logic, I then implemented use-case interactors and defined the interfaces they depend on.

# book_club/application/interfaces.py from abc import abstractmethod from typing import Protocol from uuid import UUID from book_club.domain.entities import BookDM class BookSaver(Protocol): @abstractmethod async def save(self, book: BookDM) -> None: ... class BookReader(Protocol): @abstractmethod async def read_by_uuid(self, uuid: str) -> BookDM | None: ... class UUIDGenerator(Protocol): def __call__(self) -> UUID: ... class DBSession(Protocol): @abstractmethod async def commit(self) -> None: ... @abstractmethod async def flush(self) -> None: ... 

BookSaver — saves information about books.

BookReader — reads information about books.

UUIDGenerator — generates UUIDs.

DBSession — an interface for working with transactions.

# book_club/application/interactors.py from book_club.application import interfaces from book_club.application.dto import NewBookDTO from book_club.domain import entities class GetBookInteractor: def __init__( self, book_gateway: interfaces.BookReader, ) -> None: self._book_gateway = book_gateway async def __call__(self, uuid: str) -> entities.BookDM | None: return await self._book_gateway.read_by_uuid(uuid) class NewBookInteractor: def __init__( self, db_session: interfaces.DBSession, book_gateway: interfaces.BookSaver, uuid_generator: interfaces.UUIDGenerator, ) -> None: self._db_session = db_session self._book_gateway = book_gateway self._uuid_generator = uuid_generator async def __call__(self, dto: NewBookDTO) -> str: uuid = str(self._uuid_generator()) book = entities.BookDM( uuid=uuid, title=dto.title, pages=dto.pages, is_read=dto.is_read ) await self._book_gateway.save(book) await self._db_session.commit() return uuid 

A use-case interactor implements a specific user action. The user has only two available actions, both described in the bodies of the interactors.

At this point, the foundation of the application is ready. Now let us look at the implementation details.

The application's Config:

# book_club/config.py from os import environ as env from pydantic import Field, BaseModel class RabbitMQConfig(BaseModel): host: str = Field(alias='RABBITMQ_HOST') port: int = Field(alias='RABBITMQ_PORT') login: str = Field(alias='RABBITMQ_USER') password: str = Field(alias='RABBITMQ_PASS') class PostgresConfig(BaseModel): host: str = Field(alias='POSTGRES_HOST') port: int = Field(alias='POSTGRES_PORT') login: str = Field(alias='POSTGRES_USER') password: str = Field(alias='POSTGRES_PASSWORD') database: str = Field(alias='POSTGRES_DB') class Config(BaseModel): rabbitmq: RabbitMQConfig = Field(default_factory=lambda: RabbitMQConfig(**env)) postgres: PostgresConfig = Field(default_factory=lambda: PostgresConfig(**env)) 

Notice that I split the configuration into separate parts and then combined them through composition in the Config class. This approach makes dependency injection more effective. If a class depends on specific configuration data, it should receive only that data instead of the configuration for the entire application.

# book_club/infrastructure/database.py from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine from book_club.config import PostgresConfig def new_session_maker(psql_config: PostgresConfig) -> async_sessionmaker[AsyncSession]: database_uri = 'postgresql+psycopg://{login}:{password}@{host}:{port}/{database}'.format( login=psql_config.login, password=psql_config.password, host=psql_config.host, port=psql_config.port, database=psql_config.database, ) engine = create_async_engine( database_uri, pool_size=15, max_overflow=15, connect_args={ "connect_timeout": 5, }, ) return async_sessionmaker(engine, class_=AsyncSession, autoflush=False, expire_on_commit=False) 

Notice that new_session_maker depends only on PostgresConfig. It knows nothing about RabbitMQConfig.

# book_club/infrastructure/broker.py from faststream.rabbit import RabbitBroker from faststream.security import SASLPlaintext from book_club.config import RabbitMQConfig def new_broker(rabbitmq_config: RabbitMQConfig) -> RabbitBroker: return RabbitBroker( host=rabbitmq_config.host, port=rabbitmq_config.port, security=SASLPlaintext( username=rabbitmq_config.login, password=rabbitmq_config.password, ), virtualhost="/", ) 

The same applies to new_broker.

# book_club/infrastructure/gateways.py from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import text from book_club.application.interfaces import BookReader, BookSaver from book_club.domain.entities import BookDM class BookGateway( BookReader, BookSaver, ): def __init__(self, session: AsyncSession): self._session = session async def read_by_uuid(self, uuid: str) -> BookDM | None: query = text("SELECT * FROM books WHERE uuid = :uuid") result = await self._session.execute( statement=query, params={"uuid": uuid}, ) row = result.fetchone() if not row: return None return BookDM( uuid=row.uuid, title=row.title, pages=row.pages, is_read=row.is_read, ) async def save(self, book: BookDM) -> None: query = text("INSERT INTO books (uuid, title, pages, is_read) VALUES (:uuid, :title, :pages, :is_read)") await self._session.execute( statement=query, params={ "uuid": book.uuid, "title": book.title, "pages": book.pages, "is_read": book.is_read, }, ) 

A Gateway encapsulates data access and lets us work with data as a collection without going into implementation details.

Note that BookGateway satisfies the BookSaver and BookReader interfaces mentioned earlier.

After implementing the domain and adapters, I placed the dependencies in a container for subsequent injection.

# book_club/ioc.py from typing import AsyncIterable from uuid import uuid4 from dishka import Provider, Scope, provide, AnyOf, from_context from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from book_club.application import interfaces from book_club.application.interactors import ( GetBookInteractor, NewBookInteractor ) from book_club.config import Config from book_club.infrastructure.database import new_session_maker from dbook_club.infrastructure.gateways import BookGateway class AppProvider(Provider): ... 

Here, I defined a provider for interfaces.UUIDGenerator. It has a scope parameter, which determines the dependency's lifetime. You can find a detailed explanation in the documentation. A dependency with APP scope does not need to be recreated. I used the uuid4 function as a first-class object that satisfies the interfaces.UUIDGenerator interface defined earlier. For uuid4, creating the object once during the application's lifetime is enough, so it uses scope=Scope.APP.

... @provide(scope=Scope.APP) def get_uuid_generator(self) -> interfaces.UUIDGenerator: return uuid4 ... 

Next, I defined the configuration provider, also with scope=Scope.APP. As with uuid4, we only need to create the object once during the application's lifetime. However, from_context means that the Config object must be created outside the provider and supplied when setting up AppProvider.

... config = from_context(provides=Config, scope=Scope.APP) ... 

The next step is to create SQLAlchemy's sessionmaker.

... @provide(scope=Scope.APP) def get_session_maker(self, config: Config) -> async_sessionmaker<...