FastAPI Forge 🔨

Production-ready toolkit for FastAPI applications

FastAPI Forge provides battle-tested patterns and utilities for building production-ready FastAPI applications with minimal boilerplate. Focus on your business logic while we handle the production concerns.

✨ Features

🪵 Production Logging: Generic JSON logging + Datadog-optimized formatter with progressive truncation
🔍 Event Loop Monitoring: Detect and diagnose blocking operations in async applications
🗑️ GC Monitoring: Track Python GC behavior with dynamic threshold calculation based on system resources
🚀 FastAPI Templates: Production-ready app templates and best practices
🤖 LangChain Integration: Robust fallback utilities for LLM chains with streaming support
📊 Observability: Platform-agnostic logging (ELK, Splunk, Grafana) with optional Datadog APM
⚙️ Gunicorn/Uvicorn: Optimized configurations for production deployment
🎯 Zero Dependencies: Core logging features require only Python stdlib

🚀 Quick Start

Installation

# Basic installation (logging only)
pip install fastapi-forge

# With FastAPI
pip install fastapi-forge[fastapi]

# With Gunicorn for production
pip install fastapi-forge[gunicorn]

# With Datadog integration
pip install fastapi-forge[datadog]

# With LangChain integration
pip install fastapi-forge[langchain]

# Everything
pip install fastapi-forge[all]

Basic Usage

from fastapi import FastAPI
from fastapi_forge.logging import configure_logging

# Configure generic JSON logging (works with any platform)
configure_logging(formatter="json")

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello from FastAPI Forge!"}

Run with Gunicorn:

gunicorn main:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

📚 Documentation

Logging

FastAPI Forge provides two production-ready JSON formatters:

JSONFormatter (default): Generic formatter compatible with any log aggregation platform
DatadogJSONFormatter: Datadog-optimized with APM trace correlation

Key Features

✅ Platform Agnostic: Works with ELK Stack, Splunk, Grafana Loki, CloudWatch, Datadog, etc.
✅ Progressive Truncation: 3-stage intelligent size management (Docker 16KB limit)
✅ Handler Isolation: Separate handlers for Gunicorn ↔ Application logs
✅ Smart Filtering: Health checks (/api/_/health), Langfuse, Langchain, httpx
✅ Exception Formatting: Structured exception info with traceback truncation
✅ stdout/stderr Separation: INFO → stdout, WARNING+ → stderr

Quick Start - Generic JSON

Works with any log aggregation platform (default):

from fastapi import FastAPI
from fastapi_forge.logging import configure_logging
import logging

# Configure generic JSON logging
configure_logging(formatter="json")  # or just configure_logging()

logger = logging.getLogger(__name__)
app = FastAPI()

@app.get("/")
def root():
    logger.info("Request received", extra={"user_id": "123"})
    return {"status": "ok"}

Output:

{
  "timestamp": "2025-10-27T10:00:00.123Z",
  "level": "INFO",
  "logger": "__main__",
  "message": "Request received",
  "user_id": "123"
}

Datadog APM Integration

Use DatadogJSONFormatter for automatic log-trace correlation:

# main.py
from dotenv import load_dotenv
load_dotenv()

from fastapi_forge.logging import configure_logging

# Use Datadog-optimized formatter
configure_logging(formatter="datadog")

from fastapi import FastAPI
app = FastAPI()

Environment Variables:

# Datadog APM
DD_SERVICE=my-api
DD_ENV=production
DD_TRACE_ENABLED=true
DD_TRACE_LOGS_INJECTION=true  # Critical for trace correlation
DD_PROFILING_ENABLED=true

# Logging
LOG_LEVEL=INFO

Run with ddtrace:

ddtrace-run gunicorn main:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000

Output (with Datadog):

{
  "timestamp": "2025-10-27T10:00:00.123Z",
  "level": "INFO",
  "status": "info",
  "logger": "__main__",
  "message": "Request received",
  "user_id": "123",
  "dd.trace_id": "1234567890123456",
  "dd.span_id": "9876543210",
  "dd_service": "my-api",
  "dd_env": "production"
}

Advanced Usage

from fastapi_forge.logging import (
    JSONFormatter,
    DatadogJSONFormatter,
    HealthCheckFilter,
    get_logging_config,
)
import logging.config

# Option 1: Get config and customize
config = get_logging_config(formatter="json")
config['root']['level'] = 'DEBUG'
logging.config.dictConfig(config)

# Option 2: Use formatter directly
import logging
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())  # or DatadogJSONFormatter()
logger = logging.getLogger()
logger.addHandler(handler)

Filters

Built-in filters to reduce log noise:

HealthCheckFilter: Filters /api/health/heartbeat, /api/_/health
LangfuseFilter: Filters Langfuse library noise
LangchainFilter: Filters Langchain library noise
InfoFilter: Stdout for INFO/DEBUG only
WarningAndAboveFilter: Stderr for WARNING/ERROR/CRITICAL

Event Loop Monitoring

FastAPI Forge includes an EventLoopMonitor that detects blocking operations in async applications.

Why Monitor the Event Loop?

In async Python applications, blocking operations (like time.sleep(), synchronous I/O, or CPU-intensive tasks) can freeze the entire event loop, causing:

Poor performance and unresponsive APIs
Request timeouts and degraded user experience
Difficulty diagnosing performance issues

The EventLoopMonitor detects these problems by measuring actual vs expected delay of scheduled asyncio.sleep() calls, and captures stack traces to help identify the blocking code.

Basic Usage

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_forge.utils import start_event_loop_monitor, stop_event_loop_monitor

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start monitoring on startup
    monitor = await start_event_loop_monitor(
        check_interval=0.1,      # Check every 100ms
        threshold=0.05,          # Warn if delayed more than 50ms
        capture_stack_trace=True # Capture stack traces on blocking
    )
    yield
    # Stop monitoring on shutdown
    await stop_event_loop_monitor()

app = FastAPI(lifespan=lifespan)

Configuration

Environment variables:

EVENT_LOOP_CHECK_INTERVAL=0.1  # Check interval in seconds (default: 0.1)
EVENT_LOOP_THRESHOLD=0.05      # Blocking threshold in seconds (default: 0.05)
EVENT_LOOP_CAPTURE_STACKS=true # Enable stack trace capture (default: true)

Example Output

When blocking is detected:

{
  "timestamp": "2025-10-27T10:00:00Z",
  "level": "WARNING",
  "logger": "fastapi_forge.utils.blocking_detector",
  "message": "[EVENT_LOOP_BLOCKED] Event loop blocking detected\n\nRunning tasks:\nTask: blocking_endpoint\n  File \"main.py\", line 57, in blocking_endpoint",
  "expected_delay_ms": 100.0,
  "actual_delay_ms": 250.0,
  "excess_delay_ms": 150.0,
  "blocking_ratio": 150.0
}

Performance Impact

The monitor has minimal overhead:

Check interval: 100ms (default)
CPU usage: <0.1% per check
Memory: ~1-2KB per captured stack trace

When to Use

✅ Use in:

Production environments with async workloads
Debugging performance issues
Detecting blocking I/O operations
Monitoring long-running synchronous code

❌ Skip for:

CPU-bound applications (expected blocking)
Single-threaded synchronous apps

GC Monitoring

FastAPI Forge includes a GCMonitor that tracks Python's garbage collection behavior and automatically optimizes GC thresholds based on system resources.

Why Monitor Garbage Collection?

In production environments, understanding GC behavior helps:

Diagnose OOM issues: Correlate memory exhaustion with GC patterns
Detect memory leaks: Track uncollectable objects (circular references)
Optimize GC thresholds: Auto-tune based on memory availability and worker count
Worker-level debugging: Identify problematic workers in multi-worker setups

Basic Usage

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi_forge.monitoring import GCMonitor

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start GC monitoring with auto-calculated threshold
    gc_monitor = GCMonitor(log_interval=60)  # Threshold automatically calculated!
    await gc_monitor.start()
    app.state.gc_monitor = gc_monitor

    yield

    # Stop monitoring on shutdown
    await gc_monitor.stop()

app = FastAPI(lifespan=lifespan)

Environment Variables:

WORKERS=4  # Number of workers (used for threshold calculation)

Dynamic Threshold Calculation

GCMonitor automatically calculates optimal GC thresholds based on:

System memory (detected via os.sysconf or /proc/meminfo)
Number of workers (from WORKERS environment variable)
Memory per worker ratio

Algorithm:

memory_per_worker = total_memory / workers
scale_factor = memory_per_worker / 0.5  # 0.5GB baseline
gen0 = clamp(700 * scale_factor, 400, 1000)
gen1 = max(5, gen0 // 70)
gen2 = max(5, gen0 // 70)

Examples:

# 2GB / 4 workers = 0.5GB per worker
# → (700, 10, 10)  Python default

# 4GB / 4 workers = 1GB per worker
# → (1000, 14, 14)  Less frequent GC (more memory available)

# 2GB / 8 workers = 0.25GB per worker
# → (400, 5, 5)  More frequent GC (memory pressure)

Manual Override

You can still override with custom thresholds if needed:

# Memory-constrained: More frequent GC
gc_monitor = GCMonitor(threshold=(500, 5, 5), log_interval=60)

# CPU-constrained: Less frequent GC
gc_monitor = GCMonitor(threshold=(1000, 20, 20), log_interval=60)

# Auto-calculate (recommended)
gc_monitor = GCMonitor(log_interval=60)  # threshold=None

Example Output

Dynamic threshold calculation (on worker startup):

{
  "timestamp": "2025-10-30T10:00:00Z",
  "level": "INFO",
  "logger": "fastapi_forge.monitoring.gc_monitor",
  "message": "GC threshold calculated dynamically",
  "worker_pid": 12345,
  "total_memory_gb": 2.0,
  "workers": 4,
  "memory_per_worker_gb": 0.5,
  "scale_factor": 1.0,
  "calculated_threshold": [700, 10, 10]
}

Initial GC state:

{
  "timestamp": "2025-10-30T10:00:00Z",
  "level": "INFO",
  "logger": "fastapi_forge.monitoring.gc_monitor",
  "message": "GC initial state",
  "worker_pid": 12345,
  "threshold": [700, 10, 10],
  "gen0_collections": 42,
  "gen0_collected": 1234,
  "gen0_uncollectable": 0
}

Periodic stats (every 60s):

{
  "timestamp": "2025-10-30T10:01:00Z",
  "level": "INFO",
  "logger": "fastapi_forge.monitoring.gc_monitor",
  "message": "GC stats snapshot",
  "worker_pid": 12345,
  "threshold": [500, 5, 5],
  "gen0_collections": 123,
  "gen0_collected": 4567,
  "gen0_uncollectable": 0,
  "gen1_collections": 12,
  "gen1_collected": 234,
  "gen2_collections": 1,
  "gen2_collected": 56
}

Observability Platform Integration

Datadog:

# Filter by worker
@worker_pid:12345 "GC stats snapshot"

# Memory leak detection
@gen0_uncollectable:>0 OR @gen1_uncollectable:>0

# Create metrics from logs
gen0_collections, gen1_collections, gen2_collections

# Alert on uncollectable objects
@gen2_uncollectable:>0

ELK/Splunk/Grafana:

Index on worker_pid for per-worker analysis
Create dashboards tracking GC frequency over time
Alert on uncollectable > 0 for potential memory leaks

Interpreting Results

Uncollectable objects:

uncollectable > 0 indicates circular references Python couldn't break
Potential memory leak requiring code review
Check for __del__ methods creating circular refs

Collection frequency:

High gen0_collections: Normal for busy workers
Low gen2_collections: Objects not surviving long enough
No gen2_collections for hours: May indicate threshold too high

Worker comparison:

Compare stats across workers with @worker_pid
Uneven GC patterns may indicate load imbalance
One worker with high uncollectable → investigate that worker's requests

LangChain Integration

FastAPI Forge provides utilities for building robust LangChain applications with automatic fallback mechanisms.

Features

Automatic Fallback: Seamlessly switch to backup chains when primary fails
Streaming Support: Full support for sync/async streaming operations
Switch Markers: Optional markers to notify clients of fallback activation
Type Safety: Fully typed with mypy support
Logging: Built-in logging for debugging and monitoring

Quick Start

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.runnables import RunnableLambda
from fastapi_forge.langchain import with_runnable_fallback

app = FastAPI()

# Create chains with fallback
def create_chain():
    # Primary chain (may fail)
    primary = RunnableLambda(lambda x: process_with_expensive_llm(x))

    # Fallback chain (more reliable)
    fallback = RunnableLambda(lambda x: process_with_cheaper_llm(x))

    return with_runnable_fallback(
        primary,
        fallback,
        switch_marker={"type": "fallback_activated"}
    )

@app.post("/process")
async def process_endpoint(data: dict):
    """Process with automatic fallback."""
    chain = create_chain()
    result = await chain.ainvoke(data)
    return {"result": result}

@app.post("/stream")
async def stream_endpoint(data: dict):
    """Stream with fallback support."""
    chain = create_chain()

    async def generate():
        async for chunk in chain.astream(data):
            if isinstance(chunk, dict) and chunk.get("type") == "fallback_activated":
                yield f"data: [FALLBACK]\n\n"
            else:
                yield f"data: {chunk}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Common Use Cases

1. LLM Fallback: Use cheaper model when primary fails

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary = ChatAnthropic(model="claude-3-opus") | output_parser
fallback = ChatOpenAI(model="gpt-3.5-turbo") | output_parser

chain = with_runnable_fallback(primary, fallback)

2. API Resilience: Fall back to cached responses

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_cached_response(key: str) -> str:
    return cached_responses.get(key, "Default response")

primary = RunnableLambda(lambda x: call_external_api(x["query"]))
fallback = RunnableLambda(lambda x: get_cached_response(x["query"]))

chain = with_runnable_fallback(primary, fallback)

3. Graceful Degradation: Reduced functionality when full features fail

primary = RunnableLambda(generate_detailed_response)
fallback = RunnableLambda(lambda x: "Service temporarily degraded. Please try again.")

chain = with_runnable_fallback(primary, fallback)

Documentation

For comprehensive documentation, see:

LangChain Guidelines - Complete guide with best practices
Example: LangChain Fallback - Working code examples

📁 Project Structure

fastapi-forge/
├── src/fastapi_forge/
│   ├── logging/              # Production logging
│   │   ├── config.py         # Configuration
│   │   ├── formatters.py     # JSONFormatter
│   │   └── filters.py        # Log filters
│   ├── monitoring/           # Production monitoring
│   │   └── gc_monitor.py     # GC monitoring
│   ├── utils/                # Utilities
│   │   └── blocking_detector.py  # Event loop monitoring
│   ├── langchain/            # LangChain integration
│   │   └── fallback.py       # Runnable fallback utilities
│   ├── templates/            # App templates (coming soon)
│   └── middleware/           # Production middleware (coming soon)
├── examples/
│   ├── 01_basic_fastapi/
│   ├── 02_with_ddtrace/
│   ├── 03_with_blocking_monitor/
│   └── 04_langchain_fallback/
└── docs/

🧪 Examples

Check the examples/ directory for complete working examples:

01_basic_fastapi: Minimal FastAPI app with logging
02_with_ddtrace: Production setup with Datadog APM
03_with_blocking_monitor: Event loop monitoring and blocking detection
04_langchain_fallback: LangChain fallback utilities with streaming support

🤝 Contributing

Contributions are welcome! This is an open-source project.

Fork the repository
Create your feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

📝 License

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

🙏 Acknowledgments

Built with experience from production FastAPI applications
Inspired by battle-tested patterns from the Python community
Datadog integration based on real-world APM requirements

🔗 Links

Documentation: GitHub Wiki
Issues: GitHub Issues
PyPI: fastapi-forge (coming soon)

🗺️ Roadmap

⚡ Performance Optimizations

Logging Optimization (OOM Prevention)

FastAPI Forge's logging system has been optimized to prevent Out of Memory (OOM) errors in high-load production environments.

Problem

In high-traffic scenarios (1000+ logs/second), the logging system consumed excessive memory:

Redundant String Formatting: Each filter and formatter called record.getMessage(), causing repeated msg % args operations
- 3 filters + 1 formatter = 4 calls per log
- 1000 logs/sec × 4 = 4000 string creations/sec
Unlimited Message Size: Large messages (50KB+) triggered _progressive_truncation
- Loop with up to 20 JSON serializations per log
- 1000 logs/sec × 20 = 20,000 serializations/sec
Memory Usage: ~150MB/s per worker, 4 workers = 600MB/s memory pressure → OOM

Solution

1. getMessage() Caching

# filters.py & formatters.py
if not hasattr(record, '_cached_message'):
    record._cached_message = record.getMessage()

Impact:

4 calls → 1 call per log (75% reduction)
LogRecord instances are shared across all handlers/filters, maximizing cache effectiveness

2. Proactive Message Size Limiting

# formatters.py _build_core_structure()
message = record._cached_message
if len(message) > 10000:  # 10KB limit
    message = message[:10000] + "...[truncated]"

Impact:

10KB message + 5KB extras = 15KB (under limit)
_progressive_truncation calls reduced by 99%
Most logs processed with single serialization

3. Batch Field Removal (No Loop Serialization)

# Before (Problem):
for key in non_core_keys:  # 20 fields
    del log_entry[key]
    json.dumps(...)  # Serialized 20 times!

# After (Solution):
keys_to_remove = [k for k in ... if k not in preserve]
for key in keys_to_remove:
    del log_entry[key]
json.dumps(...)  # Serialized only once!

Impact:

20 serializations → 3 serializations (85% reduction)

Performance Impact

Metric	Before	After	Improvement
getMessage() calls	4/log	1/log	75% ↓
_progressive_truncation calls	Frequent	< 1%	99% ↓
JSON serializations (truncation)	Up to 20	3	85% ↓
Memory usage (1000 logs/sec, 4 workers)	600MB/s	100MB/s	83% ↓

Implementation Details

Modified files:

src/fastapi_forge/logging/filters.py
- HealthCheckFilter: getMessage() caching
- LangfuseFilter: getMessage() caching
- LangchainFilter: getMessage() caching
src/fastapi_forge/logging/formatters.py
- JSONFormatter._build_core_structure(): caching + 10KB limit
- JSONFormatter._progressive_truncation(): batch field removal
- DatadogJSONFormatter: inherits optimizations from parent

Key insight: Python logging's msg % args is lazy-evaluated only when getMessage() is called. By caching the result on the shared LogRecord instance, we eliminate redundant formatting across the entire logging pipeline.

Reference:

Commit: d3a60c0 (logging optimization)
Python logging internals: getMessage() performs string formatting each time it's called

Made with ❤️ for the FastAPI community

Name		Name	Last commit message	Last commit date
Latest commit History 15 Commits
docs		docs
examples		examples
src/fastapi_forge		src/fastapi_forge
tests		tests
.gitignore		.gitignore
AGENTS.md		AGENTS.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml

Folders and files

Latest commit

History

Repository files navigation

FastAPI Forge 🔨

✨ Features

🚀 Quick Start

Installation

Basic Usage

📚 Documentation

Logging

Key Features

Quick Start - Generic JSON

Datadog APM Integration

Advanced Usage

Filters

Event Loop Monitoring

Why Monitor the Event Loop?

Basic Usage

Configuration

Example Output

Performance Impact

When to Use

GC Monitoring

Why Monitor Garbage Collection?

Basic Usage

Dynamic Threshold Calculation

Manual Override

Example Output

Observability Platform Integration

Interpreting Results

LangChain Integration

Features

Quick Start

Common Use Cases

Documentation

📁 Project Structure

🧪 Examples

🤝 Contributing

📝 License

🙏 Acknowledgments

🔗 Links

🗺️ Roadmap

⚡ Performance Optimizations

Logging Optimization (OOM Prevention)

Problem

Solution

1. getMessage() Caching

2. Proactive Message Size Limiting

3. Batch Field Removal (No Loop Serialization)

Performance Impact

Implementation Details

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages