Skip to content

app

FastAPI application factory with lifespan management and CORS configuration.

app module-attribute 

app = FastAPI(root_path='/api', lifespan=lifespan)

AIHordeStatus

Bases: BaseModel

Status of the external AI Horde API connection.

Source code in src/horde_model_reference/service/app.py 
30
31
32
33
34
35
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 
38
39
40
41
42
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 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@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

Root endpoint for the Horde Model Reference API1.

Source code in src/horde_model_reference/service/app.py 
100
101
102
103
104
105
@app.get("/")
async def read_root() -> ContainsMessage:
    """Root endpoint for the Horde Model Reference API1."""
    return ContainsMessage(
        message="Welcome to the Horde Model Reference API Check `/api/docs` for 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 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@app.get("/heartbeat")
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 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@app.get("/replicate_mode")
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,
    )