pyantz.infrastructure.distributed_queue.common package

pyantz.infrastructure.distributed_queue.common package#

Submodules#

pyantz.infrastructure.distributed_queue.common.interface module#

Required function signatures for all our queues.

While the queues are function orientated, we want to ensure an exact consistent interface, so a protocol is used just to wrap the public interfaces.

Queues are expected to be stateless. The “state” should be persisted in the storage of the queue itself, but the queue instance is merely a convenience wrapper to access whatever persistent layer is used.

Classes may, however, store accessor details in their state. SQLite queues, for example, store a sqlalchemy engine to access the database file. But, after initialization this cannot be updated or changed without a new class instance.

class QueueInterface(*args, **kwargs)[source]#

Bases: Protocol

Type checking interface to ensure consistent accesses.

_abc_impl = <_abc._abc_data object>#
_is_protocol = True#
abstractmethod add_job(job_config: JobWithContext, parent_job_id: str | UUID | None)[source]#

Add the provided job to the queue.

Parameters:

  • job_config (JobWithContext) – job to enqueue

  • parent_job_id (str | UUID | None) – jobs added inherit their parent dependency “responsibilities”. So a job submitting new jobs should pass its own id to the function. For example if B depends on A and A adds C then B will depend on C. If, however, the user DOESN’T want that (B should run parallel to C), then they can pass “None” as the parent_job_id.

Returns:

True if successfully added; False if unable to add

Return type:

bool

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod change_config_in_place(job_config: JobWithContext)[source]#

Change the config of a job.

In practice, this is used to restart jobs during error recovery. A worker will update the num_attempted_runs and change the job. Then it will set its status to PENDING.

The queue will then set PENDING to READY on its next check and return the job to be re-run (the job is put back in the queue if a worker had already taken it out).

Parameters:

job_config (JobWithContext) – configuration to run

Return type:

None

Returns:

None.

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod fail_deadlocked_jobs()[source]#

Mark PENDING jobs as ERROR if nothing is READY or RUNNING.

In that state nothing can ever change the status of a pending job (for example, it depends on a job id that was never added), so the pipeline would wait forever.

Parameters:

None.

Returns:

ids of the jobs which were failed

Return type:

list[str]

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod fail_jobs_of_worker(worker_id: int)[source]#

Mark the RUNNING jobs of a worker which died as ERROR.

Parameters:

worker_id (int) – the worker (its process id) which died.

Returns:

ids of the jobs which were failed

Return type:

list[str]

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod get_failed_jobs()[source]#

Get the status (with the error details) of every job which errored.

Parameters:

None.

Returns:

the failed jobs, sorted by job id.

Return type:

list[JobStatusReport]

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod get_job()[source]#

Return a job from the queue.

Parameters:

None.

Returns:

either

  • A job to run

  • Pending to indicate that nothing is ready but to keep waiting

  • Complete to tell the worker to kill itself

Return type:

GetJobReturn

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod get_job_status(job_id: str | UUID)[source]#

Get the current status of a job, and the details given with it.

Parameters:

job_id (str | UUID) – job to look up

Returns:

the job’s status and its details.

Return type:

JobStatusReport

Raises:

KeyError – if there is no such job

abstractmethod get_jobs_report()[source]#

Get the jobs report.

Useful for debugging or determining the size of the queue.

Parameters:

None.

Returns:

report of the jobs and their states.

Return type:

JobsReport

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod set_job_worker(job_id: str | UUID, worker_id: int)[source]#

Record which worker is running the job.

Parameters:

  • job_id (str | UUID) – the job being run.

  • worker_id (int) – the worker (its process id) running it.

Return type:

None

Returns:

None.

Raises:

Exception – implementations may raise if the queue’s storage fails.

abstractmethod update_job_status(job_id: str | UUID, status: JobStatus, details: str | None = None)[source]#

Update the status of the job.

Parameters:

  • job_id (str | UUID) – job to be updated

  • status (JobStatus) – new state after this update

  • details (str | None) – additional information about status eg., may be the error message for debugging.

Return type:

None

Returns:

None.

Raises:

Exception – implementations may raise if the queue’s storage fails.

pyantz.infrastructure.distributed_queue.common.return_types module#

Return types used by queue implementations.

GetJobReturn is the discriminated union returned by get_job(). It wraps one of three payload types:

JobsReport is a snapshot of job counts by status, used for debugging and progress monitoring.

Example of GetJobReturn:

from pyantz.infrastructure.distributed_queue.common import (
    GetJobReturn,
)

report = {
    "total": 1, "success": 1, "error": 0, "pending": 0, "ready": 0,
    "running": 0,
}
result = GetJobReturn.model_validate(
    {"payload": {"type_": "complete", "jobs_report": report}}
)
print(result.payload.type_, result.payload.jobs_report.all_succeeded)

complete True
class CompleteReturn(*, type_: Literal['complete'] = 'complete', jobs_report: JobsReport)[source]#

Bases: BaseModel

Indicates that the queue is empty and will not have additional jobs added.

Parameters:

  • type_ (Literal['complete'])

  • jobs_report (JobsReport)

_abc_impl = <_abc._abc_data object>#
jobs_report: JobsReport#

The final state of the queue.

type_: Literal['complete']#

Tells the returns apart at runtime; faster/easier than an isinstance check.

class GetJobReturn(*, payload: JobReturn | CompleteReturn | PendingReturn)[source]#

Bases: BaseModel

Return value of the get_job function.

This is used like a Rust enum return to return a “union” type.

Rather than always returning a job, get_job can return either

  1. Job: Successfully dequeue a job

  2. Pending: either jobs are ready and awaiting dependencies or a job is running and it may still submit something

  3. Complete: the queue is empty and the worker should kill itself

Parameters:

payload (JobReturn | CompleteReturn | PendingReturn)

_abc_impl = <_abc._abc_data object>#
payload: JobReturn | CompleteReturn | PendingReturn#

The actual return value, told apart by its type_.

class JobReturn(*, type_: Literal['job'] = 'job', job: JobWithContext)[source]#

Bases: BaseModel

Actual job returned to be run.

Parameters:
_abc_impl = <_abc._abc_data object>#
job: JobWithContext#

The job to run.

type_: Literal['job']#

Tells the returns apart at runtime; faster/easier than an isinstance check.

class JobStatus(*values)[source]#

Bases: IntEnum

Possible statuses of a job.

A job starts PENDING, becomes READY once every job it depends on succeeded, RUNNING when a worker takes it, and finally SUCCESS or ERROR. A job whose dependency fails becomes ERROR too.

ERROR = 3#

Something went wrong; this job will never be completed. Depending on the setup, this might cause the whole pipeline to shutdown.

PENDING = 1#

Job is waiting for dependencies.

READY = 0#

Job can be run! Just waiting for a worker.

RUNNING = 2#

Job is running on a worker right now.

SUCCESS = 4#

Success! The job completed.

class JobStatusReport(*, job_id: str | UUID, status: JobStatus, status_details: str | None = None)[source]#

Bases: BaseModel

Status of a job.

Parameters:

  • job_id (str | UUID)

  • status (JobStatus)

  • status_details (str | None)

_abc_impl = <_abc._abc_data object>#
job_id: str | UUID#

Identifier for the job.

status: JobStatus#

Actual status of the job.

status_details: str | None#

Additional information about the status; for example, for errors this is the error message.

class JobsReport(*, total: int, success: int, error: int, pending: int, ready: int, running: int)[source]#

Bases: BaseModel

Debug message with counts of various jobs for runtime analysis.

Parameters:

  • total (Annotated[int, Ge(ge=0)])

  • success (Annotated[int, Ge(ge=0)])

  • error (Annotated[int, Ge(ge=0)])

  • pending (Annotated[int, Ge(ge=0)])

  • ready (Annotated[int, Ge(ge=0)])

  • running (Annotated[int, Ge(ge=0)])

_abc_impl = <_abc._abc_data object>#
property all_succeeded: bool#

True if no job errored and none is left unfinished.

Parameters:

None.

Returns:

True if every job finished successfully.

Return type:

bool

Raises:

None. –

Example

from pyantz.infrastructure.distributed_queue.common import JobsReport

report = JobsReport(
    total=3, success=2, error=0, pending=0, ready=0, running=1
)
print(report.all_succeeded)

False
error: int#

Number of jobs which updated themselves as error.

pending: int#

Number of jobs waiting for dependencies before running.

ready: int#

Number of jobs that could be run RIGHT NOW.

running: int#

Number of jobs currently being run by a worker.

success: int#

Number of jobs which updated themselves as successful.

total: int#

Total number of unique jobs put on the queue.

class PendingReturn(*, type_: Literal['pending'] = 'pending', jobs_report: JobsReport)[source]#

Bases: BaseModel

Indicates that the queue returned nothing but may return something later.

Parameters:

  • type_ (Literal['pending'])

  • jobs_report (JobsReport)

_abc_impl = <_abc._abc_data object>#
jobs_report: JobsReport#

The state of the queue.

type_: Literal['pending']#

Tells the returns apart at runtime; faster/easier than an isinstance check.

Module contents#

Common types across queue implementations.