flightrisk.serving

class flightrisk.serving.schemas.ScoreRequest(*, customer_id=None, features)

Bases: BaseModel

Per-customer feature payload submitted for scoring.

Parameters:

• customer_id (str | None) – Optional opaque identifier echoed in the response.

• features (dict[str, float | int | str | None]) – Mapping from feature name to numeric or string value.

customer_id: str | None

features: dict[str, float | int | str | None]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class flightrisk.serving.schemas.BatchScoreRequest(*, track, run_id=None, horizon_days=None, items)

Bases: BaseModel

Batch payload of multiple ScoreRequest items.

Parameters:

• track (Literal['risk', 'survival', 'uplift']) – Which track to score with.

• run_id (str | None) – Optional run id to pin a specific model version.

• horizon_days (int | None) – Required when track == "survival".

• items (list[ScoreRequest]) – Per-customer feature payloads.

track: Literal['risk', 'survival', 'uplift']

run_id: str | None

horizon_days: int | None

items: list[ScoreRequest]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class flightrisk.serving.schemas.ScoreResponseItem(*, customer_id, score)

Bases: BaseModel

One row of the scoring response.

Parameters:

• customer_id (str | None) – Mirrors the incoming customer_id if any.

• score (float) – Per-row score interpreted per track.

customer_id: str | None

score: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class flightrisk.serving.schemas.BatchScoreResponse(*, track, run_id, score_meaning, scores)

Bases: BaseModel

Batch scoring response.

Parameters:

• track (Literal['risk', 'survival', 'uplift']) – Track that produced the scores.

• run_id (str) – Run id of the model that served the request.

• score_meaning (str) – Human-readable description of what score means.

• scores (list[ScoreResponseItem]) – One ScoreResponseItem per request item.

track: Literal['risk', 'survival', 'uplift']

run_id: str

score_meaning: str

scores: list[ScoreResponseItem]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class flightrisk.serving.schemas.HealthResponse(*, status='ok', loaded)

Bases: BaseModel

Liveness check response.

Parameters:

• status (Literal['ok']) – Always "ok" when the service is reachable.

• loaded (dict[str, str]) – Mapping from track name to loaded run id.

status: Literal['ok']

loaded: dict[str, str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class flightrisk.serving.registry.LoadedModel(track, run_id, model, feature_names)

Bases: object

Metadata for a model loaded into the in-process registry.

Parameters:

• track (Literal['risk', 'survival', 'uplift']) – Track identifier ("risk", "survival" or "uplift").

• run_id (str) – MLflow run identifier (or directory name on disk).

• model (object) – The deserialised estimator bundle.

• feature_names (list[str]) – Feature ordering required by the model.

track: Literal['risk', 'survival', 'uplift']

run_id: str

model: object

feature_names: list[str]

class flightrisk.serving.registry.ModelRegistry

Bases: object

Thread-safe in-memory registry that lazily loads track artifacts.

The registry resolves paths under reports/<track>/ and caches the most recently loaded artifact for each track. Concurrent requests reuse the same instance so reloads only happen when the chosen run id changes.

get(track, run_id=None)

Return a loaded model for track, loading from disk on first use.

Parameters:

• track (Literal['risk', 'survival', 'uplift']) – Track identifier.

• run_id (str | None) – Optional run id; None selects the latest run.

Returns:

A LoadedModel ready for inference.

Raises:

FileNotFoundError – If the artifact is missing.

Return type:

LoadedModel

loaded_tracks()

Return a snapshot of the currently cached run ids per track.

Returns:

Mapping from track name to run id.

Return type:

dict[Literal[‘risk’, ‘survival’, ‘uplift’], str]

flightrisk.serving.registry.get_registry()

Return the process-wide ModelRegistry instance.

Returns:

The shared registry singleton.

Return type:

ModelRegistry

flightrisk.serving.api.health(request)

Return liveness information and the currently cached models.

Excluded from rate-limiting so liveness checks always succeed.

Parameters:

request (Request) – The incoming request, required by slowapi’s signature.

Returns:

A HealthResponse payload.

Return type:

HealthResponse

flightrisk.serving.api.score(request, payload, api_key=Depends(dependency=<function require_api_key>, use_cache=True, scope=None))

Score a batch of customers with the requested track.

Protected by the X-API-Key header (when FLIGHTRISK_API_KEY / FLIGHTRISK_API_KEYS is set) and rate-limited per client IP.

Parameters:

• request (Request) – The incoming request, required by slowapi.

• payload (BatchScoreRequest) – A BatchScoreRequest payload.

• api_key (str) – The validated API key (or sentinel when auth is disabled).

Returns:

A BatchScoreResponse with per-customer scores.

Raises:

HTTPException – When the model artifact is missing or invalid.

Return type:

BatchScoreResponse