generated from coulomb/repo-seed
54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
|
|
import core_hub.models # noqa: F401
|
|
from core_hub import __version__
|
|
from core_hub.api.v2 import router as api_v2_router
|
|
from core_hub.config import get_settings
|
|
from core_hub.console import router as console_router
|
|
from core_hub.db import Base, get_engine
|
|
from core_hub.schemas import HealthResponse, ReadinessResponse
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
|
settings = get_settings()
|
|
if settings.auto_create_tables:
|
|
engine = get_engine()
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
yield
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
settings = get_settings()
|
|
app = FastAPI(
|
|
title="Core Hub API",
|
|
version=__version__,
|
|
description="Third-generation production interaction framework API.",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
@app.get("/healthz", response_model=HealthResponse, tags=["system"])
|
|
def healthz() -> HealthResponse:
|
|
return HealthResponse(service="core-hub", status="ok", version=__version__)
|
|
|
|
@app.get("/readyz", response_model=ReadinessResponse, tags=["system"])
|
|
def readyz() -> ReadinessResponse:
|
|
checks = {
|
|
"configuration": "ok",
|
|
"database_url": "configured" if settings.database_url else "missing",
|
|
"api_auth": "configured" if settings.auth_configured else "not_configured",
|
|
}
|
|
status = "ok" if checks["database_url"] == "configured" else "degraded"
|
|
return ReadinessResponse(service="core-hub", status=status, checks=checks)
|
|
|
|
app.include_router(api_v2_router)
|
|
app.include_router(console_router)
|
|
return app
|
|
|
|
|
|
app = create_app()
|