Job Operations

Operations for managing background jobs.

See also

wait_for_job_with_updates() is in Subscription since it requires a WebSocket connection.

Bases: StashClientProtocol

Mixin for job-related client methods.

Functions

find_job async

find_job(job_id: str) -> Job | None

Find a job by ID.

Parameters:

Name	Type	Description	Default
job_id	str	Job ID to find	required

Returns:

Type	Description
Job | None	Job object if found, None otherwise

Examples:

Find a job and check its status:

job = await client.find_job("123")
if job:
    print(f"Job status: {job.status}")

wait_for_job async

wait_for_job(
    job_id: str | int,
    status: JobStatus = FINISHED,
    period: float = 1.5,
    timeout: float = 120.0,
) -> bool | None

Wait for a job to reach a specific status.

Parameters:

Name	Type	Description	Default
job_id	str | int	Job ID to wait for	required
status	JobStatus	Status to wait for (default: JobStatus.FINISHED)	FINISHED
period	float	Time between checks in seconds (default: 1.5)	1.5
timeout	float	Maximum time to wait in seconds (default: 120)	120.0

Returns:

Type	Description
bool | None	True if job reached desired status
bool | None	False if job finished with different status
bool | None	None if job not found

Raises:

Type	Description
TimeoutError	If timeout is reached
ValueError	If job is not found

stop_job async

stop_job(job_id: str) -> bool

Stop a specific job.

Parameters:

Name	Type	Description	Default
job_id	str	Job ID to stop	required

Returns:

Type	Description
bool	True if job was stopped successfully, False otherwise

Examples:

Stop a running job:

job_id = await client.metadata_generate(...)
success = await client.stop_job(job_id)
if success:
    print(f"Job {job_id} stopped")

stop_all_jobs async

stop_all_jobs() -> bool

Stop all running jobs.

Returns:

Type	Description
bool	True if all jobs were stopped successfully, False otherwise

Examples:

Stop all running jobs:

success = await client.stop_all_jobs()
if success:
    print("All jobs stopped")

job_queue async

job_queue() -> list[Job]

Get all jobs in the queue.

Returns:

Type	Description
list[Job]	List of Job objects representing all jobs (running, pending, finished).
list[Job]	Returns empty list if no jobs are running (jobQueue is null in GraphQL response).

Note

When no jobs are running, Stash returns jobQueue: null in the GraphQL response. This method handles that case and returns an empty list.

Examples:

Get all jobs:

jobs = await client.job_queue()
for job in jobs:
    print(f"Job {job.id}: {job.status} - {job.description}")

Filter by status:

jobs = await client.job_queue()
running = [j for j in jobs if j.status == JobStatus.RUNNING]
print(f"Running jobs: {len(running)}")