pixano_inference.ray.app

DeploymentManager and FastAPI app factory for Ray Serve.

BodySizeLimitMiddleware(app, max_body_bytes)

Bases: BaseHTTPMiddleware

Reject requests whose body exceeds a configured maximum, before reading it.

Source code in pixano_inference/ray/app.py

def __init__(self, app: Any, max_body_bytes: int) -> None:
    """Store the maximum allowed body size in bytes."""
    super().__init__(app)
    self._max = max_body_bytes

dispatch(request, call_next) async

Return 413 when the declared Content-Length exceeds the configured maximum.

Source code in pixano_inference/ray/app.py

async def dispatch(self, request: Request, call_next):  # type: ignore[override]
    """Return 413 when the declared Content-Length exceeds the configured maximum."""
    if self._max > 0:
        content_length = request.headers.get("content-length")
        if content_length is not None:
            try:
                if int(content_length) > self._max:
                    return JSONResponse(
                        status_code=413,
                        content={"error": {"code": "payload_too_large", "message": "Request body too large."}},
                    )
            except ValueError:
                pass
    return await call_next(request)

DeploymentManager(config, should_abort=None)

In-process manager for Ray Serve model deployments and async tracking jobs.

Each model runs as its own Serve application (serve.run(app, name=..., route_prefix=None)). Handles are obtained lazily via serve.get_app_handle and inference is dispatched through the native async DeploymentResponse. Deployment/config state is process-local; Serve owns replica supervision, autoscaling, and batching. Async jobs are delegated to a :class:~pixano_inference.jobs.JobManager.

Parameters:

Name	Type	Description	Default
config	RayServeConfig	Ray Serve configuration.	required
should_abort	Callable[[], bool] | None	Optional predicate polled while waiting on Ray; when it returns True the wait is cut short (see :func:_run_with_node_watchdog).	None

Source code in pixano_inference/ray/app.py

def __init__(self, config: RayServeConfig, should_abort: Callable[[], bool] | None = None) -> None:
    """Initialize the deployment manager.

    Args:
        config: Ray Serve configuration.
        should_abort: Optional predicate polled while waiting on Ray; when it returns True
            the wait is cut short (see :func:`_run_with_node_watchdog`).
    """
    self._config = config
    self._should_abort = should_abort
    self._handles: dict[str, Any] = {}  # model_name -> Serve DeploymentHandle (cache)
    self._configs: dict[str, ModelDeploymentConfig] = {}  # model_name -> config
    self._metadata_cache: dict[str, dict[str, Any]] = {}  # model_name -> metadata
    self._interface_cache: dict[str, ModelInterface | None] = {}  # model_name -> declared interface
    self.jobs = JobManager()

config property

Server configuration.

cancel_tracking_job(job_id)

Cancel a tracking job on a best-effort basis.

Source code in pixano_inference/ray/app.py

def cancel_tracking_job(self, job_id: str) -> JobRecord | None:
    """Cancel a tracking job on a best-effort basis."""
    return self.jobs.cancel(job_id)

deploy_model(config)

Deploy a model as its own Ray Serve application.

Parameters:

Name	Type	Description	Default
config	ModelDeploymentConfig	Model deployment configuration.	required

Raises:

Type	Description
ValueError	If the model is already deployed or resources are insufficient.
KeyError	If the model class is not registered.
StartupAborted	If shutdown was requested while waiting for the deployment.
RuntimeError	If the Serve deployment fails to become healthy.

Source code in pixano_inference/ray/app.py

def deploy_model(self, config: ModelDeploymentConfig) -> None:
    """Deploy a model as its own Ray Serve application.

    Args:
        config: Model deployment configuration.

    Raises:
        ValueError: If the model is already deployed or resources are insufficient.
        KeyError: If the model class is not registered.
        StartupAborted: If shutdown was requested while waiting for the deployment.
        RuntimeError: If the Serve deployment fails to become healthy.
    """
    if config.name in self._configs:
        raise ValueError(f"Model '{config.name}' is already deployed.")

    model_class = ModelClassRegistry.get(config.model_class)
    self._preflight_resource_check(config)

    app = build_model_app(model_class, config)
    try:
        self._run_serve_app(app, config.name, timeout_s=self._deploy_timeout_s())
    except StartupAborted:
        # Shutting down: the caller drains Serve and Ray wholesale, so skip the per-app
        # cleanup rather than spend the teardown budget on it.
        raise
    except Exception as exc:
        self._best_effort_delete(config.name)
        raise RuntimeError(f"Failed to deploy model '{config.name}': {exc}") from exc

    self._configs[config.name] = config
    self._handles.pop(config.name, None)
    self._interface_cache[config.name] = self._fetch_interface(config.name)
    logger.info(
        "Deployed model '%s' (class=%s, capability=%s)", config.name, config.model_class, config.capability
    )

get_gpu_info()

Get GPU resource information from Ray.

Source code in pixano_inference/ray/app.py

def get_gpu_info(self) -> dict[str, Any]:
    """Get GPU resource information from Ray."""
    if not ray.is_initialized():
        return {"num_gpus": 0, "available_gpus": 0.0, "gpus_used": 0.0}
    cluster_resources = ray.cluster_resources()
    available_resources = ray.available_resources()
    total_gpus = float(cluster_resources.get("GPU", 0.0))
    available_gpus = float(available_resources.get("GPU", 0.0))
    used_gpus = max(0.0, total_gpus - available_gpus)
    return {
        "num_gpus": int(total_gpus),
        "available_gpus": available_gpus,
        "gpus_used": used_gpus,
    }

get_handle(name)

Get a Serve deployment handle by model name (cached), or None if not deployed.

Source code in pixano_inference/ray/app.py

def get_handle(self, name: str) -> Any | None:
    """Get a Serve deployment handle by model name (cached), or None if not deployed."""
    if name in self._handles:
        return self._handles[name]
    if name not in self._configs:
        return None
    handle = serve.get_app_handle(name)
    self._handles[name] = handle
    return handle

get_model_capability(name)

Get the deployed capability for a model.

Source code in pixano_inference/ray/app.py

def get_model_capability(self, name: str) -> str | None:
    """Get the deployed capability for a model."""
    config = self._configs.get(name)
    return config.capability if config is not None else None

get_model_interface(name)

How a deployed model is called, as fetched when it was deployed (None if unknown).

Source code in pixano_inference/ray/app.py

def get_model_interface(self, name: str) -> ModelInterface | None:
    """How a deployed model is called, as fetched when it was deployed (``None`` if unknown)."""
    return self._interface_cache.get(name)

get_model_metadata(name)

Get metadata for a deployed model.

Source code in pixano_inference/ray/app.py

def get_model_metadata(self, name: str) -> dict[str, Any]:
    """Get metadata for a deployed model."""
    if name in self._metadata_cache:
        return self._metadata_cache[name]
    config = self._configs.get(name)
    if config is None:
        return {}
    metadata = {
        "model_name": config.name,
        "capability": config.capability,
        "model_class": config.model_class,
    }
    self._metadata_cache[name] = metadata
    return metadata

get_timeout(name, capability)

Resolve the inference timeout for a model, honoring a per-model override.

Source code in pixano_inference/ray/app.py

def get_timeout(self, name: str, capability: str) -> float:
    """Resolve the inference timeout for a model, honoring a per-model override."""
    config = self._configs.get(name)
    if config is not None and config.timeout_s is not None:
        return config.timeout_s
    spec = find_capability(capability)
    return spec.default_timeout_s if spec is not None else _FALLBACK_TIMEOUT_S

get_tracking_job(job_id)

Return the current state of a tracking job.

Source code in pixano_inference/ray/app.py

def get_tracking_job(self, job_id: str) -> JobRecord | None:
    """Return the current state of a tracking job."""
    return self.jobs.get(job_id)

list_models()

List all deployed models.

Source code in pixano_inference/ray/app.py

def list_models(self) -> list[ModelInfo]:
    """List all deployed models."""
    return [
        ModelInfo(
            name=config.name,
            capability=config.capability,
            model_path=config.model_params.get("path")
            if isinstance(config.model_params.get("path"), str)
            else None,
            model_class=config.model_class,
        )
        for config in self._configs.values()
    ]

model_statuses()

Return {model_name: Serve status string} for all configured models.

Source code in pixano_inference/ray/app.py

def model_statuses(self) -> dict[str, str]:
    """Return {model_name: Serve status string} for all configured models."""
    # serve.status() auto-starts a local Ray cluster when none is connected; a status
    # query must never do that.
    apps: dict[str, Any] = {}
    if ray.is_initialized():
        try:
            apps = serve.status().applications
        except Exception:
            pass
    result: dict[str, str] = {}
    for name in self._configs:
        app = apps.get(name)
        if app is None:
            result[name] = "NOT_STARTED"
        else:
            status = app.status
            result[name] = status.value if hasattr(status, "value") else str(status)
    return result

num_nodes()

Number of alive Ray nodes, or 1 when Ray is not initialized.

Source code in pixano_inference/ray/app.py

def num_nodes(self) -> int:
    """Number of alive Ray nodes, or 1 when Ray is not initialized."""
    if not ray.is_initialized():
        return 1
    try:
        return sum(1 for node in ray.nodes() if node.get("Alive"))
    except Exception:
        return 1

readiness()

Report readiness: every configured model must be RUNNING.

Source code in pixano_inference/ray/app.py

def readiness(self) -> dict[str, Any]:
    """Report readiness: every configured model must be RUNNING."""
    statuses = self.model_statuses()
    running = sum(1 for s in statuses.values() if s == "RUNNING")
    ready = all(s == "RUNNING" for s in statuses.values())
    return {
        "ready": ready,
        "models": statuses,
        "models_loaded": running,
        "version": __version__,
    }

submit_tracking_job(model_name, input_data)

Submit a tracking request as an asynchronous job over the Serve handle.

Source code in pixano_inference/ray/app.py

def submit_tracking_job(self, model_name: str, input_data: Any) -> str:
    """Submit a tracking request as an asynchronous job over the Serve handle."""
    handle = self.get_handle(model_name)
    if handle is None:
        raise ValueError(f"Model '{model_name}' is not deployed.")
    response = handle.predict.remote(input_data)
    return self.jobs.submit(response, model_name=model_name, metadata=self.get_model_metadata(model_name))

undeploy_model(name)

Undeploy a model: delete its Serve app (freeing GPU via replica cleanup).

Parameters:

Name	Type	Description	Default
name	str	Model name.	required

Raises:

Type	Description
ValueError	If the model is not deployed.

Source code in pixano_inference/ray/app.py

def undeploy_model(self, name: str) -> None:
    """Undeploy a model: delete its Serve app (freeing GPU via replica cleanup).

    Args:
        name: Model name.

    Raises:
        ValueError: If the model is not deployed.
    """
    if name not in self._configs:
        raise ValueError(f"Model '{name}' is not deployed.")

    self._best_effort_delete(name)

    self._configs.pop(name, None)
    self._handles.pop(name, None)
    self._metadata_cache.pop(name, None)
    self._interface_cache.pop(name, None)
    self.jobs.cancel_for_model(name)
    logger.info("Undeployed model '%s'", name)

StartupAborted

Bases: RuntimeError

Raised when startup is cut short because shutdown was requested (Ctrl-C / SIGTERM).

create_ray_serve_app(config=None, should_abort=None)

Create the FastAPI application and DeploymentManager for Ray Serve.

The returned app carries a lifespan that starts Ray + Serve and deploys startup models on startup, and drains Serve + Ray on shutdown. The lifespan runs when the app is served (or under with TestClient(app)), not on bare construction.

Parameters:

Name	Type	Description	Default
config	RayServeConfig | None	Ray Serve configuration. If None, uses defaults.	None
should_abort	Callable[[], bool] | None	Optional predicate polled during startup; when it returns True, startup is cut short. :class:~pixano_inference.ray.server.InferenceServer wires this to uvicorn's exit flags so Ctrl-C interrupts a long startup instead of being queued behind it.	None

Returns:

Type	Description
tuple[FastAPI, DeploymentManager]	Tuple of (FastAPI app, DeploymentManager).

Source code in pixano_inference/ray/app.py

def create_ray_serve_app(
    config: RayServeConfig | None = None,
    should_abort: Callable[[], bool] | None = None,
) -> tuple[FastAPI, DeploymentManager]:
    """Create the FastAPI application and DeploymentManager for Ray Serve.

    The returned app carries a lifespan that starts Ray + Serve and deploys startup models
    on startup, and drains Serve + Ray on shutdown. The lifespan runs when the app is served
    (or under ``with TestClient(app)``), not on bare construction.

    Args:
        config: Ray Serve configuration. If None, uses defaults.
        should_abort: Optional predicate polled during startup; when it returns True, startup is
            cut short. :class:`~pixano_inference.ray.server.InferenceServer` wires this to
            uvicorn's exit flags so Ctrl-C interrupts a long startup instead of being queued
            behind it.

    Returns:
        Tuple of (FastAPI app, DeploymentManager).
    """
    if config is None:
        config = RayServeConfig()

    # Register the installed model packages (entry-point plugins).
    from pixano_inference.plugins import ensure_models_loaded

    ensure_models_loaded()

    server_settings = ServerSettings()
    warn_if_auth_disabled(server_settings, config.host)

    deployment_manager = DeploymentManager(config, should_abort=should_abort)

    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        # Ray/Serve startup and model loading block for a long time. Keep them off the event loop:
        # uvicorn replaces SIGINT with a handler that only sets a flag the loop polls, so blocking
        # the loop here makes Ctrl-C inert for the whole of startup. abandon_on_cancel lets us walk
        # away from a Ray call that never returns instead of hanging on the worker thread.
        def _aborted() -> bool:
            return should_abort is not None and should_abort()

        try:
            await to_thread.run_sync(_start_ray_and_serve, config, should_abort, abandon_on_cancel=True)
            failures: list[str] = []
            for model_config in config.models:
                if _aborted():
                    raise StartupAborted("Startup aborted: shutdown was requested.")
                try:
                    await to_thread.run_sync(deployment_manager.deploy_model, model_config, abandon_on_cancel=True)
                    logger.info("Startup model '%s' deployed", model_config.name)
                except StartupAborted:
                    raise
                except Exception as exc:
                    logger.error("Failed to deploy startup model '%s': %s", model_config.name, exc)
                    failures.append(model_config.name)
            if failures and config.strict_startup:
                raise RuntimeError(
                    f"Strict startup: {len(failures)} model(s) failed to deploy: {failures}. "
                    "Pass --no-strict-startup to start anyway."
                )
        except BaseException as exc:
            # Startup failed or was interrupted: drain whatever came up, so a half-started (or
            # already dead) cluster cannot hang the exit path through Ray's own atexit hooks.
            # Nothing has been served yet, so drain on a short leash rather than the full
            # graceful-shutdown window.
            if isinstance(exc, StartupAborted):
                logger.info("Startup interrupted before completion; shutting down")
            await _drain_ray_and_serve(config, timeout_s=_ABORT_DRAIN_TIMEOUT_S)
            raise

        try:
            yield
        finally:
            await _drain_ray_and_serve(config)

    app = FastAPI(
        title="Pixano Inference (Ray)",
        description="Pixano Inference API powered by Ray Serve",
        version=__version__,
        lifespan=lifespan,
    )

    # Body-size limit (before any body is read).
    app.add_middleware(BodySizeLimitMiddleware, max_body_bytes=server_settings.max_request_body_bytes)

    # CORS, only when explicitly configured.
    if server_settings.cors_allow_origins:
        from fastapi.middleware.cors import CORSMiddleware

        app.add_middleware(
            CORSMiddleware,
            allow_origins=server_settings.cors_allow_origins,
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # Request-id propagation (outermost) + Prometheus HTTP metrics.
    from pixano_inference.observability import install_observability_middleware

    install_observability_middleware(app)

    # Consistent {"error": {...}} envelope; unhandled errors never leak their text.
    register_exception_handlers(app)

    auth_dependency = make_api_key_dependency(server_settings)

    # Mount the versioned API (/v1) plus the unversioned /health alias.
    register_v1_api(app, deployment_manager, auth_dependency=auth_dependency)

    # Store references in app state for access by routes
    app.state.config = config
    app.state.server_settings = server_settings
    app.state.deployment_manager = deployment_manager

    return app, deployment_manager