pixano_inference.jobs

In-process async job manager for long-running inference (e.g. video tracking).

A job wraps a Ray Serve DeploymentResponse in an asyncio task, so status can be polled without ever blocking the event loop on ray.get. The store is bounded: terminal jobs are evicted past a TTL and when a size cap is exceeded. State is process-local and lost on restart (the computation lives in Serve replicas the server manages), which is appropriate for a single-node deployment.

JobManager(max_jobs=DEFAULT_MAX_JOBS, ttl_s=DEFAULT_TTL_S)

Bounded, in-process manager of asynchronous jobs over Serve responses.

Parameters:

Name	Type	Description	Default
max_jobs	int	Maximum number of retained jobs before terminal jobs are evicted.	DEFAULT_MAX_JOBS
ttl_s	float	Time-to-live for terminal jobs before eviction.	DEFAULT_TTL_S

Source code in pixano_inference/jobs.py

def __init__(self, max_jobs: int = DEFAULT_MAX_JOBS, ttl_s: float = DEFAULT_TTL_S) -> None:
    """Initialize the job manager.

    Args:
        max_jobs: Maximum number of retained jobs before terminal jobs are evicted.
        ttl_s: Time-to-live for terminal jobs before eviction.
    """
    self._jobs: dict[str, JobRecord] = {}
    self._max_jobs = max_jobs
    self._ttl_s = ttl_s

count property

Number of retained jobs.

cancel(job_id)

Cancel a job on a best-effort basis.

Source code in pixano_inference/jobs.py

def cancel(self, job_id: str) -> JobRecord | None:
    """Cancel a job on a best-effort basis."""
    job = self._jobs.get(job_id)
    if job is None:
        return None
    if job.status in TERMINAL_STATES:
        return job
    try:
        if job.response is not None:
            job.response.cancel()
    except Exception as exc:
        logger.warning("Failed to cancel job %s response: %s", job_id, exc)
    if job.task is not None:
        job.task.cancel()
    return self._finalize(job_id, status="canceled", detail="Job canceled.")

cancel_all()

Cancel every non-terminal job (used on shutdown).

Source code in pixano_inference/jobs.py

def cancel_all(self) -> None:
    """Cancel every non-terminal job (used on shutdown)."""
    for job_id in list(self._jobs):
        self.cancel(job_id)

cancel_for_model(model_name, detail='Model undeployed.')

Cancel all non-terminal jobs belonging to a model.

Source code in pixano_inference/jobs.py

def cancel_for_model(self, model_name: str, detail: str = "Model undeployed.") -> None:
    """Cancel all non-terminal jobs belonging to a model."""
    for job_id, job in list(self._jobs.items()):
        if job.model_name == model_name and job.status not in TERMINAL_STATES:
            self.cancel(job_id)
            self._finalize(job_id, status="canceled", detail=detail)

evict_now()

Run one eviction pass (for a periodic background sweep).

Source code in pixano_inference/jobs.py

def evict_now(self) -> None:
    """Run one eviction pass (for a periodic background sweep)."""
    self._evict()

get(job_id)

Return the current state of a job (kept up to date by its task).

Source code in pixano_inference/jobs.py

def get(self, job_id: str) -> JobRecord | None:
    """Return the current state of a job (kept up to date by its task)."""
    return self._jobs.get(job_id)

submit(response, *, model_name, metadata=None)

Register a job for a Serve response and start awaiting it.

Must be called from within a running event loop (i.e. an async route handler).

Parameters:

Name	Type	Description	Default
response	Any	A Serve DeploymentResponse (awaitable, cancellable).	required
model_name	str	Name of the model handling the job.	required
metadata	dict[str, Any] | None	Optional metadata echoed back in job status.	None

Returns:

Type	Description
str	The generated job id.

Source code in pixano_inference/jobs.py

def submit(self, response: Any, *, model_name: str, metadata: dict[str, Any] | None = None) -> str:
    """Register a job for a Serve response and start awaiting it.

    Must be called from within a running event loop (i.e. an async route handler).

    Args:
        response: A Serve ``DeploymentResponse`` (awaitable, cancellable).
        model_name: Name of the model handling the job.
        metadata: Optional metadata echoed back in job status.

    Returns:
        The generated job id.
    """
    job_id = f"job-{uuid4().hex}"
    record = JobRecord(model_name=model_name, response=response, metadata=metadata or {})
    self._jobs[job_id] = record
    record.task = asyncio.get_running_loop().create_task(self._await(job_id, response))
    self._evict()
    return job_id

JobRecord(model_name, response=None, task=None, status='running', detail=None, result=None, metadata=dict(), timestamp=_utcnow(), submitted_at_monotonic=time.time(), processing_time=0.0) dataclass

State for a single asynchronous job.

serialize_job(job_id, job)

Serialize a job record into the camelCase API status shape.

Source code in pixano_inference/jobs.py

def serialize_job(job_id: str, job: JobRecord) -> dict[str, Any]:
    """Serialize a job record into the camelCase API status shape."""
    return {
        "jobId": job_id,
        "status": job.status,
        "detail": job.detail,
        "data": job.result if job.status == "completed" else None,
        "metadata": job.metadata,
        "timestamp": job.timestamp,
        "processingTime": job.processing_time,
    }