app

FastAPI application factory with lifespan management and CORS configuration.

_SERVICE_VERSION module-attribute

_SERVICE_VERSION = version('horde_model_reference')

_API_DESCRIPTION module-attribute

_API_DESCRIPTION = "\nThe **Horde Model Reference API** is the authoritative source of AI model metadata for the\n[AI-Horde](https://aihorde.net) ecosystem. It serves the curated lists of image, text, and\nutility models (CLIP, ControlNet, ESRGAN, …) that workers download and that clients display.\n\n### Who uses this API\n\n- **Workers & clients** read model references - either directly over HTTP or via the\n  `horde-model-reference` Python library running in REPLICA mode (which calls this same API,\n  falling back to GitHub if the PRIMARY is unreachable).\n- **The AI-Horde backend** runs this service in PRIMARY mode at\n  [`models.aihorde.net`](https://models.aihorde.net/api/docs) as the canonical source.\n\n### Two API versions\n\n- **v2** (`/model_references/v2`) - the current format, with search, per-model retrieval,\n  statistics, and the full text-model grouping toolkit. Prefer this for new integrations.\n- **v1** (`/model_references/v1`) - the legacy GitHub-compatible format, retained unchanged for\n  backward compatibility with existing AI-Horde workers.\n\nBoth versions are readable regardless of deployment configuration. **Reads are open; writes are\nnot.** Write operations require a PRIMARY deployment and a valid `apikey`, and they are not\napplied immediately - they enter a [pending queue](https://models.aihorde.net/api/docs) for\ntwo-person review (propose -> approve -> apply).\n\n### Discovering capabilities\n\nCall [`GET /replicate_mode`](#operations-default-replicate_mode_replicate_mode_get) on startup to\nlearn whether an instance is writable and which canonical format it serves.\n\nFull documentation, tutorials, and guides: <https://github.com/Haidra-Org/horde-model-reference>\n"

_OPENAPI_TAGS module-attribute

_OPENAPI_TAGS = [
    {
        "name": "v2",
        "description": "Current model-reference format: reads, CRUD, per-model retrieval, and metadata.",
    },
    {
        "name": "v1",
        "description": "Legacy GitHub-compatible format, retained unchanged for existing AI-Horde workers.",
    },
    {
        "name": "search",
        "description": "Filter, sort, and paginate models within a category or across all categories.",
    },
    {
        "name": "statistics",
        "description": "Aggregated per-category counts, baseline/tag distributions, and download statistics.",
    },
    {
        "name": "deletion-risk",
        "description": "Live-usage-informed risk analysis identifying models that are candidates for removal.",
    },
    {
        "name": "text_utils",
        "description": "Text-generation grouping toolkit: name parsing/composition, groups, aliases, families, and naming schemas.",
    },
    {
        "name": "pending_queue",
        "description": "Propose -> approve -> apply workflow for model changes on PRIMARY deployments.",
    },
    {
        "name": "audit",
        "description": "Read-only history of pending-queue batches and their net effect.",
    },
    {
        "name": "metadata",
        "description": "Per-category last-updated timestamps for change detection by REPLICA clients.",
    },
    {
        "name": "user",
        "description": "Authenticated user identity and pending-queue roles (requestor/approver).",
    },
]

app module-attribute

app = FastAPI(
    root_path="/api",
    lifespan=lifespan,
    title="Horde Model Reference API",
    summary="Authoritative AI model metadata for the AI-Horde ecosystem.",
    description=_API_DESCRIPTION,
    version=_SERVICE_VERSION,
    openapi_tags=_OPENAPI_TAGS,
    contact={
        "name": "Haidra-Org / AI-Horde",
        "url": "https://github.com/Haidra-Org/horde-model-reference",
    },
    license_info={
        "name": "AGPL-3.0",
        "url": "https://www.gnu.org/licenses/agpl-3.0.en.html",
    },
    servers=[
        {
            "url": "https://models.aihorde.net/api",
            "description": "Public PRIMARY deployment",
        },
        {
            "url": "http://localhost:19800/api",
            "description": "Local development server",
        },
    ],
)

AIHordeStatus

Bases: BaseModel

Status of the external AI Horde API connection.

Source code in src/horde_model_reference/service/app.py

class AIHordeStatus(BaseModel):
    """Status of the external AI Horde API connection."""

    degraded: bool
    consecutive_failures: int
    seconds_until_retry: float | None

degraded instance-attribute

degraded: bool

consecutive_failures instance-attribute

consecutive_failures: int

seconds_until_retry instance-attribute

seconds_until_retry: float | None

HeartbeatResponse

Bases: BaseModel

Enhanced heartbeat response with external service status.

Source code in src/horde_model_reference/service/app.py

class HeartbeatResponse(BaseModel):
    """Enhanced heartbeat response with external service status."""

    status: str
    ai_horde: AIHordeStatus

status instance-attribute

status: str

ai_horde instance-attribute

ai_horde: AIHordeStatus

lifespan async

lifespan(app: FastAPI) -> AsyncGenerator[None]

Manage application lifespan events.

Starts background cache hydration on startup and stops it on shutdown.

Source code in src/horde_model_reference/service/app.py

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    """Manage application lifespan events.

    Starts background cache hydration on startup and stops it on shutdown.
    """
    # Startup
    if horde_model_reference_settings.cache_hydration_enabled:
        from horde_model_reference.analytics.cache_hydrator import get_cache_hydrator

        hydrator = get_cache_hydrator()
        logger.info("Starting cache hydration on application startup...")
        await hydrator.start()

    yield

    # Shutdown
    from horde_model_reference.service.shared import httpx_client

    await httpx_client.aclose()

    if horde_model_reference_settings.cache_hydration_enabled:
        from horde_model_reference.analytics.cache_hydrator import get_cache_hydrator

        hydrator = get_cache_hydrator()
        logger.info("Stopping cache hydration on application shutdown...")
        await hydrator.stop()

read_root async

read_root() -> ContainsMessage

Return a welcome message pointing to the interactive documentation.

Source code in src/horde_model_reference/service/app.py

@app.get("/", summary="API landing message", tags=["default"])
async def read_root() -> ContainsMessage:
    """Return a welcome message pointing to the interactive documentation."""
    return ContainsMessage(
        message="Welcome to the Horde Model Reference API. See `/api/docs` for interactive documentation.",
    )

heartbeat async

heartbeat() -> HeartbeatResponse

Heartbeat endpoint to check the service status.

Returns overall service status and the state of the external AI Horde API connection. When the AI Horde API is unreachable, ai_horde.degraded is True and ai_horde.seconds_until_retry indicates when the next probe request will be attempted.

Source code in src/horde_model_reference/service/app.py

@app.get("/heartbeat", summary="Service health check", tags=["default"])
async def heartbeat() -> HeartbeatResponse:
    """Heartbeat endpoint to check the service status.

    Returns overall service status and the state of the external AI Horde API
    connection. When the AI Horde API is unreachable, ``ai_horde.degraded`` is
    ``True`` and ``ai_horde.seconds_until_retry`` indicates when the next probe
    request will be attempted.
    """
    cb_status = horde_api_circuit_breaker.get_status_dict()
    return HeartbeatResponse(
        status="ok",
        ai_horde=AIHordeStatus(
            degraded=cb_status["degraded"],
            consecutive_failures=cb_status["consecutive_failures"],
            seconds_until_retry=cb_status["seconds_until_retry"],
        ),
    )

replicate_mode async

replicate_mode() -> BackendInfo

Get backend configuration and capabilities.

Returns information about the backend's replication mode, canonical format, and whether write operations are supported.

Clients should use this endpoint on startup to determine: - Whether the backend supports write operations (writable=True) - Which API version to use for CRUD operations (based on canonical_format)

Note: For backward compatibility, this endpoint path is retained but now returns a richer BackendInfo response instead of just the ReplicateMode.

Source code in src/horde_model_reference/service/app.py

@app.get("/replicate_mode", summary="Backend capabilities probe", tags=["default"])
async def replicate_mode() -> BackendInfo:
    """Get backend configuration and capabilities.

    Returns information about the backend's replication mode, canonical format,
    and whether write operations are supported.

    Clients should use this endpoint on startup to determine:
    - Whether the backend supports write operations (writable=True)
    - Which API version to use for CRUD operations (based on canonical_format)

    Note: For backward compatibility, this endpoint path is retained but now
    returns a richer BackendInfo response instead of just the ReplicateMode.
    """
    from horde_model_reference import horde_model_reference_settings

    # Map the string setting to the enum
    canonical_format = horde_model_reference_settings.canonical_format

    return BackendInfo(
        replicate_mode=horde_model_reference_settings.replicate_mode,
        canonical_format=canonical_format,
        writable=horde_model_reference_settings.replicate_mode == ReplicateMode.PRIMARY,
    )