Job Operations - Stash Tools & Scrapers

Operations for managing background jobs.

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

Bases: StashClientProtocol Mixin for job-related client methods.

Functions

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

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)}")