chancy.worker module

class chancy.worker.Worker(chancy, *, worker_id=None, heartbeat_poll_interval=30, heartbeat_timeout=90, queue_change_poll_interval=10, shutdown_timeout=30, tags=None, register_signal_handlers=True, backoff_initial=1.0, backoff_max=60.0, backoff_max_retries=None)

Bases: object

The Worker is responsible for polling queues for new jobs, running any configured plugins, and internal management such as heartbeats.

Starting a worker is simple:

async with Chancy(
    "postgresql://localhost/postgres",
) as chancy:
    await chancy.migrate()
    async with Worker(chancy) as worker:
        await worker.wait_for_shutdown()

Worker Tags

Worker tags control which workers run which queues. A tag is a string (typically in the format key=value) assigned to a worker. Declare a queue with a list of tags, and only workers with matching tags will run that queue.

Every worker automatically gets some tags, like hostname, worker_id, python, arch, os, and * (wildcard). You can also add custom tags when creating a worker.

async with Worker(chancy, tags={"has=gpu", "has=large-disk"}) as worker:
    await worker.wait_for_shutdown()

You could then assign a queue to only run on workers with the has=gpu tag:

await chancy.declare(
    Queue(name="default", tags={"has=gpu"}, concurrency=10),
    upsert=True  # Replaces existing settings
)

Tags are regexes, allowing for flexible matching:

Queue(name="default", tags={r"python=3\.11\.[0-9]+"}, concurrency=10)

Events

The worker emits events that can be listened to by plugins. The following events are emitted:

Event	Kwargs	Description
worker.started	worker	Emitted when the worker is started.
worker.stopped	worker	Emitted when the worker is stopped.
worker.shutdown_timeout	worker	Emitted when the worker encounters a timeout while trying to shut down.
worker.queue.started	queue, executor, worker	Emitted when a queue is started.
worker.queue.removed	queue, executor, worker	Emitted when a queue is removed.
worker.queue.updated	queue, executor, worker	Emitted when a queue is updated.

param chancy:

The Chancy application.

param worker_id:

The ID of the worker, which must be globally unique. If not provided, a random UUID will be generated.

param heartbeat_poll_interval:

The number of seconds between heartbeat poll intervals.

param heartbeat_timeout:

The number of seconds before a worker is considered to have lost connection to the cluster.

param queue_change_poll_interval:

The number of seconds between checks for changes to the queues that the worker should be processing.

param shutdown_timeout:

The number of seconds to wait for a clean shutdown before forcing the worker to stop.

param tags:

Extra tags to associate with the worker.

param register_signal_handlers:

Whether to register signal handlers.

param backoff_initial:

The initial ceiling in seconds for retry delays after a transient database error in internal maintenance loops. Each actual delay is drawn uniformly from [0, ceiling] (full jitter), so workers don’t reconnect in lockstep.

param backoff_max:

The maximum ceiling in seconds. Exponential growth of the ceiling is capped at this value.

param backoff_max_retries:

The maximum number of consecutive retries before allowing the error to propagate. If None, retries indefinitely.

async announce_worker(conn)

Announce the worker to the cluster.

This will insert the worker into the workers table, or update the last_seen timestamp if the worker is already present.

Parameters:

conn (AsyncConnection) – The connection to use for the announcement.

backoff_initial

Initial backoff delay in seconds for transient errors in internal maintenance loops.

backoff_max

Maximum backoff delay in seconds for transient errors in internal maintenance loops.

backoff_max_retries

Maximum consecutive retries before propagating the error, or None for unlimited retries.

chancy

The Chancy application that the worker is associated with.

database_retry(name)

Create independent retry state using this worker’s backoff settings.

Parameters:

name (str)

Return type:

DatabaseRetry

property executors: dict[str, Executor]

Get the executors that the worker is currently using to service queues.

Returns:

A dictionary of executors keyed by queue name.

async fetch_jobs(queue, conn, *, up_to=1)

Fetch jobs from the queue for processing.

This method will fetch up to up_to jobs from the queue, mark them as running, and return them as a list of QueuedJob objects. If no jobs are available, an empty list will be returned.

It’s safe to call this method concurrently, as the jobs will be locked for the duration of the transaction.

Parameters:

• queue (Queue) – The queue to fetch jobs from.

• conn (AsyncConnection) – The database connection to use.

• up_to (int) – The maximum number of jobs to fetch.

Return type:

list[QueuedJob]

async flush()

Persist pending job updates without stopping the worker.

Updates accepted by queue_update() are written in batches until the outgoing queue is empty. This also waits for each batch’s on_job_updated plugin hooks, but does not wait for running jobs to finish. Concurrent calls, including the periodic writer and shutdown, are serialized.

Database errors propagate to the caller. If a write fails or is cancelled, its batch is retained ahead of newer updates for a later attempt. Earlier batches may already be committed.

Within each batch, database writes follow unique-key order to avoid deadlocks with concurrent pushes. Updates to the same job retain their original order, as do the plugin callbacks.

on_jobs_updated_in_transaction hooks run before commit using the update transaction; failures roll back the batch. on_job_updated hooks run only after the transaction has committed.

Updates from a claim invalidated by recovery, a new execution, or a manual retry are discarded without invoking either update hook.

await worker.queue_update(updated_job)
await worker.flush()

Return type:

None

heartbeat_poll_interval

The number of seconds between heartbeat poll intervals.

heartbeat_timeout

The number of seconds before a worker is considered to have lost connection to the cluster.

hub

An event hub with cluster and local telemetry, useful for plugins.

async increment_counter(metric_key, value, *, unit='count')

Increment a counter metric by the specified value.

This method emits a metrics.counter event that metrics plugins can subscribe to. If no metrics plugin is active, the event is simply ignored.

Parameters:

• metric_key (str) – The hierarchical key for the metric (e.g., “workflow:created”, “queue:default:throughput”)

• unit (str) – Unit shared by all observations of this metric key.

• value (float) – The value to increment the counter by

is_leader

An event that is set when the worker is the leader. This functionality is not enabled by default - a leadership plugin must be used to enable this event.

manager

The task manager for the worker, responsible for managing all the worker’s internal asyncio tasks.

async on_job_completed(*, queue, job)

Called when a job is completed by an executor.

If the queue is configured for eager polling, this will set the queue’s wake event to immediately poll for more jobs.

Parameters:

• queue (Queue)

• job (QueuedJob)

async on_signal(signum)

Called when a signal is received.

By default, handles SIGTERM and SIGINT by starting worker shutdown. If the signal is received again, it will force a shutdown.

Parameters:

signum (int)

outgoing: Queue[QueuedJob]

A queue of updates waiting to be sent to the database.

queue_change_poll_interval

How frequently to check for changes to the queues that this worker should be processing.

async queue_update(update)

Enqueue an update to a job instance.

This method will queue an update to a job instance to be processed in periodic batches, reducing the number of transactions that need to be made. You can control the frequency of these updates by setting the send_outgoing_interval attribute on the worker. Await flush() after an accepted update to persist it immediately.

Parameters:

update (QueuedJob) – The job instance to update.

Returns:

Whether the update was accepted by the worker.

Return type:

bool

queue_wake_events: dict[str, Event]

Events set whenever a job is pushed to a queue.

async record_gauge(metric_key, value, *, unit='number')

Record a gauge metric value.

This method emits a metrics.gauge event that metrics plugins can subscribe to. If no metrics plugin is active, the event is simply ignored.

Parameters:

• metric_key (str) – The hierarchical key for the metric (e.g., “workflow:active_count”, “queue:default:size”)

• unit (str) – Unit shared by all observations of this metric key.

• value (float) – The current gauge value

async record_histogram_value(metric_key, value, *, unit='number')

Record a value for a histogram metric.

This method emits a metrics.histogram event that metrics plugins can subscribe to. If no metrics plugin is active, the event is simply ignored.

Parameters:

• metric_key (str) – The hierarchical key for the metric (e.g., “workflow:execution_time”, “job:my_function:duration”)

• unit (str) – Unit shared by all observations of this metric key.

• value (float) – The value to record in the histogram

register_signal_handlers

Whether to register signal handlers on startup of the worker.

send_outgoing_interval

The number of seconds between sending outgoing updates to the database.

shutdown_event

An event that is set when the worker is shutting down.

shutdown_timeout

The number of seconds to wait for a clean shutdown before forcing the worker to stop.

async start()

Start the worker and begin processing jobs.

This method will return immediately - to wait until the worker is stopped, use the wait_for_shutdown method:

worker = Worker(chancy)
await worker.start()
await worker.wait_for_shutdown()

If possible, it’s better to use the worker as an async context manager to ensure that it’s properly cleaned up:

async with Worker(chancy) as worker:
    await worker.wait_for_shutdown()

async stop(*, timeout=None)

Stop the worker.

Stops polling for new work and gives active executors up to shutdown_timeout seconds to drain. If that deadline expires, worker maintenance tasks are cancelled and any job updates already produced are flushed before this method returns. Jobs already running in a thread or sub-interpreter cannot be forcibly interrupted and may continue in the background. Their later updates are ignored, leaving the jobs for recovery by another worker.

Returns True if the executors drained before the deadline, or False if forced teardown was required.

If final persistence fails, a later call retries it without repeating completed teardown or changing the original drain result.

Parameters:

timeout (float | None) – An optional timeout in seconds to wait for the worker to stop before forcing a shutdown. If not provided, the worker’s shutdown_timeout attribute will be used.

Return type:

bool

async wait_for_shutdown()

Wait until the worker is stopped, including its final job updates.

worker_id

The ID of the worker, which must be globally unique.

worker_tags()

Return the tags associated with the worker.

Tags are used to limit which Queues get assigned to a worker, and may also be used by some plugins.

Returns:

A set of tags associated with the worker.

Return type:

set[str]

Parameters:

• chancy (Chancy)

• worker_id (str | None)

• heartbeat_poll_interval (int)

• heartbeat_timeout (int)

• queue_change_poll_interval (int)

• shutdown_timeout (int)

• tags (set[str] | None)

• register_signal_handlers (bool)

• backoff_initial (float)

• backoff_max (float)

• backoff_max_retries (int | None)

class chancy.retry.DatabaseRetry(name, *, calculate_delay, max_retries, log)

Backoff state for one database maintenance loop.

Parameters:

• name (str)

• calculate_delay (Callable[[int], float])

• max_retries (int | None)

• log (Logger)

reset()

Record successful database work.

async wait(error)

Back off after a transient failure, or propagate an exhausted retry.

Parameters:

error (BaseException)