lilith-ml-service-base

⚠️ DEPRECATED: This package has been superseded by @service/fastapi-bootstrap v2.1.0. All functionality from this package is now available in lilith-fastapi-service-base>=2.1.0 with additional features. Migration: Replace lilith-ml-service-base with lilith-fastapi-service-base>=2.1.0 in your dependencies and use ML_SERVICE_PRESET for ML services. See: fastapi-bootstrap Migration Guide

Base utilities and common patterns for building FastAPI-based ML microservices.

Overview

lilith-ml-service-base provides a standardized foundation for ML 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-ml-service-base

# With Redis support
pip install lilith-ml-service-base[redis]

# With database support
pip install lilith-ml-service-base[database]

# All optional features
pip install lilith-ml-service-base[all]

# Development dependencies (testing, etc.)
pip install lilith-ml-service-base[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)}

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_ml_service_base --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-ml-service-base/
├── src/
│   └── lilith_ml_service_base/
│       ├── __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

---

Migration from ml-service-base

This package is deprecated. Migrate to @service/fastapi-bootstrap v2.1.0:

1. Update Dependencies

Before:

dependencies = [
    "lilith-ml-service-base>=0.2.0",
]

After:

dependencies = [
    "lilith-fastapi-service-base>=2.1.0",
]

2. Update Imports

Before:

from lilith_ml_service_base import (
    create_ml_service,
    BaseServiceSettings,
    LifespanManager,
)

After:

from lilith_fastapi_service_base import (
    create_service,  # Note: create_ml_service → create_service
    BaseServiceSettings,
    LifespanManager,
    ML_SERVICE_PRESET,  # New: preset configuration
)

3. Update App Creation

Before:

app = create_ml_service(
    title="My ML Service",
    description="ML service",
    version="0.1.0",
    settings=settings,
    lifespan_manager=lifespan,
)

After (async with ML preset):

async def create_app():
    return await create_service(
        title="My ML Service",
        description="ML service",
        version="0.1.0",
        settings=settings,
        lifespan_manager=lifespan,
        **ML_SERVICE_PRESET,  # Adds: flexible validation, 100MB body limit, GPU monitoring
    )

# Module-level
import asyncio
app = asyncio.run(create_app())

4. New Features Available

By migrating, you gain access to:

• Dependency Startup (v2.0): Auto-start service dependencies from service registry
• Security Headers (v2.1): HSTS, X-Frame-Options, CSP
• Swagger/OpenAPI (v2.1): Auto-configured API documentation
• Body Size Limits (v2.1): Prevent memory exhaustion
• CORS Discovery (v2.1): Auto-discover from service registry
• Bootstrap Presets (v2.1): ML_SERVICE_PRESET, API_PRESET, etc.

Support

All services in the Lilith platform have been migrated. For assistance, see:

• @service/fastapi-bootstrap documentation
• Migration examples in @applications/@ml/*