FastAPI-Dishka-FastStream

This project is an implementation of "Clean architecture" in combining:

• FastAPI
• Dishka
• FastStream

⚡ Why FastAPI?
The original example application was built using the Litestar framework. However, as the framework evolved, it took a rather specific development direction, so we decided to rewrite the materials using FastAPI. As of today, it is the most optimal framework choice for working with HTTP.

Architecture Overview

Original articles on Habr (Russian):

• Пишем универсальный прототип бэкенд-приложения
• Практическое тестирование приложений

English translations on DEV, published as the Python Backend Architecture and Testing series:

• Clean Architecture in Python with FastAPI, FastStream, and Dishka
• Testing a Python Backend: Unit, Integration, and Service Tests

Quick Start

Prerequisites

• Python 3.10+
• Docker & Docker Compose

Installation

Set up virtual environment and install dependencies:

python3 -m venv venv  # Edit .env if needed
source venv/bin/activate
pip install -r requirements.txt

Configure environment and start services:

cp .env.dist .env
docker compose up -d
export $(grep -v '^#' .env | xargs)  # This command can be useful in the next stages

Initialize database:

alembic upgrade head

Create RabbitMQ queues:

docker exec -it book-club-rabbitmq rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS -V / declare queue name=create_book durable=false
docker exec -it book-club-rabbitmq rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS -V / declare queue name=book_statuses durable=false

Run the project

Full Application HTTP + AMQP (Recommended for demo):

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

but you also can run HTTP API only or AMQP consumer only

// HTTP API Only
uvicorn --factory book_club.main:get_fastapi_app --reload

// AMQP Consumer Only
faststream run --factory book_club.main:get_faststream_app --reload

Usage Examples

// Create a Book via AMQP
docker exec -it book-club-rabbitmq 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 uuid of created book
docker exec -it book-club-rabbitmq rabbitmqadmin -u $RABBITMQ_USER -p $RABBITMQ_PASS get queue=book_statuses count=10

// Get book info by http api
curl http://localhost:8000/book/{uuid}

Testing

Create test database:

docker exec -it book-club-postgres psql -U $POSTGRES_USER -d $POSTGRES_DB -c "CREATE DATABASE test_db"

Run tests:

TEST_DB=test_db pytest tests/ --asyncio-mode=auto