FastAPI Factory Utilities

A comprehensive library to build production-ready microservices with FastAPI, Beanie, Taskiq, AioPika, and OpenTelemetry.

This library consolidates common patterns, plugins, and utilities for creating modern Python microservices with observability, security, and message-driven architectures built-in from the start.

📚 Documentation

Document	Description
Documentation Index	Master navigation for all documentation
Project Overview	Executive summary and tech stack
Architecture	Technical architecture reference with diagrams
Architecture Decisions	Formal architectural decisions and implementation patterns
Source Tree	Annotated directory structure
Development Guide	Setup, testing, and contribution guide
Agent Skill	Points to canonical skill in DeerHide/agent_skills

Features

🏗️ Application Framework

Abstract Application Builder with plugin architecture for composable microservices
Configuration Management using YAML files and environment variables
Environment-aware configuration (development, staging, production)
Lifecycle Management with startup/shutdown hooks

📖 See Architecture Documentation for detailed component design.

🔐 Security & Authentication

JWT Bearer Authentication with token verification and decoding
Ory Kratos Integration for identity and user management
Ory Hydra Integration for OAuth2 and OpenID Connect flows
Flexible Authentication with custom JWT verifiers and JWK stores

📖 See Security Architecture for authentication flows.

🗄️ Database & ODM

Beanie ODM Plugin for MongoDB with async operations
Document Models with Pydantic v2 validation
Repository Pattern support for clean architecture

📖 See Data Architecture for repository patterns.

📨 Message Broker & Task Queue

AioPika Plugin for RabbitMQ message broker integration
Taskiq Plugin for distributed task queue with Redis backend
S3 Plugin for MinIO / S3 via async aioboto3 (named buckets + DI)
Message-Driven Architecture support with async consumers/producers

📖 See Plugin Architecture for plugin details.

📊 Observability & Monitoring

OpenTelemetry Plugin with automatic instrumentation for:
- FastAPI endpoints
- MongoDB operations
- HTTP client requests (aiohttp)
- RabbitMQ messaging (AioPika)
Distributed Tracing with OTLP exporters (HTTP/gRPC)
Structured Logging with structlog integration
Status Endpoint for health checks and monitoring

📖 See Observability Architecture for tracing setup.

🌐 HTTP Client

AioHttp Plugin with OpenTelemetry instrumentation
Async HTTP operations with connection pooling
Automatic tracing of outbound HTTP requests

🛠️ Services

Status Service - Health check endpoints with reactive monitoring
Audit Service - Event auditing capabilities
Kratos Service - Identity management operations
Hydra Service - OAuth2/OIDC operations

📖 See Service Layer for service details.

Requirements

Python: >= 3.12
Mandatory: FastAPI, Pydantic, structlog, Uvicorn, aiohttp, PyJWT
Extras: Beanie/PyMongo (mongo), AioPika (amqp), aioboto3 (s3), Redis (redis), Taskiq Redis broker (taskiq), OpenTelemetry SDK (otel)

📖 See Project Overview for complete dependency list.

Installation

Using pip

pip install fastapi-factory-utilities
# Backing technologies are extras:
pip install 'fastapi-factory-utilities[mongo,otel]'
pip install 'fastapi-factory-utilities[all]'

Using Poetry

poetry add fastapi-factory-utilities
poetry add fastapi-factory-utilities --extras mongo --extras otel

Extras: mongo, amqp, s3, redis, taskiq, otel, all, testing. Importing a plugin without its extra raises MissingExtraError naming the extra to install. The bundled ASGI server is Uvicorn.

Public API and deprecation

A symbol is public if and only if it is listed in a package __init__.__all__. Import it from that package, not from the defining submodule. Everything else — module paths, module names, class internals — is private and may move in a minor release without a shim.

SemVer

Breaking: removing or renaming a symbol in an __all__; removing an extra; raising the minimum Python.
Adding a public symbol, or moving a private module, is a minor or patch.

Deprecation

A public removal first ships a DeprecationWarning that names the replacement, is held for at least one minor, and is removed no earlier than the next major. The CHANGELOG states the removal version. A deprecation with live consumers is either migrated or withdrawn — it is not left warning indefinitely.

Plugins that open an external connection accept that client (or its factory) as an optional constructor argument. Do not monkeypatch FFU internals.

Quick Start

Here's a minimal example to create a microservice with MongoDB and OpenTelemetry:

from typing import ClassVar
from beanie import Document
from fastapi_factory_utilities.core.app import (
    ApplicationAbstract,
    ApplicationGenericBuilder,
    RootConfig,
)
from fastapi_factory_utilities.core.plugins import PluginAbstract
from fastapi_factory_utilities.core.plugins.odm_plugin import ODMPlugin
from fastapi_factory_utilities.core.plugins.opentelemetry_plugin import OpenTelemetryPlugin


class MyAppConfig(RootConfig):
    """Custom application configuration."""
    pass


class MyApp(ApplicationAbstract):
    """Your microservice application."""

    CONFIG_CLASS: ClassVar[type[RootConfig]] = MyAppConfig
    PACKAGE_NAME: ClassVar[str] = "my_app"
    ODM_DOCUMENT_MODELS: ClassVar[list[type[Document]]] = []

    def configure(self) -> None:
        """Configure your application routes and middleware."""
        # Add your API routers here
        pass

    async def on_startup(self) -> None:
        """Actions to perform on application startup."""
        pass

    async def on_shutdown(self) -> None:
        """Actions to perform on application shutdown."""
        pass


class MyAppBuilder(ApplicationGenericBuilder[MyApp]):
    """Application builder."""

    def get_default_plugins(self) -> list[PluginAbstract]:
        """Get the default plugins."""
        return [
            ODMPlugin(),
            OpenTelemetryPlugin(),
        ]

    def __init__(self, plugins: list[PluginAbstract] | None = None) -> None:
        """Initialize the builder."""
        if plugins is None:
            plugins = self.get_default_plugins()
        super().__init__(plugins=plugins)


# Build and run your application
if __name__ == "__main__":
    MyAppBuilder().build_and_serve()

Create an application.yaml configuration file in your package:

application:
  service_namespace: "my-company"
  service_name: "my-app"
  description: "My awesome microservice"
  version: "1.0.0"
  environment: "development"

server:
  host: "0.0.0.0"
  port: 8000

cors:
  allow_origins: ["*"]
  allow_credentials: true
  allow_methods: ["*"]
  allow_headers: ["*"]

📖 See Architecture - Configuration System for complete configuration options.

Core Components

Application Framework

The ApplicationAbstract class provides the foundation for your microservice:

Plugin System: Extend functionality through composable plugins
Configuration: Type-safe configuration with Pydantic models
Lifecycle Management: Control startup and shutdown behavior
FastAPI Integration: Built-in FastAPI application with customizable routes

The ApplicationGenericBuilder handles:

Configuration loading from YAML files
Plugin initialization and registration
FastAPI application setup
Uvicorn server management

📖 See Architecture - Core Components for detailed class documentation.

Available Plugins

Each plugin extends your application with specific capabilities:

Plugin	Purpose	Documentation
`ODMPlugin`	MongoDB operations with Beanie ODM	Plugin Details
`OpenTelemetryPlugin`	Distributed tracing and metrics	Observability
`TaskiqPlugin`	Background task processing with Redis	Plugin Architecture
`AioPikaPlugin`	RabbitMQ messaging capabilities	Plugin Architecture
`AioHttpPlugin`	Instrumented HTTP client	Plugin Architecture
`S3Plugin`	Async MinIO / S3 (aioboto3), named buckets	S3 skill reference

Plugins follow a consistent lifecycle:

on_load() - Initial setup when plugin is registered
on_startup() - Async initialization during application startup
on_shutdown() - Cleanup during application shutdown

📖 See Plugin Lifecycle for detailed flow.

Agent skill docs (config examples, DI patterns): DeerHide/agent_skills — fastapi-factory-utilities (see also docs/SKILL.md).

Security & Authentication

JWT Authentication

from fastapi_factory_utilities.core.security.jwt import (
    JWTAuthenticationService,
    JWTBearerAuthenticationConfig,
)

# Configure JWT authentication
jwt_config = JWTBearerAuthenticationConfig(
    issuer="https://your-auth-server.com",
    audience="your-api",
)

# Use in FastAPI dependencies
from fastapi import Depends

async def get_current_user(
    token: str = Depends(JWTAuthenticationService),
):
    # Token is automatically verified
    return token.sub

Ory Kratos Integration

from fastapi_factory_utilities.core.services.kratos import (
    KratosIdentityGenericService,
    KratosGenericWhoamiService,
)

# Identity management
kratos_service = KratosIdentityGenericService(base_url="http://kratos:4434")
identity = await kratos_service.get_identity(identity_id="...")

# Session validation
whoami_service = KratosGenericWhoamiService(base_url="http://kratos:4433")
session = await whoami_service.whoami(cookie="...")

📖 See Security Architecture for complete authentication flows.

Configuration System

The configuration system supports:

YAML Files: Store configuration in application.yaml
Environment Variables: Override values via environment variables
Type Safety: Pydantic models ensure type correctness
Environment-Specific: Different configs for dev/staging/production
Frozen Models: Immutable configuration prevents accidental changes

from fastapi_factory_utilities.core.app.config import (
    RootConfig,
    BaseApplicationConfig,
)
from pydantic import Field

class MyCustomConfig(BaseModel):
    """Custom configuration section."""
    api_key: str = Field(description="External API key")
    timeout: int = Field(default=30, description="Request timeout")

class MyAppConfig(RootConfig):
    """Extended application configuration."""
    my_custom: MyCustomConfig = Field(description="Custom configuration")

📖 See Configuration Hierarchy for all configuration options.

Example Application

This library includes a complete example application demonstrating key features:

# Run the example application
fastapi_factory_utilities-example

The example shows:

Application structure with plugins
Configuration management
API router organization
Document models with Beanie
OpenTelemetry instrumentation

Source code: src/fastapi_factory_utilities/example/

📖 See Source Tree Analysis for complete directory structure.

Development

Prerequisites

Python 3.12+
Poetry for dependency management
Docker (optional, for containerized development)

Setup Development Environment

# Clone the repository
git clone https://github.com/DeerHide/fastapi_factory_utilities.git
cd fastapi_factory_utilities

# Run the setup script
./scripts/setup_dev_env.sh

# Or manually:
poetry install --with test
poetry run pre-commit install

📖 See Development Guide for complete setup instructions.

Running Tests

# Run all tests with coverage
poetry run pytest --cov=src --cov-report=html --cov-report=term

# Run specific tests
poetry run pytest tests/units/test_exceptions.py

# Run tests in parallel
poetry run pytest -n auto

Downstream services that want the shipped infra doubles (mongomock, fakeredis, moto, Taskiq InMemoryBroker, OTel in-memory exporters, recording AMQP publisher) install the optional extra and get fixtures via the pytest11 plugin:

pip install 'fastapi_factory_utilities[testing]'

📖 See Development Guide - Testing for the driver-seam strategy, RepositoryContract, and container-only boundaries.

Code Quality

# Run all pre-commit hooks
poetry run pre-commit run --all-files

# Format code
poetry run ruff format src tests
poetry run ruff check --fix src tests

# Type checking
poetry run mypy

📖 See Development Guide - Code Style for conventions.

Docker Development

# Build and run in container
./scripts/dev-in-container.sh

📖 See Development Guide - Docker for container setup.

Architecture

graph TB
    App[ApplicationAbstract]
    Builder[ApplicationGenericBuilder]
    FastAPI[FastAPI Instance]

    Builder -->|builds| App
    App -->|provides| FastAPI

    subgraph Plugins
        ODM[ODM Plugin<br/>MongoDB/Beanie]
        OTel[OpenTelemetry Plugin<br/>Tracing]
        Taskiq[Taskiq Plugin<br/>Task Queue]
        AioPika[AioPika Plugin<br/>RabbitMQ]
        Http[AioHttp Plugin<br/>HTTP Client]
        S3[S3 Plugin<br/>MinIO/S3]
    end

    App -->|registers| Plugins

    subgraph Services
        Status[Status Service]
        Audit[Audit Service]
        Kratos[Kratos Service]
        Hydra[Hydra Service]
    end

    App -->|provides| Services

📖 See Architecture Documentation for detailed architecture diagrams and patterns.

Project Structure

fastapi_factory_utilities/
├── src/fastapi_factory_utilities/
│   ├── core/           # 🎯 Main library code
│   │   ├── app/        # Application framework
│   │   ├── plugins/    # Plugin implementations
│   │   ├── security/   # Authentication/authorization
│   │   ├── services/   # Business services
│   │   └── utils/      # Utility functions
│   └── example/        # 📚 Usage example
├── tests/              # Test suite
├── docs/knowledge/     # 📖 Detailed documentation
└── docker/             # Docker configurations

📖 See Source Tree Analysis for complete annotated structure.

Contributing

Contributions are welcome! This project follows clean architecture principles and emphasizes:

Type safety with comprehensive type annotations
Async/await patterns for I/O operations
Plugin-based extensibility
Comprehensive testing
Clean code with proper documentation

Please ensure:

All tests pass (poetry run pytest)
Code is properly formatted (poetry run ruff format)
Type checking passes (poetry run mypy)
Pre-commit hooks pass (poetry run pre-commit run --all-files)

📖 See Development Guide for complete contribution workflow.

Security

For security concerns, please review our Security Policy.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Resources

Repository: https://github.com/DeerHide/fastapi_factory_utilities
Issues: https://github.com/DeerHide/fastapi_factory_utilities/issues
PyPI: https://pypi.org/project/fastapi-factory-utilities/
Documentation: docs/knowledge/index.md

Related Projects

FastAPI - Modern web framework
Beanie - MongoDB ODM
Taskiq - Distributed task queue
AioPika - RabbitMQ client
OpenTelemetry - Observability framework
Ory - Identity & access management

Built with ❤️ for modern Python microservices

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fastapi_factory_utilities-6.4.0.tar.gz (161.7 kB view details)

Uploaded Aug 15, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

fastapi_factory_utilities-6.4.0-py3-none-any.whl (243.1 kB view details)

Uploaded Aug 15, 2026 Python 3

File details

Details for the file fastapi_factory_utilities-6.4.0.tar.gz.

File metadata

Download URL: fastapi_factory_utilities-6.4.0.tar.gz
Upload date: Aug 15, 2026
Size: 161.7 kB
Tags: Source
Uploaded using Trusted Publishing? Yes
Uploaded via: twine/6.1.0 CPython/3.13.7

File hashes

Hashes for fastapi_factory_utilities-6.4.0.tar.gz
Algorithm	Hash digest
SHA256	`77dfb00faead62966575ded1e537035300824a11c8907b0a567a8e35acba3a0d`
MD5	`098dc45c73ed3874898c259a54ccf45b`
BLAKE2b-256	`62912b8e737efc77e8b79b021e7ede1a95b8648c1e4677f92813f6f5cebfb135`

See more details on using hashes here.

Provenance

The following attestation bundles were made for fastapi_factory_utilities-6.4.0.tar.gz:

Publisher: ci.yml on DeerHide/fastapi_factory_utilities

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Statement:
- Statement type: https://in-toto.io/Statement/v1
- Predicate type: https://docs.pypi.org/attestations/publish/v1
- Subject name: fastapi_factory_utilities-6.4.0.tar.gz
- Subject digest: 77dfb00faead62966575ded1e537035300824a11c8907b0a567a8e35acba3a0d
- Sigstore transparency entry: 2475079156
- Sigstore integration time: Aug 15, 2026
Source repository:
- Permalink: DeerHide/fastapi_factory_utilities@8b9f3898123a1f639eb0e0118ef1ec0c1d4c20b3
- Branch / Tag: refs/tags/v6.4.0
- Owner: https://github.com/DeerHide
- Access: public
Publication detail:
- Token Issuer: https://token.actions.githubusercontent.com
- Runner Environment: self-hosted
- Publication workflow: ci.yml@8b9f3898123a1f639eb0e0118ef1ec0c1d4c20b3
- Trigger Event: push