pixano_inference.client

Client for the Pixano Inference /v1 API.

Two clients share one contract: :class:PixanoInferenceClient (async) and :class:SyncPixanoInferenceClient (sync). Both hold a pooled httpx transport, send the optional API key, retry transient failures with backoff, and raise :class:PixanoInferenceError for every failure: the server's {code, message, requestId} envelope on an error status, connection_error when the server cannot be reached, and invalid_response when a successful response does not match its schema. Requests serialize as camelCase JSON (by_alias=True); media is passed by value as a URL, base64 data-URI, or media-root path (the server also exposes /binary multipart routes for callers that prefer raw uploads).

PixanoInferenceClient(url, *, transport=None, **kwargs)

Bases: _ClientBase

Asynchronous client for the Pixano Inference /v1 API.

Parameters:

Name	Type	Description	Default
url	str	Base server URL.	required
transport	AsyncBaseTransport | None	Optional httpx transport (e.g. an ASGI transport for in-process tests).	None
**kwargs	Any	Shared client options (api_key, timeout, retries, ...).	{}

Source code in pixano_inference/client.py

def __init__(self, url: str, *, transport: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> None:
    """Create the client and its pooled async transport.

    Args:
        url: Base server URL.
        transport: Optional httpx transport (e.g. an ASGI transport for in-process tests).
        **kwargs: Shared client options (api_key, timeout, retries, ...).
    """
    super().__init__(url, **kwargs)
    self._client = httpx.AsyncClient(timeout=self._timeout, transport=transport)

__aenter__() async

Enter the async context manager.

Source code in pixano_inference/client.py

async def __aenter__(self) -> PixanoInferenceClient:
    """Enter the async context manager."""
    return self

__aexit__(*exc) async

Close the client on context exit.

Source code in pixano_inference/client.py

async def __aexit__(self, *exc: Any) -> None:
    """Close the client on context exit."""
    await self.aclose()

aclose() async

Close the underlying connection pool.

Source code in pixano_inference/client.py

async def aclose(self) -> None:
    """Close the underlying connection pool."""
    await self._client.aclose()

cancel_job(job_id) async

Cancel a job.

Source code in pixano_inference/client.py

async def cancel_job(self, job_id: str) -> JobStatus:
    """Cancel a job."""
    response = await self._request("DELETE", f"/v1/jobs/{job_id}")
    return self._parse(response, JobStatus)

connect(url, *, api_key=None, **kwargs) classmethod

Construct a client for url (kept for backward compatibility).

Source code in pixano_inference/client.py

@classmethod
def connect(cls, url: str, *, api_key: str | None = None, **kwargs: Any) -> PixanoInferenceClient:
    """Construct a client for *url* (kept for backward compatibility)."""
    return cls(url, api_key=api_key, **kwargs)

deploy_model(request, *, timeout=None) async

Deploy a model at runtime.

Source code in pixano_inference/client.py

async def deploy_model(self, request: DeployModelRequest, *, timeout: float | None = None) -> ModelStatusInfo:
    """Deploy a model at runtime."""
    response = await self._request(
        "POST", "/v1/models", json=self._json_body(request), timeout=timeout or DEPLOY_TIMEOUT
    )
    return self._parse(response, ModelStatusInfo)

detection(request, *, timeout=None) async

Run object detection.

Source code in pixano_inference/client.py

async def detection(self, request: DetectionRequest, *, timeout: float | None = None) -> DetectionResponse:
    """Run object detection."""
    return await self._infer("/v1/inference/detection", request, DetectionResponse, timeout)

embedding(request, *, timeout=None) async

Compute image or text embeddings (CLIP-style shared space).

Source code in pixano_inference/client.py

async def embedding(self, request: EmbeddingRequest, *, timeout: float | None = None) -> EmbeddingResponse:
    """Compute image or text embeddings (CLIP-style shared space)."""
    return await self._infer("/v1/inference/embedding", request, EmbeddingResponse, timeout)

get_job(job_id) async

Poll the status of a job.

Source code in pixano_inference/client.py

async def get_job(self, job_id: str) -> JobStatus:
    """Poll the status of a job."""
    response = await self._request("GET", f"/v1/jobs/{job_id}")
    return self._parse(response, JobStatus)

health() async

Return the liveness report.

Source code in pixano_inference/client.py

async def health(self) -> dict[str, Any]:
    """Return the liveness report."""
    return (await self._request("GET", "/health")).json()

info() async

Return server and cluster information.

Source code in pixano_inference/client.py

async def info(self) -> dict[str, Any]:
    """Return server and cluster information."""
    return (await self._request("GET", "/v1/info")).json()

list_models() async

List deployed models with their live status.

Source code in pixano_inference/client.py

async def list_models(self) -> list[ModelStatusInfo]:
    """List deployed models with their live status."""
    response = await self._request("GET", "/v1/models")
    return self._parse(response, list[ModelStatusInfo])

ner(request, *, timeout=None) async

Run named entity recognition.

Source code in pixano_inference/client.py

async def ner(self, request: NERRequest, *, timeout: float | None = None) -> NERResponse:
    """Run named entity recognition."""
    return await self._infer("/v1/inference/ner", request, NERResponse, timeout)

ready() async

Return the readiness report (does not raise on 503).

Source code in pixano_inference/client.py

async def ready(self) -> dict[str, Any]:
    """Return the readiness report (does not raise on 503)."""
    response = await self._request("GET", "/v1/ready", retry_statuses=frozenset(), raise_on_error=False)
    return response.json()

segmentation(request, *, timeout=None) async

Run image segmentation.

Source code in pixano_inference/client.py

async def segmentation(
    self, request: SegmentationRequest, *, timeout: float | None = None
) -> SegmentationResponse:
    """Run image segmentation."""
    return await self._infer("/v1/inference/segmentation", request, SegmentationResponse, timeout)

submit_tracking_job(request, *, timeout=None) async

Submit a tracking request as an asynchronous job.

Source code in pixano_inference/client.py

async def submit_tracking_job(self, request: TrackingRequestV1, *, timeout: float | None = None) -> JobStatus:
    """Submit a tracking request as an asynchronous job."""
    response = await self._request(
        "POST", "/v1/inference/tracking/jobs", json=self._json_body(request), timeout=timeout
    )
    return self._parse(response, JobStatus)

tracking(request, *, timeout=None) async

Run synchronous video tracking (short intervals).

Source code in pixano_inference/client.py

async def tracking(self, request: TrackingRequestV1, *, timeout: float | None = None) -> TrackingResponse:
    """Run synchronous video tracking (short intervals)."""
    return await self._infer(
        "/v1/inference/tracking", request, TrackingResponse, timeout or self._tracking_timeout
    )

undeploy_model(name) async

Undeploy a model.

Source code in pixano_inference/client.py

async def undeploy_model(self, name: str) -> dict[str, Any]:
    """Undeploy a model."""
    response = await self._request("DELETE", f"/v1/models/{name}")
    return response.json()

vlm(request, *, timeout=None) async

Run vision-language generation.

Source code in pixano_inference/client.py

async def vlm(self, request: VLMRequest, *, timeout: float | None = None) -> VLMResponse:
    """Run vision-language generation."""
    return await self._infer("/v1/inference/vlm", request, VLMResponse, timeout)

wait_for_job(job_id, *, poll_interval=1.0, timeout=None) async

Poll a job until it reaches a terminal state.

Source code in pixano_inference/client.py

async def wait_for_job(
    self, job_id: str, *, poll_interval: float = 1.0, timeout: float | None = None
) -> JobStatus:
    """Poll a job until it reaches a terminal state."""
    deadline = None if timeout is None else time.monotonic() + timeout
    while True:
        job = await self.get_job(job_id)
        if job.status in _TERMINAL_JOB_STATES:
            return job
        if deadline is not None and time.monotonic() > deadline:
            raise PixanoInferenceError(0, "timeout", f"Job '{job_id}' did not finish within {timeout}s.")
        await asyncio.sleep(poll_interval)

PixanoInferenceError(status_code, code, message, request_id=None)

Bases: Exception

Error raised by the client, carrying the server error envelope.

Attributes:

Name	Type	Description
status_code		HTTP status of the response, 0 when there was none.
code		The server's error code, or one of the client's own: connection_error, timeout, invalid_response (a successful response that does not match its schema, so the server and the client disagree on the contract).
message		What went wrong.
request_id		The server's request id, when a response carried one.

Source code in pixano_inference/client.py

def __init__(self, status_code: int, code: str, message: Any, request_id: str | None = None) -> None:
    """Store the status code, error code, message, and optional request id."""
    self.status_code = status_code
    self.code = code
    self.message = message
    self.request_id = request_id
    suffix = f" (requestId={request_id})" if request_id else ""
    super().__init__(f"[{status_code} {code}] {message}{suffix}")

SyncPixanoInferenceClient(url, *, transport=None, **kwargs)

Bases: _ClientBase

Synchronous twin of :class:PixanoInferenceClient for callers not on an event loop.

Source code in pixano_inference/client.py

def __init__(self, url: str, *, transport: httpx.BaseTransport | None = None, **kwargs: Any) -> None:
    """Create the client and its pooled sync transport."""
    super().__init__(url, **kwargs)
    self._client = httpx.Client(timeout=self._timeout, transport=transport)

__enter__()

Enter the context manager.

Source code in pixano_inference/client.py

def __enter__(self) -> SyncPixanoInferenceClient:
    """Enter the context manager."""
    return self

__exit__(*exc)

Close the client on context exit.

Source code in pixano_inference/client.py

def __exit__(self, *exc: Any) -> None:
    """Close the client on context exit."""
    self.close()

cancel_job(job_id)

Cancel a job.

Source code in pixano_inference/client.py

def cancel_job(self, job_id: str) -> JobStatus:
    """Cancel a job."""
    return self._parse(self._request("DELETE", f"/v1/jobs/{job_id}"), JobStatus)

close()

Close the underlying connection pool.

Source code in pixano_inference/client.py

def close(self) -> None:
    """Close the underlying connection pool."""
    self._client.close()

connect(url, *, api_key=None, **kwargs) classmethod

Construct a client for url.

Source code in pixano_inference/client.py

@classmethod
def connect(cls, url: str, *, api_key: str | None = None, **kwargs: Any) -> SyncPixanoInferenceClient:
    """Construct a client for *url*."""
    return cls(url, api_key=api_key, **kwargs)

deploy_model(request, *, timeout=None)

Deploy a model at runtime.

Source code in pixano_inference/client.py

def deploy_model(self, request: DeployModelRequest, *, timeout: float | None = None) -> ModelStatusInfo:
    """Deploy a model at runtime."""
    response = self._request(
        "POST", "/v1/models", json=self._json_body(request), timeout=timeout or DEPLOY_TIMEOUT
    )
    return self._parse(response, ModelStatusInfo)

detection(request, *, timeout=None)

Run object detection.

Source code in pixano_inference/client.py

def detection(self, request: DetectionRequest, *, timeout: float | None = None) -> DetectionResponse:
    """Run object detection."""
    return self._infer("/v1/inference/detection", request, DetectionResponse, timeout)

embedding(request, *, timeout=None)

Compute image or text embeddings (CLIP-style shared space).

Source code in pixano_inference/client.py

def embedding(self, request: EmbeddingRequest, *, timeout: float | None = None) -> EmbeddingResponse:
    """Compute image or text embeddings (CLIP-style shared space)."""
    return self._infer("/v1/inference/embedding", request, EmbeddingResponse, timeout)

get_job(job_id)

Poll the status of a job.

Source code in pixano_inference/client.py

def get_job(self, job_id: str) -> JobStatus:
    """Poll the status of a job."""
    return self._parse(self._request("GET", f"/v1/jobs/{job_id}"), JobStatus)

health()

Return the liveness report.

Source code in pixano_inference/client.py

def health(self) -> dict[str, Any]:
    """Return the liveness report."""
    return self._request("GET", "/health").json()

info()

Return server and cluster information.

Source code in pixano_inference/client.py

def info(self) -> dict[str, Any]:
    """Return server and cluster information."""
    return self._request("GET", "/v1/info").json()

list_models()

List deployed models with their live status.

Source code in pixano_inference/client.py

def list_models(self) -> list[ModelStatusInfo]:
    """List deployed models with their live status."""
    response = self._request("GET", "/v1/models")
    return self._parse(response, list[ModelStatusInfo])

ner(request, *, timeout=None)

Run named entity recognition.

Source code in pixano_inference/client.py

def ner(self, request: NERRequest, *, timeout: float | None = None) -> NERResponse:
    """Run named entity recognition."""
    return self._infer("/v1/inference/ner", request, NERResponse, timeout)

ready()

Return the readiness report (does not raise on 503).

Source code in pixano_inference/client.py

def ready(self) -> dict[str, Any]:
    """Return the readiness report (does not raise on 503)."""
    return self._request("GET", "/v1/ready", retry_statuses=frozenset(), raise_on_error=False).json()

segmentation(request, *, timeout=None)

Run image segmentation.

Source code in pixano_inference/client.py

def segmentation(self, request: SegmentationRequest, *, timeout: float | None = None) -> SegmentationResponse:
    """Run image segmentation."""
    return self._infer("/v1/inference/segmentation", request, SegmentationResponse, timeout)

submit_tracking_job(request, *, timeout=None)

Submit a tracking request as an asynchronous job.

Source code in pixano_inference/client.py

def submit_tracking_job(self, request: TrackingRequestV1, *, timeout: float | None = None) -> JobStatus:
    """Submit a tracking request as an asynchronous job."""
    response = self._request("POST", "/v1/inference/tracking/jobs", json=self._json_body(request), timeout=timeout)
    return self._parse(response, JobStatus)

tracking(request, *, timeout=None)

Run synchronous video tracking (short intervals).

Source code in pixano_inference/client.py

def tracking(self, request: TrackingRequestV1, *, timeout: float | None = None) -> TrackingResponse:
    """Run synchronous video tracking (short intervals)."""
    return self._infer("/v1/inference/tracking", request, TrackingResponse, timeout or self._tracking_timeout)

undeploy_model(name)

Undeploy a model.

Source code in pixano_inference/client.py

def undeploy_model(self, name: str) -> dict[str, Any]:
    """Undeploy a model."""
    return self._request("DELETE", f"/v1/models/{name}").json()

vlm(request, *, timeout=None)

Run vision-language generation.

Source code in pixano_inference/client.py

def vlm(self, request: VLMRequest, *, timeout: float | None = None) -> VLMResponse:
    """Run vision-language generation."""
    return self._infer("/v1/inference/vlm", request, VLMResponse, timeout)

wait_for_job(job_id, *, poll_interval=1.0, timeout=None)

Poll a job until it reaches a terminal state.

Source code in pixano_inference/client.py

def wait_for_job(self, job_id: str, *, poll_interval: float = 1.0, timeout: float | None = None) -> JobStatus:
    """Poll a job until it reaches a terminal state."""
    deadline = None if timeout is None else time.monotonic() + timeout
    while True:
        job = self.get_job(job_id)
        if job.status in _TERMINAL_JOB_STATES:
            return job
        if deadline is not None and time.monotonic() > deadline:
            raise PixanoInferenceError(0, "timeout", f"Job '{job_id}' did not finish within {timeout}s.")
        time.sleep(poll_interval)