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)

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

Source code in pixano_inference/ray/app.py

def __init__(self, config: RayServeConfig) -> None:
    """Initialize the deployment manager.

    Args:
        config: Ray Serve configuration.
    """
    self._config = config
    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.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.
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.
        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=_DEPLOY_TIMEOUT_S)
    except Exception as exc:
        try:
            serve.delete(config.name)
        except Exception:
            pass
        raise RuntimeError(f"Failed to deploy model '{config.name}': {exc}") from exc

    self._configs[config.name] = config
    self._handles.pop(config.name, None)
    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_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
    return _DEFAULT_TIMEOUTS.get(capability, 120.0)

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."""
    try:
        apps = serve.status().applications
    except Exception:
        apps = {}
    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.")

    try:
        serve.delete(name)
    except Exception as exc:
        logger.warning("Error deleting Serve app for '%s': %s", name, exc)

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

create_ray_serve_app(config=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

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,
) -> 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.

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

    # Register built-in backends and installed 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)

    @asynccontextmanager
    async def lifespan(_app: FastAPI):
        _start_ray_and_serve(config)
        failures: list[str] = []
        for model_config in config.models:
            try:
                deployment_manager.deploy_model(model_config)
                logger.info("Startup model '%s' deployed", model_config.name)
            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."
            )
        try:
            yield
        finally:
            try:
                serve.shutdown()
            except Exception as exc:
                logger.warning("serve.shutdown error: %s", exc)
            try:
                if ray.is_initialized():
                    ray.shutdown()
            except Exception as exc:
                logger.warning("ray.shutdown error: %s", exc)

    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