The complete FastAPI control plane

The FastAPI control plane is the single front door to twenty services. It builds a service container at startup, registers six health checks as named callables, and exposes seven controllers worth of routes. The shape is lifted 1:1 from the .NET reference. The runtime is Python.

main.py
python
from dataclasses import dataclass


@dataclass
class ServiceContainer:
    db: DbService
    storage: MinioService
    kafka: KafkaService
    ge: GEValidationService
    batch: BatchService
    streaming: StreamingService
    atlas: AtlasService
    mlflow: MLflowService
    ci: CIService
    monitoring: MonitoringService


def build_container() -> ServiceContainer:
    db_opts = DatabaseOptions()
    airflow_opts = AirflowOptions()
    # ...eight more option classes load here

    mysql_hc = MySqlHealthCheck(db_opts)
    postgres_hc = PostgresHealthCheck(db_opts)
    # ...four more health checks register here

    monitoring = MonitoringService({
        'mysql': mysql_hc.check,
        'postgres': postgres_hc.check,
        'minio': MinioHealthCheck(minio_opts).check,
        'kafka': KafkaHealthCheck(kafka_opts).check,
        'airflow': AirflowHealthCheck(airflow_opts).check,
        'mlflow': MLflowHealthCheck(mlflow_opts).check,
    })

    return ServiceContainer(
        db=DbService(db_opts),
        storage=MinioService(minio_opts),
        kafka=KafkaService(kafka_opts),
        ge=GEValidationService(ge_opts),
        batch=BatchService(airflow_opts),
        streaming=StreamingService(airflow_opts),
        atlas=AtlasService(atlas_opts),
        mlflow=MLflowService(mlflow_opts),
        ci=CIService(github_opts),
        monitoring=monitoring,
    )

build_container mirrors builder.Services.AddSingleton<...> in the .NET Program.cs. Ten services, six health checks, one monitoring service that carries the named-callable registry. All wired at startup, all injected into routes at request time.

router.py
python
@router.post('/api/batch/ingest', response_model=BatchResponse)
async def batch_ingest(req: BatchRequest, request: Request) -> BatchResponse:
    container = request.app.state.container
    try:
        rows = await container.db.read_mysql_table(req.source_table, req.limit)
    except Exception:
        rows = []  # Graceful: smoke test runs with no MySQL

    body = json.dumps(rows, default=str).encode('utf-8')
    object_key = f'{req.destination_prefix or req.source_table}/{timestamp()}.json'

    try:
        await container.storage.upload_raw(object_key, io.BytesIO(body))
    except Exception:
        pass  # Graceful

    run_id = await container.batch.trigger_batch() if req.trigger_airflow else f'stub_{timestamp()}'
    return BatchResponse(object_key=object_key, run_id=run_id, status='accepted')

The ported Controllers become a single router. Each endpoint reads container from request.app.state and calls the service. Errors degrade to stub responses so the API stays alive when a single downstream dep is down.

Because the control plane is a surface for many consumers. A dashboard, a notebook, the smoke test, the on-call runbook. Returning stubs with accepted status keeps the contract alive. The health endpoint tells operators what is really down. Clients that specifically need MinIO check /health first. Others see the accepted stub and move on.

Validation checklist: Verify the complete control plane

Loading practice…

Quiz: Quiz

Loading practice…

Checkpoint: Final checkpoint: you own the enterprise platform

Loading practice…