Subscription Operations

WebSocket subscriptions for real-time updates.

Bases: StashClientProtocol

Mixin for subscription-related client methods.

Functions

subscribe_to_jobs async

subscribe_to_jobs() -> AsyncIterator[
    AsyncIterator[JobStatusUpdate]
]

Subscribe to job status updates.

Returns:

Type	Description
AsyncIterator[AsyncIterator[JobStatusUpdate]]	An async context manager that yields an async iterator of JobStatusUpdate objects

Example

async with client.subscribe_to_jobs() as subscription:
    async for update in subscription:
        print(f"Job {update.job.id}: {update.status} ({update.progress}%)")
        if update.status == "FINISHED":
            break

subscribe_to_logs async

subscribe_to_logs() -> AsyncIterator[
    AsyncIterator[list[LogEntry]]
]

Subscribe to log entries.

Returns:

Type	Description
AsyncIterator[AsyncIterator[list[LogEntry]]]	An async context manager that yields an async iterator of LogEntry lists

Example

async with client.subscribe_to_logs() as subscription:
    async for logs in subscription:
        for entry in logs:
            print(f"{entry.time} [{entry.level}] {entry.message}")

subscribe_to_scan_complete async

subscribe_to_scan_complete() -> AsyncIterator[
    AsyncIterator[bool]
]

Subscribe to scan completion events.

Returns:

Type	Description
AsyncIterator[AsyncIterator[bool]]	An async context manager that yields an async iterator of scan completion events

Example

async with client.subscribe_to_scan_complete() as subscription:
    async for _ in subscription:
        print("Scan completed!")
        await client.metadata_generate(...)  # Generate after scan

wait_for_job_with_updates async

wait_for_job_with_updates(
    job_id: str,
    status: JobStatus = FINISHED,
    timeout: float = 120,
) -> bool | None

Wait for a job to complete with real-time updates.

Parameters:

Name	Type	Description	Default
job_id	str	Job ID to wait for	required
status	JobStatus	Status to wait for	FINISHED
timeout	float	Maximum time to wait in seconds	120

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

Example

job_id = await client.metadata_generate(...)
if await client.wait_for_job_with_updates(job_id):
    print("Generation complete!")