lilith-service-fastapi-bootstrap

Base utilities and common patterns for building FastAPI-based services.

Overview

lilith-service-fastapi-bootstrap provides a standardized foundation for FastAPI services, including:

• Configuration Management: Environment-based settings with validation
• Health Checks: Flexible health check aggregation system
• Lifespan Management: Startup/shutdown task orchestration
• Logging: Structured logging with JSON and human-readable formats
• Middleware: Pre-configured CORS and extensible middleware support
• Exception Hierarchy: Common exception types for ML services

Installation

# Core package
pip install lilith-service-fastapi-bootstrap

# With Redis support
pip install lilith-service-fastapi-bootstrap[redis]

# With database support
pip install lilith-service-fastapi-bootstrap[database]

# All optional features
pip install lilith-service-fastapi-bootstrap[all]

# Development dependencies (testing, etc.)
pip install lilith-service-fastapi-bootstrap[dev]

Quick Start

from lilith_ml_service_base import (
    create_ml_service,
    BaseServiceSettings,
    LifespanManager,
    HealthChecker,
    BaseHealthResponse,
)

# 1. Configure settings
class MySettings(BaseServiceSettings):
    model_path: str = "/models/default"

settings = MySettings(service_name="embedding-service")

# 2. Setup lifespan hooks
lifespan = LifespanManager()

@lifespan.on_startup
async def load_model():
    model = load_my_model(settings.model_path)
    lifespan.set_state("model", model)

@lifespan.on_shutdown
async def cleanup():
    model = lifespan.get_state("model")
    if model:
        model.unload()

# 3. Setup health checks
health = HealthChecker()

@health.check("model")
async def check_model():
    model = lifespan.get_state("model")
    return model is not None

# 4. Create app
app = create_ml_service(
    title="Embedding Service",
    description="Generate text embeddings",
    version="1.0.0",
    settings=settings,
    lifespan_manager=lifespan,
    health_checker=health,
)

# 5. Define health endpoint
class MyHealthResponse(BaseHealthResponse):
    model_loaded: bool = False

@app.get("/health", response_model=MyHealthResponse)
async def health_check():
    results = await health.run_checks()
    return MyHealthResponse(
        status=health.get_status(results),
        version=settings.service_version,
        model_loaded=results.get("model", False),
    )

# 6. Define your routes
@app.post("/embed")
async def embed(text: str):
    model = app.state.lifespan_manager.get_state("model")
    return {"embedding": model.embed(text)}

Automatic Dependency Startup (v2.0+)

New in v2.0: Services can automatically start their dependencies before the application starts, with idempotent behavior that gracefully accepts already-running services.

When to Use

Use automatic dependency startup when your service depends on other services (databases, Redis, other APIs) that may or may not already be running. This feature:

• Idempotent: Detects already-running services via health checks, PID files, Docker containers, or TCP ports
• Deduplicates: Multiple services can start the same dependency without conflicts
• Orchestrates: Starts dependencies in topological order (respecting dependency chains)
• Validates: Waits for dependencies to be healthy before continuing

Requirements

Install the lilith-service-addresses package:

pip install lilith-service-fastapi-bootstrap[service-addresses]
# Or separately:
pip install lilith-service-addresses

Basic Usage

from lilith_service_fastapi_bootstrap import (
    create_service,
    DependencyStartupConfig,
)

app = await create_service(
    title="Analytics Service",
    description="Usage metrics API",
    dependencies=DependencyStartupConfig(
        feature="analytics",  # Must match feature ID in services.yaml
    ),
)

Configuration Options

DependencyStartupConfig(
    feature="analytics",          # Feature name from services.yaml (required)
    auto_start=True,              # Enable automatic dependency startup (default: True, set False to disable)
    wait_for_health=True,         # Wait for dependencies to be healthy (default: True)
    health_check_timeout=60000,   # Health check timeout per dependency in ms (default: 60000)
    skip_dependencies=["analytics.redis"],  # Skip specific dependencies (optional)
    on_progress=lambda event: print(f"[{event.service}] {event.phase}: {event.message}"),
)

Progress Monitoring

Track dependency startup progress with callbacks:

from lilith_service_fastapi_bootstrap import DependencyStartupEvent, DependencyPhase

def on_progress(event: DependencyStartupEvent):
    print(f"[{event.service}] {event.phase}: {event.message}")
    if event.phase == DependencyPhase.READY:
        print(f"  ✓ {event.service} is ready (metadata: {event.metadata})")

app = await create_service(
    title="My Service",
    description="My API",
    dependencies=DependencyStartupConfig(
        feature="my-feature",
        on_progress=on_progress,
    ),
)

Default Progress Logger Helper

The create_default_progress_logger() helper provides standardized logging without boilerplate:

from lilith_service_fastapi_bootstrap import (
    create_service,
    DependencyStartupConfig,
    create_default_progress_logger,
)

app = await create_service(
    title="Analytics API",
    dependencies=DependencyStartupConfig(
        feature="analytics",
        on_progress=create_default_progress_logger("Analytics"),
    ),
)

Type-Safe Phase Constants

Use DependencyPhases constants for cleaner, safer code:

from lilith_service_fastapi_bootstrap import (
    DependencyStartupConfig,
    DependencyPhases,
)

def on_progress(event):
    if event.phase == DependencyPhases.READY:
        print(f"✓ {event.service} ready")
    elif event.phase == DependencyPhases.ERROR:
        print(f"✗ {event.service} failed: {event.message}")

app = await create_service(
    title="ML Service",
    dependencies=DependencyStartupConfig(
        feature="embeddings",
        on_progress=on_progress,
    ),
)

How It Works

Multi-Strategy Detection:

The system uses multiple strategies to determine if a service is already running:

1. PID Files: Check if we previously started this service
2. Health Endpoints: Probe HTTP health endpoints (e.g., /health)
3. Docker Containers: Query Docker for running containers
4. TCP Ports: Test if ports are listening

If a dependency is already running, it's accepted and startup continues. If not, the system starts it and waits for it to become healthy.

Startup Behavior:

# If PostgreSQL and Redis are NOT running:
# 1. Start PostgreSQL → wait for health check to pass
# 2. Start Redis → wait for health check to pass
# 3. Create FastAPI app
# 4. Start listening

# If PostgreSQL and Redis ARE already running:
# 1. Detect via health check → accept as valid
# 2. Create FastAPI app immediately
# 3. Start listening

Environment Variables

Configure registry paths via environment variables:

export LILITH_SERVICES_PATH="codebase/features"       # Path to feature directories
export LILITH_PORTS_PATH="infrastructure/ports.yaml"  # Path to ports.yaml
export LILITH_STRICT_VALIDATION="false"               # Recommended for dev environment

Complete Example

from lilith_service_fastapi_bootstrap import (
    create_service,
    BaseServiceSettings,
    LifespanManager,
    DependencyStartupConfig,
    DependencyStartupEvent,
)

# Configure settings
settings = BaseServiceSettings(service_name="analytics-service")

# Setup lifespan
lifespan = LifespanManager()

@lifespan.on_startup
async def init_resources():
    # Dependencies are already running at this point
    print("Initializing resources...")

# Create app with automatic dependency startup
app = await create_service(
    title="Analytics Service",
    description="Usage metrics API",
    version="1.0.0",
    settings=settings,
    lifespan_manager=lifespan,
    dependencies=DependencyStartupConfig(
        feature="analytics",
        on_progress=lambda event: print(
            f"[{event.service}] {event.phase}: {event.message}"
        ),
    ),
)

# At this point:
# - All dependencies are running and healthy
# - Lifespan startup tasks have executed
# - App is ready to handle requests

Integration with services.yaml

Dependencies are declared in services.yaml:

# codebase/features/analytics/services.yaml
feature: analytics
dependencies:
  - analytics.postgresql
  - analytics.redis

services:
  - id: analytics.api
    type: nestjs
    port: 3012
    health: http://localhost:3012/health

  - id: analytics.postgresql
    type: docker
    container: analytics-postgresql
    port: 5433

  - id: analytics.redis
    type: docker
    container: analytics-redis
    port: 6380

The system reads this configuration and automatically starts PostgreSQL and Redis before starting the FastAPI application.

Error Handling

If dependency startup fails, the application will not start:

try:
    app = await create_service(
        title="My Service",
        dependencies=DependencyStartupConfig(
            feature="my-feature",
        ),
    )
except RuntimeError as e:
    # e.g., "Dependency startup failed: PostgreSQL connection timeout"
    print(f"Failed to start: {e}")

Features

BaseServiceSettings

Pydantic-based configuration with environment variable support:

from lilith_ml_service_base import BaseServiceSettings

class MySettings(BaseServiceSettings):
    custom_field: str = "default"

# Loads from environment: ML_SERVICE_SERVICE_NAME, ML_SERVICE_DEBUG, etc.
settings = MySettings(service_name="my-service")

Built-in fields:

• service_name (required)
• service_version (default: "0.1.0")
• debug (default: False)
• log_level (default: "INFO", validates against DEBUG/INFO/WARNING/ERROR/CRITICAL)
• redis_url (optional)
• database_url (optional)
• cors_origins (default: ["http://localhost:3000"], accepts list or comma-separated string)

Environment variables (prefix: ML_SERVICE_):

export ML_SERVICE_SERVICE_NAME="embedding-service"
export ML_SERVICE_DEBUG="true"
export ML_SERVICE_LOG_LEVEL="DEBUG"
export ML_SERVICE_CORS_ORIGINS="https://app.example.com,https://admin.example.com"

LifespanManager

Decorator-based startup/shutdown hooks:

lifespan = LifespanManager()

@lifespan.on_startup
async def init():
    lifespan.set_state("db", await connect_db())

@lifespan.on_shutdown
async def cleanup():
    await lifespan.get_state("db").disconnect()

HealthChecker

Concurrent health check aggregation:

health = HealthChecker()

@health.check("redis")
async def check_redis():
    return await redis.ping()

@health.check("db")
async def check_db():
    return await db.health_check()

# Run all checks concurrently
results = await health.run_checks()  # {"redis": True, "db": False}
status = health.get_status(results)  # "degraded"

RedisClient (optional)

from lilith_ml_service_base.cache import RedisClient

cache = RedisClient("redis://localhost:6379")
await cache.connect()
await cache.set("key", "value", ttl=3600)
await cache.set_json("data", {"foo": "bar"})

AsyncDatabaseManager (optional)

from lilith_ml_service_base.database import AsyncDatabaseManager

db = AsyncDatabaseManager("postgresql+asyncpg://...")
await db.connect()
await db.create_tables(Base)

async with db.session() as session:
    result = await session.execute(select(User))

API Reference

create_ml_service()

Factory function that creates a configured FastAPI app:

app = create_ml_service(
    title="Service Name",
    description="Service description",
    version="1.0.0",
    settings=settings,           # Optional: BaseServiceSettings
    lifespan_manager=lifespan,   # Optional: LifespanManager
    health_checker=health,       # Optional: HealthChecker
    apply_default_cors=True,     # Optional: Apply CORS middleware
)

Returns: Configured FastAPI instance with:

• CORS middleware applied
• Structured logging configured
• Settings/lifespan/health stored in app.state

Exceptions

from lilith_ml_service_base import (
    ServiceError,          # Base exception
    ConfigurationError,    # Invalid configuration
    HealthCheckError,      # Health check failure
    ResourceNotReadyError, # Resource not initialized
    ConnectionError,       # External service connection failure
)

Development

Running Tests

# Install with dev dependencies
pip install -e ".[dev]"

# Run all tests
pytest

# Run with coverage
pytest --cov=lilith_service_fastapi_bootstrap --cov-report=html --cov-report=term

# Run specific test file
pytest tests/test_health.py -v

# Run specific test
pytest tests/test_health.py::TestHealthChecker::test_run_checks_passing -v

Test Coverage

The test suite covers:

• Configuration: Settings validation, environment variable loading, defaults
• Health Checks: Check registration, execution, status determination, error handling
• Lifespan: Startup/shutdown tasks, execution order, state management, error scenarios
• App Factory: App creation, middleware application, state injection

Project Structure

lilith-service-fastapi-bootstrap/
├── src/
│   └── lilith_service_fastapi_bootstrap/
│       ├── __init__.py          # Public API exports
│       ├── app.py               # Application factory
│       ├── config.py            # Configuration models
│       ├── health.py            # Health check system
│       ├── lifespan.py          # Lifecycle management
│       ├── logging.py           # Logging configuration
│       ├── middleware.py        # Middleware utilities
│       └── exceptions.py        # Exception hierarchy
├── tests/
│   ├── conftest.py              # Pytest fixtures
│   ├── test_app.py              # App factory tests
│   ├── test_config.py           # Configuration tests
│   ├── test_health.py           # Health check tests
│   └── test_lifespan.py         # Lifespan tests
├── pyproject.toml               # Project metadata & dependencies
└── README.md                    # This file

Design Principles

1. Configuration as Code: Type-safe, validated configuration with environment variable support
2. Dependency Injection: Pass dependencies explicitly rather than global state
3. Async First: All I/O operations use async/await for better performance
4. Fail Fast: Startup tasks that fail prevent the service from starting
5. Observable: Structured logging and health checks for production monitoring
6. Composable: Mix and match components as needed

License

MIT