pyantz.infrastructure.distributed_queue.sql package

Submodules

pyantz.infrastructure.distributed_queue.sql.orm module

SQLAlchemy ORM models for the pyantz SQLite job queue.

The queue is backed by four tables:

• job_queue — ordered list of job IDs waiting to be dequeued.

• queue_items — chunked JSON representations of each job config.

• dependencies — directed edges of the job dependency graph.

• job_status — current lifecycle state of every known job.

Large job configs are split into CHUNKSIZE-character chunks so that SQLite’s per-column string limit is not exceeded.

class Base(**kwargs: Any)

Bases: DeclarativeBase

The base class for our ORM.

Parameters:

kwargs (Any)

_sa_registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

CHUNKSIZE: Final[int] = 2000

Number of characters of a job’s config stored per row; multiple rows might make up one job.

class DependencyTable(**kwargs)

Bases: Base

The dependency table holds the dependency graph of the jobs.

_sa_class_manager = {'depends_on': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'job_id': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}

depends_on: Mapped[str]

Id of the job it waits for.

job_id: Mapped[str]

Id of the job which waits.

class JobConfigTable(**kwargs)

Bases: Base

The job config table holds the configurations for the jobs in the queue.

Note that a job may span multiple tuples (rows) if the config is very large. In that case, it will have multiple sub-indices.

_sa_class_manager = {'job_config_content': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'job_id': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'job_subindex': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}

job_config_content: Mapped[str]

A chunk of the job’s config, as JSON, of at most CHUNKSIZE characters.

job_id: Mapped[str]

Id of the job.

job_subindex: Mapped[int]

Position of this chunk in the job’s config.

class JobQueue(**kwargs)

Bases: Base

Actual queue, holds jobs as they’re added.

_sa_class_manager = {'job_id': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'q_index': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}

job_id: Mapped[str]

Id of the queued job.

q_index: Mapped[int]

Position in the queue; jobs are dequeued in increasing order.

class StatusTable(**kwargs)

Bases: Base

The status table holds the current status of each job.

_sa_class_manager = {'details': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'job_id': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'job_status': <sqlalchemy.orm.attributes.InstrumentedAttribute object>, 'worker_id': <sqlalchemy.orm.attributes.InstrumentedAttribute object>}

details: Mapped[str | None]

Why the job is in this status, for example the error message for failures.

job_id: Mapped[str]

Id of the job.

job_status: Mapped[int]

The job’s JobStatus.

worker_id: Mapped[int | None]

The worker (its process id) which is running the job, while it is RUNNING.

pyantz.infrastructure.distributed_queue.sql.sql_queue module

Queue running with sql.

SqliteQueue keeps the queue in a SQLite file (see orm for its tables), so several worker processes can share it.

BUSY_TIMEOUT_SECONDS = 30

How long a connection waits for another process to release the database lock.

class SqliteQueue(database_file: Path)

Bases: QueueInterface

Queue of jobs stored in a SQLite database file.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.common import JobStatus
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("example.db"))
job = JobWithContext(
    job_id="a", function="pyantz.jobs.testing.nop.do_nothing", parameters={}
)
queue.add_job(job, None)

result = queue.get_job()
print(result.payload.type_, result.payload.job.job_id)
queue.update_job_status("a", JobStatus.SUCCESS)
print(queue.get_job().payload.type_)

job a
complete

Parameters:

database_file (Path)

_abc_impl = <_abc._abc_data object>

static _check_empty_type(sesh: Session)

Check why the queue is empty (pending? or dead/complete?).

Parameters:

sesh (Session) – current sqlalchemy session

Returns:

pending if jobs are running, pending or ready; otherwise complete (every job is ERROR or SUCCESS).

Return type:

GetJobReturn

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

_create_engine(database_file: Path)

Create the engine and return it.

Connections use write-ahead logging and wait for locks rather than failing immediately, so many worker processes can share one queue file. Note that WAL needs a local filesystem (not NFS) for the queue file.

Parameters:

database_file (Path) – the SQLite file holding the queue.

Returns:

the engine, which sets up each new connection.

Return type:

Engine

Raises:

None. –

_engine: Engine

Engine connecting to database_file.

static _generate_jobs_report(sesh: Session)

Create a report of the current jobs on the queue.

Parameters:

sesh (Session) – the current sqlalchemy session.

Returns:

the number of jobs in each status, and in total.

Return type:

JobsReport

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

static _get_job_config(sesh: Session, job_id: str)

Get the job config by piecing the various chunks together.

Parameters:

• sesh (Session) – the current sqlalchemy session.

• job_id (str) – the job to load.

Returns:

the job.

Return type:

JobReturn

Raises:

• RuntimeError – if the job has no stored config.

• pydantic.ValidationError – if the stored config isn’t a valid job.

_insert_job(job_config: JobWithContext, parent_job_id: str | UUID | None)

Insert all rows for a job in one transaction.

Parameters:

• job_config (JobWithContext) – the job to insert, as PENDING.

• parent_job_id (str | UUID | None) – the job which submitted it, whose dependents will wait for this job too; None for no parent.

Return type:

None

Returns:

None.

Raises:

sqlalchemy.exc.IntegrityError – if the job id is already present

_is_protocol = False

_try_dequeue()

Dequeue the next READY job and mark it RUNNING in one transaction.

Parameters:

None.

Returns:

the result of the dequeue, or None if the job

dequeued had a config which could not be loaded. That job is marked ERROR rather than being lost.

Return type:

GetJobReturn | None

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

_update_job_statuses()

Update statuses of all the jobs in the queue.

Propagates errors and successes down the dependency chain.

Specifically, if a job status is ERROR, set its immediate dependents to ERROR; with each call, this will propagate down the dependency chain. And, once every job a PENDING job depends on is SUCCESS, set it to READY.

Parameters:

None.

Return type:

None

Returns:

None.

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

add_job(job_config: JobWithContext, parent_job_id: str | UUID | None)

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

(for example, a job with the same id already exists)

Return type:

bool

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("add_job.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
parent = JobWithContext(job_id="parent", function=nop, parameters={})
after = JobWithContext(
    job_id="after", depends_on={"parent"}, function=nop, parameters={}
)
child = JobWithContext(job_id="child", function=nop, parameters={})
print(queue.add_job(parent, None), queue.add_job(parent, None))
queue.add_job(after, None)
# "after" now waits for "child" too, as "parent" submitted it
queue.add_job(child, parent_job_id="parent")

True False

change_config_in_place(job_config: JobWithContext)

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.common import JobStatus
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("retry.db"))
job = JobWithContext(
    job_id="a",
    function="pyantz.jobs.testing.nop.do_nothing",
    parameters={},
)
queue.add_job(job, None)
taken = queue.get_job().payload.job
retry = taken.model_copy(update={"num_attempted_runs": 1})
queue.change_config_in_place(retry)
queue.update_job_status("a", JobStatus.PENDING)
print(queue.get_job().payload.job.num_attempted_runs)

1

database_file: Path

The SQLite file holding the queue.

fail_deadlocked_jobs()

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

Callers should confirm the state persists across a few polls: a job that just finished may not have had its dependents promoted yet.

Parameters:

None.

Returns:

ids of the jobs which were failed

Return type:

list[str]

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("deadlock.db"))
stuck = JobWithContext(
    job_id="stuck",
    depends_on={"never_added"},
    function="pyantz.jobs.testing.nop.do_nothing",
    parameters={},
)
queue.add_job(stuck, None)
print(queue.fail_deadlocked_jobs())

['stuck']

fail_jobs_of_worker(worker_id: int)

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

The jobs depending on them fail too.

Parameters:

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

Returns:

ids of the jobs which were failed

Return type:

list[str]

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("worker.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
job_a = JobWithContext(job_id="a", function=nop, parameters={})
queue.add_job(job_a, None)
queue.get_job()  # a worker takes the job...
queue.set_job_worker("a", 1234)
print(queue.fail_jobs_of_worker(1234))  # ...and dies

['a']

get_failed_jobs()

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.common import JobStatus
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("failed.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
job_a = JobWithContext(job_id="a", function=nop, parameters={})
queue.add_job(job_a, None)
queue.add_job(
    JobWithContext(
        job_id="b", depends_on={"a"}, function=nop, parameters={}
    ),
    None,
)
queue.update_job_status("a", JobStatus.ERROR, "it broke")
for report in queue.get_failed_jobs():
    print(report.job_id, report.status_details)

a it broke
b A job this job depends on failed

get_job()

Return a job from the queue.

The job is taken out of the queue and marked RUNNING. A job whose config can’t be loaded is marked ERROR and skipped.

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("get_job.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
first = JobWithContext(job_id="first", function=nop, parameters={})
second = JobWithContext(
    job_id="second", depends_on={"first"}, function=nop, parameters={}
)
queue.add_job(first, None)
queue.add_job(second, None)
print(queue.get_job().payload.job.job_id)
print(queue.get_job().payload.type_)  # second waits for first

first
pending

get_job_ids_with_status(status: JobStatus)

Get the ids of the jobs which currently have the status, sorted.

Parameters:

status (JobStatus) – the status to look for.

Returns:

the ids of the jobs with the status, sorted.

Return type:

list[str]

Raises:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.common import JobStatus
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("with_status.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
for job_id in ("b", "a"):
    queue.add_job(
        JobWithContext(job_id=job_id, function=nop, parameters={}),
        None,
    )
print(queue.get_job_ids_with_status(JobStatus.READY))

['a', 'b']

get_job_status(job_id: str | UUID)

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

Example

from pathlib import Path

from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("status.db"))
try:
    queue.get_job_status("unknown")
except KeyError:
    print("no such job")

no such job

get_jobs_report()

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("report.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
job_a = JobWithContext(job_id="a", function=nop, parameters={})
queue.add_job(job_a, None)
queue.add_job(
    JobWithContext(
        job_id="b", depends_on={"a"}, function=nop, parameters={}
    ),
    None,
)
report = queue.get_jobs_report()
print(report.total, report.ready, report.pending)

2 1 1

set_job_worker(job_id: str | UUID, worker_id: int)

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

update_job_status(job_id: str | UUID, status: JobStatus, details: str | None = None)

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:

sqlalchemy.exc.OperationalError – if the database can’t be used.

Example

from pathlib import Path

from pyantz.infrastructure.config import JobWithContext
from pyantz.infrastructure.distributed_queue.common import JobStatus
from pyantz.infrastructure.distributed_queue.sql import SqliteQueue

queue = SqliteQueue(Path("update.db"))
nop = "pyantz.jobs.testing.nop.do_nothing"
job_a = JobWithContext(job_id="a", function=nop, parameters={})
queue.add_job(job_a, None)
queue.update_job_status("a", JobStatus.ERROR, "out of memory")
report = queue.get_job_status("a")
print(report.status.name, report.status_details)

ERROR out of memory

Module contents

Distributed queue that supports multiple processes accessing it simultaneously.

This version relies on sqlite (mostly sqlite).