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

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

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