hsml.default_predictor #

DefaultPredict #

Predictor for deployments whose requests are described by a deployment schema.

Validates each request against the schema, looks up and transforms the feature vector through the feature view, runs the model when the deployment has one, and logs the request when the feature view has logging enabled. Without a model it returns the transformed vectors. Subclass it and override load_model or model_predict for models the default loader cannot handle.

Example

# predictor.py, deployed with model.deploy(script_file="predictor.py",
# default_predictor=True) so the schema is still inferred
from hsml.default_predictor import DefaultPredict


class Predict(DefaultPredict):
    def load_model(self, model_files_path):
        return my_framework.load(model_files_path)

    def model_predict(self, feature_vectors):
        return self.model.predict_proba(
            feature_vectors[self.model_input_columns]
        )

check_model_input_contract #

check_model_input_contract() -> None

Fail unless the model's input schema is served by the transformed feature view schema.

Columnar inputs must all exist among the transformed columns with a compatible type family; a tensor input must be a single tensor whose last dimension is the number of transformed columns. Nothing is guessed: a missing or incompatible input schema raises.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.ModelServingException`	On any mismatch, naming the columns or shapes involved.

close #

close(timeout: float = 5.0) -> bool

Drain the queued log requests.

Registered to run at interpreter exit; call it earlier for an orderly shutdown.

PARAMETER	DESCRIPTION
`timeout`	Seconds to wait for the queue to empty and the worker to stop. TYPE: `float` DEFAULT: `5.0`

RETURNS	DESCRIPTION
`bool`	`True` when the queue emptied in time.

fetch_feature_vectors #

fetch_feature_vectors(rows: list[dict[str, Any]]) -> Any

Look up and transform the feature vectors for rows.

PARAMETER	DESCRIPTION
`rows`	Validated rows as returned by `prepare_rows`. TYPE: `list[dict[str, Any]]`

RETURNS	DESCRIPTION
`Any`	A pandas DataFrame in the transformed schema order, one row per input row.

RAISES	DESCRIPTION
`fastapi.HTTPException`	404 when an entity is missing, 422 when a transformation fails, 503 when the feature store is unreachable.

load_model #

load_model(model_files_path: str) -> Any

Load the single pickle or joblib file under model_files_path.

PARAMETER	DESCRIPTION
`model_files_path`	Directory holding the model files (`MODEL_FILES_PATH` in the pod). TYPE: `str`

RETURNS	DESCRIPTION
`Any`	The loaded model object.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.ModelServingException`	If there is not exactly one candidate file; override this method or pass your own script then.

log #

log(
    rows: list[dict[str, Any]],
    feature_vectors: Any,
    predictions: Any,
    request_id: str | None = None,
) -> None

Queue the request for logging through the feature view when logging is enabled.

The logging frame is built and handed to the wrapper's async logger on a background thread, so this returns as soon as the request is queued. The backlog holds at most FEATURE_LOGGER_QUEUE_SIZE rows, or one request's batch when that alone is larger; beyond that the request's rows are dropped and counted, and a failure on the worker is counted and logged and never fails a prediction.

PARAMETER	DESCRIPTION
`rows`	Validated rows as returned by `prepare_rows`. TYPE: `list[dict[str, Any]]`
`feature_vectors`	The DataFrame returned by `fetch_feature_vectors`, carrying the logging metadata. TYPE: `Any`
`predictions`	The model output, ignored without a model. TYPE: `Any`
`request_id`	Correlation id stored with every logged row. TYPE: `str \| None` DEFAULT: `None`

model_predict #

model_predict(feature_vectors: Any) -> Any

Run the model on the transformed feature vectors, or return them unchanged when there is no model.

PARAMETER	DESCRIPTION
`feature_vectors`	The DataFrame returned by `fetch_feature_vectors`. TYPE: `Any`

RETURNS	DESCRIPTION
`Any`	The model's predictions as a list, or `feature_vectors` itself without a model.

RAISES	DESCRIPTION
`fastapi.HTTPException`	500 when the model raises.

predict #

predict(inputs: Any, request_id: str | None = None) -> Any

Serve one request: validate, look up, predict or return the vectors, log.

PARAMETER	DESCRIPTION
`inputs`	The request rows, as objects keyed by field name or arrays in schema order. TYPE: `Any`
`request_id`	Correlation id from the `x-request-id` header; generated when absent. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	The predictions list with a model, else `{"predictions": [...vectors...], "columns": [...]}`.

prepare_rows #

prepare_rows(inputs: Any) -> list[dict[str, Any]]

Validate the request rows against the schema.

PARAMETER	DESCRIPTION
`inputs`	The request rows, as objects keyed by field name or arrays in schema order. TYPE: `Any`

RETURNS	DESCRIPTION
`list[dict[str, Any]]`	The rows as dicts keyed by field name.

RAISES	DESCRIPTION
`fastapi.HTTPException`	400 with the structured errors on a mismatch, 413 on an oversize batch.

PredictionError #

Bases: ModelServingException

A request could not be served; carries the HTTP status and the structured error detail.

run_kserve_wrapper #

run_kserve_wrapper() -> None

Replace this process with the KServe wrapper that serves the Predict class of SCRIPT_PATH.

Deployments without a model artifact are started as python <script>; calling this from the script's __main__ block gives them the same server, payload handling, and feature logger as model deployments.