Async Python service — Litestar + FastStream/NATS, clean layered architecture, DI and import-linter
Example web service for an internet library, built as a reference for clean, layered Python services with both a REST API and async messaging in a single app.
- 🧱 Layered architecture (
domain/adapters/presentors/application) with import boundaries enforced by import-linter - 💉 Dependency Injection with dishka
- 🌐 REST API based on Litestar
- 📨 Async pub/sub messaging with FastStream over NATS JetStream
- 🔌 External API integration (Open Library) with caching and response reduction via asyncly
- 🗄️ Database layer with SQLAlchemy + asyncpg, the Unit of Work pattern, and Alembic migrations
- ⚡ Redis for caching
- 📊 Observability: Prometheus metrics, structured logging with structlog, and optional Sentry
- 🛡️ Rate limiting and CORS out of the box
- 🧪 Auto tests with pytest — unit tests on fakes plus integration tests against containerized services and external APIs
- 🎨 Formatting & linting with Ruff and mypy
- 🐳 Dockerfile with best practices
- 🔁 CI/CD with GitHub Workflows split into separate actions
- ✅ pre-commit hooks
The codebase is split into layers with one-way dependencies, verified on every commit by import-linter:
presentors → application → adapters → domain
(REST / FastStream) (db, redis, nats, open_library)
Enforced contracts:
domainmust not importadaptersorpresentorsadaptersmust not importpresentors- production code must not import
tests
Create a venv in the project folder and install everything with uv:
make developStart the Postgres, Redis, and NATS containers from docker-compose.dev.yaml:
make local # start
make local-down # stop and drop volumespython -m libraryThe API is served on http://127.0.0.1:8000 (see .env for host/port):
| Endpoint | Description |
|---|---|
/docs/swagger |
Swagger UI |
/docs/redoc |
ReDoc |
/metrics |
Prometheus metrics |
Run after dependencies are installed and while make local is running in a
separate terminal:
pytest -vx ./testsTo connect to the database, set the env variables APP_DATABASE_HOST,
APP_DATABASE_PORT, APP_DATABASE_USER, APP_DATABASE_PASSWORD,
APP_DATABASE_NAME.
# apply all migrations
python -m library.adapters.database upgrade head
# generate a new migration (apply existing ones first)
python -m library.adapters.database revision --autogenerate -m "Your message"Separate Makefile targets (suffixed -ci) install dependencies, run checks,
and run tests:
make develop-ci # install dependencies
make lint-ci # run linters - ruff, mypy and import-linter
make test-ci # run tests with pytest and coverageCheck import boundaries on their own:
make import-linterGET /api/v1/books/ Fetch Books
POST /api/v1/books/ Create Book
GET /api/v1/books/{book_id}/ Fetch Book by ID
PATCH /api/v1/books/{book_id}/ Update Book by ID
DELETE /api/v1/books/{book_id}/ Delete Book by ID
GET /api/v1/users/ Fetch Users
POST /api/v1/users/ Create User
GET /api/v1/users/{user_id}/ Fetch User by ID
PATCH /api/v1/users/{user_id}/ Update User by ID
DELETE /api/v1/users/{user_id}/ Delete User by ID
GET /api/v1/open-library/search Search books in Open Library
Released under the MIT License.