DEV Community: Sehat1137

Testing a Python Backend: Unit, Integration, and Service Tests

Sehat1137 — Mon, 12 Oct 2026 01:18:06 +0000

How do you test a backend that accepts both HTTP requests and AMQP messages? I tested the same application through both interfaces, using a real test database and an in-memory message broker. This article compares unit, integration, and service tests and explains where mocks help and where they make tests fragile.

This is an English translation of my article originally published in Russian on Habr: Practical Testing of Python Applications.

This is the second part of the series. Part 1 explains the backend architecture. Here, I share how I tested that application and how its architecture made the process easier.

I have divided the application's tests into three categories:

Unit tests
Integration tests
Service tests

We will use pytest and Faker to write the tests. These are key libraries for testing in Python, and I strongly recommend getting familiar with them if you have not used them before.

Jump to a section:

Introduction to Mock Objects
Writing Unit Tests
Writing Integration Tests
Writing Service Tests
Conclusion

Introduction to Mock Objects

Mock objects can seem complicated to developers who are just starting out. Many articles mix up mock objects and monkeypatching, but when we use Dependency Injection (DI), we do not need monkeypatching in our tests.

Moreover, if your tests require monkeypatching, that is a bad sign. Your code probably uses global variables. You can read about their drawbacks here.

What Is a Mock Object?

A mock object is a special object that simulates the behavior of a real object during testing.

Imagine we are writing a function that retrieves a weather forecast in kelvins and converts the temperature to degrees Celsius:

from typing import Any

def get_celsius_temp(weather_service: Any) -> str:
    kelvin_temp: float = weather_service.get_temp()
    return kelvin_temp - 273.15

Without mock objects, testing this function would require:

A working weather service. If it is unavailable, the tests will fail.
A city with a constant temperature of +15°C. No such city exists.
A city with a constant temperature of −15°C. That does not exist either.

Using a mock object solves these problems and lets us test the function as it is:

from unittest.mock import Mock

def test_positive_temp() -> None:
    weather_svc = Mock()
    weather_svc.get_temp = Mock(return_value=288.15)
    assert get_celsius_temp(weather_svc) == 15

def test_negative_temp() -> None:
    weather_svc = Mock()
    weather_svc.get_temp = Mock(return_value=258.15)
    assert get_celsius_temp(weather_svc) == -15

Types of Mock Objects in Python

The unittest library provides several useful classes:

Mock — the base class
MagicMock — an extension of Mock with magic methods implemented
AsyncMock — an extension of MagicMock with async/await support

What Mock Objects Can Do

Mock objects can:

Record call history and arguments
Count calls
Raise exceptions through side_effect

from unittest.mock import Mock

m = Mock(return_value=42)

result1 = m(foo="bar")
result2 = m("test", 1, 3)

print(m.mock_calls)  # [call(foo='bar'), call('test', 1, 3)]
print(m.call_count)  # 2
print(m.called)  # True
print(result1, result2)  # 42, 42

z = Mock(side_effect=Exception("Mocked exception"))

z()  # Exception: Mocked exception

Checking Calls

Mock objects provide convenient methods for checking how they were used:

from unittest.mock import Mock

x = Mock()
x(foo="bar")

x.assert_called()  # Checks that the object was called
x.assert_called_with(foo="bar")  # Checks the call arguments
x.assert_called_once()  # Checks that it was called exactly once

The create_autospec Function

The create_autospec function creates a mock object that checks method signatures:

from unittest.mock import create_autospec

class Foo:
    def bar(self, some_id: int): ...

mocked_foo = create_autospec(Foo)

mocked_foo.bar(some_id=1)  # OK
mocked_foo.baar()  # AttributeError: Mock object has no attribute 'baar'
mocked_foo.bar()   # TypeError: missing a required argument: 'some_id'

Writing Unit Tests

Unit tests check individual parts of a program, such as functions, methods, and classes, in isolation from the rest of the code. They are usually the most numerous and fragile tests in a project. In our example, we will write unit tests for the business logic implemented in the interactors.

Here is an example of the interactor that retrieves a book, along with its test:

# book_club/application/interactors.py
...

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)


# tests/test_application.py
...
@pytest.fixture
def get_book_interactor() -> GetBookInteractor:
    book_gateway = create_autospec(interfaces.BookReader)
    return GetBookInteractor(book_gateway)


@pytest.mark.parametrize("uuid", [str(uuid4()), str(uuid4())])
async def test_get_book(get_book_interactor: GetBookInteractor, uuid: str) -> None:
    result = await get_book_interactor(uuid=uuid)
    get_book_interactor._book_gateway.read_by_uuid.assert_awaited_once_with(
        uuid=uuid
    )
    assert result == get_book_interactor._book_gateway.read_by_uuid.return_value

In the fixture, I create an interactor, but replace the real BookGateway, which accesses the database, with a mock object that conforms to the BookReader interface. This lets us check that the interactor works correctly.

Now let us look at the interactor that saves a book:

# book_club/application/interactors.py
...

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


# tests/test_application.py
...

@pytest.fixture
def new_book_interactor(faker: Faker) -> NewBookInteractor:
    db_session = create_autospec(interfaces.DBSession)
    book_gateway = create_autospec(interfaces.BookSaver)
    uuid_generator = MagicMock(return_value=faker.uuid4())
    return NewBookInteractor(db_session, book_gateway, uuid_generator)


async def test_new_book_interactor(new_book_interactor: NewBookInteractor, faker: Faker) -> None:
    dto = NewBookDTO(
        title=faker.pystr(),
        pages=faker.pyint(),
        is_read=faker.pybool(),
    )
    result = await new_book_interactor(dto=dto)
    uuid = str(new_book_interactor._uuid_generator())
    new_book_interactor._book_gateway.save.assert_awaited_with(
        entities.BookDM(
            uuid=uuid,
            title=dto.title,
            pages=dto.pages,
            is_read=dto.is_read,
        )
    )
    new_book_interactor._db_session.commit.assert_awaited_once()
    assert result == uuid

Because I used dependency injection (DI) when writing the code, I can easily test the business logic without monkeypatching. All dependencies are replaced with mock objects, which fully isolates the tests.

TDD and Clean Architecture

Clean Architecture, combined with extensive use of DI, makes TDD (Test-Driven Development) much easier. When dependencies are clearly defined and injected from outside, writing tests before the implementation becomes a natural process. You can first describe the expected behavior through tests, create mock objects for all dependencies, and then implement the logic without waiting for external services or infrastructure to be ready. This approach leads to a more carefully considered API design and provides quick feedback on code quality.

Solitary vs. Sociable: Choosing Test Boundaries and the Cost of Mock Objects

When writing unit tests, it is important to define their boundaries:

Solitary tests, which mock all neighboring components, are useful for checking logic in isolation, but make changes to the codebase painful.
Sociable tests, which mock only external contracts, check how components interact through their public API. This gives you more freedom to change the internal implementation.

In our case, testing the interactors is a solitary approach: we isolate the business logic by mocking all dependencies, including database gateways and UUID generators. This lets us quickly check the interactor's logic, but comes with risks. Each mock fixes a specific interaction interface between components. During refactoring, for example, if we rename book_gateway.read_by_uuid(), all the tests will break even though the application itself still works. We become tied to internal APIs that should not be rigidly fixed.

For a complex domain model, a more resilient solution is the sociable approach, where several domain objects are tested together without mocks. These tests make it easy to check both the happy path and edge cases while fixing only the public API. This makes tests less fragile and gives you more freedom to change the codebase later. However, if the domain is trivial, service tests can cover it completely.

You can learn more about the types of unit tests and their use in a separate article.

In real projects with complex business logic, you should prefer sociable unit tests for testing the domain. The business logic in this educational example does not allow us to implement such tests. However, here is an example of what they could look like:

async def test_book_creation_with_validation(
    # Real dependencies, not mock objects
    session,
    book_gateway,
    uuid_generator,
    faker
):
    db_session = session
    interactor = NewBookInteractor(
        db_session=db_session,
        book_gateway=book_gateway,
        uuid_generator=uuid_generator
    )

    book_data = NewBookDTO(
        title=faker.pystr(min_chars=1, max_chars=200),
        pages=faker.pyint(min_value=1, max_value=1000),
        is_read=faker.pybool()
    )

    book_id = await interactor(book_data)

    created_book = await book_gateway.read_by_uuid(book_id)

    assert created_book is not None
    assert created_book.title == book_data.title
    assert created_book.pages == book_data.pages
    assert created_book.is_read == book_data.is_read
    assert created_book.uuid == book_id

async def test_book_creation_with_invalid_data(
    # Real dependencies, not mock objects
    session,
    book_gateway,
    uuid_generator
):
    interactor = NewBookInteractor(
        db_session=session,
        book_gateway=book_gateway,
        uuid_generator=uuid_generator
    )

    invalid_book_data = NewBookDTO(
        title="",
        pages=-5,
        is_read=False
    )

    with pytest.raises(ValueError) as exc_info:
        await interactor(invalid_book_data)

    assert "Book title cannot be empty" in str(exc_info.value)

Writing Integration Tests

Integration tests check how an application interacts with external systems. Our application has two external integrations:

RabbitMQ
Postgres

FastStream handles all communication with the queue, so we will not write tests for that: there is no point in testing code we do not write. If you have your own abstractions for working with the queue, you will, of course, need to cover them with integration tests as well.

The Postgres integration is encapsulated in BookGateway, so we will test it through this class's public API:

# tests/conftest.py
...

@pytest.fixture(scope="session")
async def session_maker(...) -> async_sessionmaker[AsyncSession]:
    engine = create_async_engine(...)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    return async_sessionmaker(
        bind=engine,
        class_=AsyncSession,
        autoflush=False,
        expire_on_commit=False
    )


@pytest.fixture
async def session(
    session_maker: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[AsyncSession, Any]:
    async with session_maker() as session:
        session.commit = AsyncMock()
        yield session
        await session.rollback()


# tests/test_infrastructure.py
...

@pytest.fixture
async def book_gateway(session: AsyncSession) -> BookGateway:
    return BookGateway(session=session)


async def test_create_book(
    session: AsyncSession, book_gateway: BookGateway, faker: Faker
) -> None:
    uuid = faker.uuid4()
    title = faker.pystr()
    pages = faker.pyint()
    is_read = faker.pybool()
    await session.execute(
        insert(Book).values(
            uuid=uuid,
            title=title,
            pages=pages,
            is_read=is_read
        )
    )
    result = await book_gateway.read_by_uuid(uuid)
    assert result.title == title
    assert result.pages == pages
    assert result.is_read is is_read


async def test_save_book(
    session: AsyncSession,
    book_gateway: BookGateway,
    faker: Faker
) -> None:
    book_dm = BookDM(
        uuid=faker.uuid4(),
        title=faker.pystr(),
        pages=faker.pyint(),
        is_read=faker.pybool(),
    )
    await book_gateway.save(book_dm)

    result = await session.execute(
        select(Book).where(Book.uuid == book_dm.uuid)
    )
    rows = result.fetchall()
    assert len(rows) == 1
    book = rows[0][0]
    assert book.title == book_dm.title
    assert book.pages == book_dm.pages
    assert book.is_read == book_dm.is_read

The tests need a separate database. The simplest option is to create it in advance alongside your local development database. In a CI environment, you can start the test database as part of pipeline jobs. Here are examples for GitHub Actions and GitLab CI.

Writing Service Tests

Service tests sit between end-to-end tests and integration tests. They check the application as it runs in production, while allowing some external dependencies to be replaced.

These tests check end-to-end scenarios through external interfaces, involving all application components. For example:

async def test_create_user(client):
   response = await client.post(
       "/users",
       json={
           "email": "test@test.com",
           "password": "password"
        },
   )
   assert response.status_code == 201
   assert "id" in response.json()

Our application has two external interfaces to test: HTTP and AMQP.

To test the application, we need to create an entry point and wrap it in a fixture. Since the application consists of many isolated components, we need to assemble them all.

Ideally, you should create a factory for generating application entry points and reuse it in fixtures. For simplicity, the demonstration example does not include such a factory.

HTTP Service Tests

Most HTTP frameworks provide built-in testing tools. FastAPI is no exception: its documentation has a section on testing. Setting up the test environment is straightforward:

from collections.abc import AsyncIterator

import pytest
from dishka import AsyncContainer
from faker import Faker
from fastapi import FastAPI
from dishka.integrations.fastapi import setup_dishka
from httpx import AsyncClient, ASGITransport
from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncSession

from book_club.infrastructure.models import Book
from book_club.controllers.http import book_router


@pytest.fixture
async def http_app(container: AsyncContainer) -> FastAPI:
    app = FastAPI()
    app.include_router(router=book_router)
    setup_dishka(container, app)
    return app


@pytest.fixture
async def http_client(http_app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=http_app),
        base_url="http://",
    ) as client:
        yield client

Since we have only one endpoint, we will use just one test scenario. We will check the successful retrieval of a book that we first save to the database:

...
async def test_get_book(
    session: AsyncSession,
    http_client: AsyncClient,
    faker: Faker,
) -> None:
    uuid = faker.uuid4()
    title = faker.pystr(min_chars=3, max_chars=120)
    pages = faker.pyint()
    is_read = faker.pybool()

    await session.execute(
        insert(Book).values(
            uuid=uuid,
            title=title,
            pages=pages,
            is_read=is_read
        ),
    )

    result = await http_client.get(f"/book/{uuid}")
    assert result.status_code == 200
    assert result.json()["title"] == title
    assert result.json()["pages"] == pages
    assert result.json()["is_read"] == is_read
...

AMQP Service Tests

FastStream also provides convenient tools for testing message handlers. You do not need to start a real queue: you can use an in-memory broker.

...
@pytest.fixture
async def broker() -> RabbitBroker:
    broker = RabbitBroker()
    broker.include_router(AMQPBookController)
    return broker


@pytest.fixture
async def amqp_app(
    broker: RabbitBroker,
    container: AsyncContainer,
) -> FastStream:
    app = FastStream(broker)
    faststream_integration.setup_dishka(container, app, auto_inject=True)
    return app


@pytest.fixture
async def amqp_client(amqp_app: FastStream) -> AsyncIterator[RabbitBroker]:
    async with TestRabbitBroker(amqp_app.broker) as br:
        yield br

...

The test itself is simple: I send a message in the required format to the in-memory queue and check that the handler has done its job by verifying that the book data appeared in the database:

@pytest.mark.asyncio
async def test_save_book(
    amqp_client: RabbitBroker,
    session: AsyncSession,
    faker: Faker,
) -> None:
    title = faker.name_nonbinary()
    pages = faker.pyint()
    is_read = faker.pybool()
    await amqp_client.publish(
        {
            "title": title,
            "pages": pages,
            "is_read": is_read
        },
        queue="create_book",
    )

    result = await session.execute(
        select(Book).where(
            Book.title == title,
            Book.pages == pages,
            Book.is_read == is_read
        )
    )
    rows = result.fetchall()
    assert len(rows) == 1
    book = rows[0][0]
    assert book.title == title
    assert book.pages == pages
    assert book.is_read == is_read
...

Conclusion

Throughout this article, we have looked at different approaches to testing, from isolated unit tests to end-to-end service tests. The main point I want to convey is this: in a real project, the best approach is to combine sociable tests for a complex domain with service and integration tests for everything else.

I encourage you to explore the prototype's source code yourself. There, you will find working versions of the examples discussed, including service tests for the HTTP and AMQP interfaces.

Also join our communities, where you can talk to contributors to the technologies mentioned in this article and ask questions:

Write tests: they make your code more reliable!

Where do you draw the boundary between unit and service tests? Which dependencies do you replace with mocks, and which do you keep real?

Clean Architecture in Python with FastAPI, FastStream, and Dishka

Sehat1137 — Mon, 12 Oct 2026 01:18:05 +0000

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?