Overview

When building production-grade asynchronous web APIs with Python, teams frequently reinvent the exact same architectural boilerplate: organizing routers and dependency injection, connecting cleanly to Redis, enforcing distributed rate limits, acquiring safe locks for critical sections, and emitting structured JSON telemetry with request correlation identifiers.

fastapi-forge provides an opinionated, clean architectural scaffold designed for long-term maintainability. It eliminates recurring setup overhead while preserving clean domain separation and high throughput.

---

Architectural Separation

The application structure enforces strict boundaries between transport protocols, business rules, and persistence layers:

src/fastapi_forge/
├── api/             # ASGI routers, request validation, response schemas
├── core/            # Configuration, telemetry, logging, and security middleware
├── services/        # Pure domain business logic and coordination
├── repos/           # Database and persistence adapters
└── schemas/         # Strongly-typed Pydantic validation models

By decoupling database queries and Redis interactions from FastAPI route handlers, business rules remain completely isolated and testable without spinning up real web servers or database instances.

---

Three Distinct Redis Patterns

Instead of wrapping Redis in generic, leaky abstraction layers, fastapi-forge isolates Redis usage into three well-defined primitives, each relying on single-command atomic executions or Lua scripts:

1. Invalidation-Safe TTL Cache

Expensive read operations are cached with strict time-to-live enforcement. A unified get_or_set helper handles cache misses atomically, preventing cache stampedes and memory leaks:

from fastapi_forge.core.cache import cache

@router.get("/metrics/{tenant_id}")
async def get_metrics(tenant_id: str):
    return await cache.get_or_set(
        key=f"tenant:{tenant_id}:metrics",
        ttl_seconds=300,
        factory=lambda: calculate_heavy_metrics(tenant_id),
    )

2. Sliding-Window Rate Limiter

Rather than simplistic fixed-window counters that permit bursts across boundary intervals, rate limiting is executed via an atomic Lua script operating on Redis sorted sets (ZSET). Requests outside the active sliding window are cleared and counted atomically within the database engine:

local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now - window)
local current = redis.call('ZCARD', KEYS[1])

if current < limit then
    redis.call('ZADD', KEYS[1], now, now)
    redis.call('EXPIRE', KEYS[1], window)
    return 1
end

return 0

3. Distributed Lock with Ownership Safety

Critical sections executed concurrently across multiple container instances are synchronized via distributed locks. Safe release semantics prevent worker A from inadvertently unlocking a lock that worker B acquired after a timeout:

async with RedisLock(redis, "invoice:generation:1042", ttl_seconds=15) as lock:
    if not lock.acquired:
        raise ConflictError("Invoice is already being generated")
    await generate_invoice(1042)

---

Automated Verification and CI Gates

Every commit in fastapi-forge is verified against four strict automated gates:

1. Linting and Formatting: Fast, exhaustive analysis via ruff with strict flake8 rules enabled.
2. Static Typing: Comprehensive type validation via mypy running under strict mode with zero unchecked calls.
3. In-Memory Integration Testing: Complete test coverage via pytest-asyncio utilizing httpx.ASGITransport and fakeredis. Entire integration suites run within seconds without external service dependencies.
4. Optimized Multi-Stage Containerization: Minimalist Debian-slim Docker images built with non-root security contexts and pinned dependencies.