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[AsyncSession]:
    return new_session_maker(config.postgres)
...

Here, I also set scope=Scope.APP because we do not need to recreate async_sessionmaker. This provider depends on Config, as specified in the function signature. We need async_sessionmaker to create AsyncSession objects.

...
@provide(scope=Scope.REQUEST)
async def get_session(self, session_maker: async_sessionmaker[AsyncSession]) -> AsyncIterable[AnyOf[
    AsyncSession,
    interfaces.DBSession,
]]:
    async with session_maker() as session:
        yield session
...

Here, the scope is Scope.REQUEST. This object will be created for each request: a new SQLAlchemy session for each user request. The provider's signature indicates that it depends on async_sessionmaker[AsyncSession] and that the returned object must satisfy two interfaces: sqlalchemy.AsyncSession and interfaces.DBSession.

Using a generator lets us maintain session state, then return to the provider after responding to the user's request and exit the context manager. This returns resources to the connection pool.

...
book_gateway = provide(
    BookGateway,
    scope=Scope.REQUEST,
    provides=AnyOf[interfaces.BookReader, interfaces.BookSaver]
)
...

This provider creates BookGateway, which depends on sqlalchemy.AsyncSession and must satisfy interfaces.BookReader and interfaces.BookSaver. The interactors depend on these interfaces. Their creation looks like this:

...
get_book_interactor = provide(GetBookInteractor, scope=Scope.REQUEST)
create_new_book_interactor = provide(NewBookInteractor, scope=Scope.REQUEST)
...

Under the hood, dishka calls the constructors of BookGateway, GetBookInteractor, and NewBookInteractor, whose signatures depend on interfaces.BookReader, interfaces.BookSaver, interfaces.DBSession, and interfaces.UUIDGenerator. Dependencies with Scope.REQUEST are created for the request. For dependencies with Scope.APP, dishka tries to find them in its storage. If they are not there, dishka creates and stores them.

When an event arrives, whether an HTTP request or a message from the queue, a handler starts the dependency creation chain. Once the event has been processed, those dependencies are finalized. In this design, all dependencies in the chain share one database connection per event. This allows transaction atomicity and efficient use of connection pool resources.

Now let us look at the controller layer.

BookSchema is the DTO we use to communicate between the controller layer and the external layer. Although the prototype has two controllers, we only need one model, equally well supported by FastStream and FastAPI.

Using a single data model for different tasks often violates the Single Responsibility Principle (SRP). In this example, this is a deliberate simplification with one purpose: to show clearly that Pydantic models are independent of frameworks and can be used with different frameworks.

This model alone provides the basis for validation and automatic documentation generation.

# controllers/schemas.py
from pydantic import BaseModel


class BookSchema(BaseModel):
    title: str
    pages: int
    is_read: bool

# book_club/controllers/amqp.py
from dishka.integrations.base import FromDishka
from faststream.rabbit import RabbitRouter

from book_club.application.dto import NewBookDTO
from book_club.application.interactors import NewBookInteractor
from book_club.controllers.schemas import BookSchema

AMQPBookController = RabbitRouter()


@AMQPBookController.subscriber("create_book")
@AMQPBookController.publisher("book_statuses")
async def handle(
    data: BookSchema,
    interactor: FromDishka[NewBookInteractor],
) -> str:
    dto = NewBookDTO(
        title=data.title,
        pages=data.pages,
        is_read=data.is_read
    )
    uuid = await interactor(dto)
    return uuid

Above is the controller that saves information about a book. As the code shows, it accepts BookSchema. The controller listens to the create_book queue and, when an event arrives, tries to convert its data into BookSchema. If that fails, a pydantic.ValidationError is raised.

If it succeeds, handle is called. It depends on NewBookInteractor, which triggers the dependency chain described above. The interactor then runs, and the UUID of the newly created book record is sent to the book_statuses queue.

# book_club/controllers/http.py
from typing import Annotated
from uuid import UUID

from dishka.integrations.base import FromDishka as Depends
from dishka.integrations.litestar import inject
from litestar import Controller, route, HttpMethod
from litestar import status_codes
from litestar.exceptions import HTTPException
from litestar.params import Body

from book_club.application.interactors import GetBookInteractor
from book_club.controllers.schemas import BookSchema


class HTTPBookController(Controller):
    path = "/book"

    @route(http_method=HttpMethod.GET, path="/{book_id:uuid}")
    @inject
    async def get_book(
            self,
            book_id: Annotated[UUID, Body(description="Book ID", title="Book ID")],
            interactor: Depends[GetBookInteractor],
    ) -> BookSchema:
        book_dm = await interactor(uuid=str(book_id))
        if not book_dm:
            raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Book not found")
        return BookSchema(
            title=book_dm.title,
            pages=book_dm.pages,
            is_read=book_dm.is_read,
        )

Above is the controller that returns information about books. The dependency chain is built in exactly the same way as in the previous example. It expects book_id as a UUID, which is used to look up the book in the database.

We can now put everything together. The application's entry point looks like this:

import dishka_faststream
from dishka import make_async_container
from dishka.integrations.fastapi import setup_dishka
from fastapi import FastAPI
from faststream import FastStream

from book_club.config import Config
from book_club.controllers.http import book_router
from book_club.controllers.amqp import AMQPBookController
from book_club.infrastructure.resources.broker import new_broker
from book_club.ioc import AppProvider

config = Config()
container = make_async_container(AppProvider(), context={Config: config})


def get_faststream_app() -> FastStream:
    broker = new_broker(config.rabbitmq)
    faststream_app = FastStream(broker)
    dishka_faststream.setup_dishka(container, faststream_app, auto_inject=True)
    broker.include_router(AMQPBookController)
    return faststream_app


def get_fastapi_app() -> FastAPI:
    fastapi_app = FastAPI(title="Book club")
    fastapi_app.include_router(book_router)

    setup_dishka(container, fastapi_app)

    return fastapi_app


def get_app():
    faststream_app = get_faststream_app()
    fastapi_app = get_fastapi_app()

    fastapi_app.add_event_handler("startup", faststream_app.broker.start)
    fastapi_app.add_event_handler("shutdown", faststream_app.broker.close)

    return fastapi_app

Config and container are placed in the global scope. This lets us run the HTTP and AMQP parts independently in separate processes if necessary.

The application is now ready to run and test.

For the application to work correctly, you need RabbitMQ and PostgreSQL containers, a virtual environment, and database migrations. You will find detailed instructions, docker-compose.yaml, and migrations in the project's source code.

Start the application:

uvicorn --factory book_club.main:get_app --reload

Send a message to the create_book queue:

rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS \
publish exchange=amq.default routing_key=create_book \
payload='{"title": "The Brothers Karamazov", "pages": 928, "is_read": true}'

Read a message from the book_statuses queue to obtain the UUID:

rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS get queue=book_statuses count=1
+---------------+--------------------------------------+
|  routing_key  |               payload                |
+---------------+--------------------------------------+
| book_statuses | bc5fe42d-bedd-42e4-b518-89d088808237 |
+---------------+--------------------------------------+

Use that UUID to retrieve the book information through the HTTP API:

curl http://localhost:8000/book/bc5fe42d-bedd-42e4-b518-89d088808237
{"title":"The Brothers Karamazov","pages":928,"is_read":true}

If you want to run HTTP and AMQP in separate processes, you can do it like this:

faststream run --factory book_club.main:get_faststream_app --reload

uvicorn --factory book_club.main:get_fastapi_app --reload

Conclusion

Thank you for reading! I encourage you to explore the prototype's source code yourself. The second part shows how to test the same backend through its HTTP and AMQP interfaces. Also join our communities, where you can meet contributors to the technologies discussed in this article and ask questions: FastAPI / Litestar (ASGI Community), FastStream Community, and Reagento Community.

Continue with Part 2: Testing a Python Backend — Unit, Integration, and Service Tests.

Do your HTTP handlers and message consumers share the same application layer? How do you keep framework-specific code out of your business logic?

Comments

[DEV SUPPORTS]

You need to verify your account.

tr.ee/dev-to

[Sehat1137]

bruh