API reference

PULQ — fair pull-based task scheduling with WDRR.

class pulq.CommandDispatcher

Bases: object

FIFO of ManagementCommand per worker_id.

has_pending_for(worker_id)

Whether worker_id has at least one queued command.

Return type:

bool

take_next_for(worker_id)

Pop the next command for worker_id (must call has_pending_for() first).

Return type:

ManagementCommand

enqueue(worker_id, command)

Append a management command for worker_id.

Return type:

None

send(worker_id, command)

Enqueue a command built from command enum and worker_id.

Return type:

None

class pulq.CommandType(*values)

Bases: StrEnum

Management commands targeted at a worker.

STOP = 'stop'

PAUSE = 'pause'

RESUME = 'resume'

class pulq.DeficitScheduler(config)

Bases: object

Tracks WDRR deficits for fair weighted service among priority buckets.

Each scheduling epoch credits every priority with its configured weight. Work claims debit quantum from the served priority. Empty priorities are zeroed so the scheduler does not stall on idle buckets.

property is_epoch_complete: bool

True when every priority has less than one quantum of budget.

property claimable_priorities: tuple[str, ...]

Priorities that can still be served this pass, in configured order.

credit_all()

Start a new epoch: add each priority’s weight to its balance.

Return type:

None

debit(priority)

Consume one quantum from priority after a successful claim.

Return type:

None

zero_out(priority)

Clear balance when no pending work exists for priority.

Return type:

None

class pulq.DeficitSchedulerConfig(**data)

Bases: BaseModel

Validated parameters for DeficitScheduler.

priority_order defines visit order among buckets that share a claimable deficit. weights must contain exactly one positive integer weight per priority in priority_order, and no other keys.

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

priority_order: tuple[str, ...]

weights: dict[str, int]

quantum: int

classmethod weights_include_all_priorities(weights, info)

Return type:

dict[str, int]

classmethod weights_have_no_unknown_priorities(weights, info)

Return type:

dict[str, int]

class pulq.HeartbeatCallback(*args, **kwargs)

Bases: Protocol

Optional hook invoked when a worker requests work.

class pulq.InMemoryTaskRepository

Bases: object

FIFO pending queues per priority with atomic claim semantics.

async schedule(task)

Enqueue a pending task under its priority bucket.

Return type:

str

async claim_next_pending(priority, worker_id)

Pop oldest pending task for priority and mark it RUNNING.

Return type:

Task | NoPendingTask

async mark_complete(task_id, result)

Set task to COMPLETED unless result contains "ok": False.

Return type:

Task

class pulq.LocalTransport(queue)

Bases: object

In-process transport delegating to PullQueue.

async request_work(worker_id)

Delegate to pulq.core.pull_queue.PullQueue.get_next().

Return type:

Task | ManagementCommand | NoWork

async report_completion(task_id, result)

Delegate to pulq.core.pull_queue.PullQueue.mark_complete().

Return type:

None

class pulq.ManagementCommand(**data)

Bases: BaseModel

High-priority instruction for a specific worker.

type: Literal['command']

command: CommandType

worker_id: str

issued_at: datetime

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.NoPendingTask(**data)

Bases: BaseModel

Null object: no pending task was available for this priority.

type: Literal['no_pending']

priority: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.NoWork(**data)

Bases: BaseModel

Null object: pull succeeded but there is nothing to execute.

type: Literal['no_work']

reason: NoWorkReason

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.NoWorkReason(*values)

Bases: StrEnum

Why the queue returned no schedulable work.

QUEUE_EMPTY = 'queue_empty'

ALL_PRIORITIES_STARVED = 'all_priorities_starved'

class pulq.PullQueue(repository, *, priority_order, weights, quantum=1, commands=None, deficits=None, on_heartbeat=None)

Bases: object

Coordinates deficit scheduling, per-worker commands, and task claims.

property commands: CommandDispatcher

Management command queues (same instance shared if injected).

property deficits: DeficitScheduler

WDRR deficit ledger.

async schedule(task)

Enqueue task for scheduling.

Return type:

str

send_command(worker_id, command)

Queue a management command for worker_id (synchronous enqueue).

Return type:

None

async mark_complete(task_id, result)

Mark task_id complete; result follows handler conventions.

Return type:

None

async get_next(worker_id)

Pull the next management command, schedulable task, or NoWork.

Return type:

Task | ManagementCommand | NoWork

pulq.Queue

alias of PullQueue

class pulq.Task(**data)

Bases: BaseModel

A unit of work with a priority bucket for WDRR scheduling.

type: Literal['task']

id: str

priority: str

payload: dict[str, Any]

metadata: dict[str, Any]

status: TaskStatus

assigned_worker_id: str | None

created_at: datetime

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.TaskHandler(*args, **kwargs)

Bases: Protocol

User-defined async handler for Task execution.

class pulq.TaskRepository(*args, **kwargs)

Bases: Protocol

Storage backend for pending tasks (implement with DB, Redis, etc.).

async claim_next_pending(priority, worker_id)

Atomically claim the next pending task for priority, if any.

Return type:

Task | NoPendingTask

async mark_complete(task_id, result)

Mark a task completed and return the updated record.

Return type:

Task

async schedule(task)

Enqueue task; return its id.

Return type:

str

class pulq.TaskStatus(*values)

Bases: StrEnum

Lifecycle state of a task.

PENDING = 'pending'

RUNNING = 'running'

COMPLETED = 'completed'

FAILED = 'failed'

class pulq.Transport(*args, **kwargs)

Bases: Protocol

How a worker obtains work and reports completion (local, HTTP, …).

async request_work(worker_id)

Pull the next schedulable item for worker_id.

Return type:

Task | ManagementCommand | NoWork

async report_completion(task_id, result)

Persist completion for task_id.

Return type:

None

class pulq.Worker(transport, worker_id, handler, *, no_work_delay_seconds=0.0, startup=None, shutdown=None, before_process=None, after_process=None)

Bases: object

Pull loop: commands, tasks, and optional backoff on NoWork.

property worker_id: str

Stable worker identity used when pulling work.

async run()

Run until a STOP command is received.

Return type:

None

pulq.parse_claim_result(data)

Parse a mapping into a ClaimResult union using Pydantic discriminated union.

Return type:

Task | NoPendingTask

pulq.parse_work_response(data)

Parse a mapping into a WorkResponse union using Pydantic discriminated union.

Return type:

Task | ManagementCommand | NoWork

Pull queue orchestrating WDRR, management commands, and storage.

class pulq.core.pull_queue.PullQueue(repository, *, priority_order, weights, quantum=1, commands=None, deficits=None, on_heartbeat=None)

Coordinates deficit scheduling, per-worker commands, and task claims.

property commands: CommandDispatcher

Management command queues (same instance shared if injected).

property deficits: DeficitScheduler

WDRR deficit ledger.

async schedule(task)

Enqueue task for scheduling.

Return type:

str

send_command(worker_id, command)

Queue a management command for worker_id (synchronous enqueue).

Return type:

None

async mark_complete(task_id, result)

Mark task_id complete; result follows handler conventions.

Return type:

None

async get_next(worker_id)

Pull the next management command, schedulable task, or NoWork.

Return type:

Task | ManagementCommand | NoWork

Weighted Deficit Round Robin (WDRR) deficit ledger.

class pulq.core.scheduler.DeficitScheduler(config)

Tracks WDRR deficits for fair weighted service among priority buckets.

Each scheduling epoch credits every priority with its configured weight. Work claims debit quantum from the served priority. Empty priorities are zeroed so the scheduler does not stall on idle buckets.

property is_epoch_complete: bool

True when every priority has less than one quantum of budget.

property claimable_priorities: tuple[str, ...]

Priorities that can still be served this pass, in configured order.

credit_all()

Start a new epoch: add each priority’s weight to its balance.

Return type:

None

debit(priority)

Consume one quantum from priority after a successful claim.

Return type:

None

zero_out(priority)

Clear balance when no pending work exists for priority.

Return type:

None

Per-worker management command queues.

class pulq.core.dispatcher.CommandDispatcher

FIFO of ManagementCommand per worker_id.

has_pending_for(worker_id)

Whether worker_id has at least one queued command.

Return type:

bool

take_next_for(worker_id)

Pop the next command for worker_id (must call has_pending_for() first).

Return type:

ManagementCommand

enqueue(worker_id, command)

Append a management command for worker_id.

Return type:

None

send(worker_id, command)

Enqueue a command built from command enum and worker_id.

Return type:

None

Async worker that pulls tasks via a Transport.

class pulq.core.worker.Worker(transport, worker_id, handler, *, no_work_delay_seconds=0.0, startup=None, shutdown=None, before_process=None, after_process=None)

Pull loop: commands, tasks, and optional backoff on NoWork.

property worker_id: str

Stable worker identity used when pulling work.

async run()

Run until a STOP command is received.

Return type:

None

Enumerations shared by task and work-item models.

class pulq.models.enums.CommandType(*values)

Management commands targeted at a worker.

STOP = 'stop'

PAUSE = 'pause'

RESUME = 'resume'

class pulq.models.enums.NoWorkReason(*values)

Why the queue returned no schedulable work.

QUEUE_EMPTY = 'queue_empty'

ALL_PRIORITIES_STARVED = 'all_priorities_starved'

class pulq.models.enums.TaskStatus(*values)

Lifecycle state of a task.

PENDING = 'pending'

RUNNING = 'running'

COMPLETED = 'completed'

FAILED = 'failed'

Task document model.

class pulq.models.task.Task(**data)

A unit of work with a priority bucket for WDRR scheduling.

type: Literal['task']

id: str

priority: str

payload: dict[str, Any]

metadata: dict[str, Any]

status: TaskStatus

assigned_worker_id: str | None

created_at: datetime

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Management commands and null-object work responses.

class pulq.models.work.ManagementCommand(**data)

High-priority instruction for a specific worker.

type: Literal['command']

command: CommandType

worker_id: str

issued_at: datetime

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.models.work.NoPendingTask(**data)

Null object: no pending task was available for this priority.

type: Literal['no_pending']

priority: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pulq.models.work.NoWork(**data)

Null object: pull succeeded but there is nothing to execute.

type: Literal['no_work']

reason: NoWorkReason

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Discriminated union type aliases for API boundaries.

Pydantic configuration for WDRR deficit scheduling.

class pulq.models.scheduler_config.DeficitSchedulerConfig(**data)

Validated parameters for DeficitScheduler.

priority_order defines visit order among buckets that share a claimable deficit. weights must contain exactly one positive integer weight per priority in priority_order, and no other keys.

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

priority_order: tuple[str, ...]

weights: dict[str, int]

quantum: int

classmethod weights_include_all_priorities(weights, info)

Return type:

dict[str, int]

classmethod weights_have_no_unknown_priorities(weights, info)

Return type:

dict[str, int]

Parse untrusted mappings into typed work and claim results.

pulq.parsing.parse_claim_result(data)

Parse a mapping into a ClaimResult union using Pydantic discriminated union.

Return type:

Task | NoPendingTask

pulq.parsing.parse_work_response(data)

Parse a mapping into a WorkResponse union using Pydantic discriminated union.

Return type:

Task | ManagementCommand | NoWork

In-memory task repository for tests and single-process workloads.

class pulq.storage.memory.InMemoryTaskRepository

FIFO pending queues per priority with atomic claim semantics.

async schedule(task)

Enqueue a pending task under its priority bucket.

Return type:

str

async claim_next_pending(priority, worker_id)

Pop oldest pending task for priority and mark it RUNNING.

Return type:

Task | NoPendingTask

async mark_complete(task_id, result)

Set task to COMPLETED unless result contains "ok": False.

Return type:

Task

In-process transport: direct calls into PullQueue.

class pulq.transport.local.LocalTransport(queue)

In-process transport delegating to PullQueue.

async request_work(worker_id)

Delegate to pulq.core.pull_queue.PullQueue.get_next().

Return type:

Task | ManagementCommand | NoWork

async report_completion(task_id, result)

Delegate to pulq.core.pull_queue.PullQueue.mark_complete().

Return type:

None

Protocols and callback types.

class pulq.types.HeartbeatCallback(*args, **kwargs)

Optional hook invoked when a worker requests work.

class pulq.types.TaskHandler(*args, **kwargs)

User-defined async handler for Task execution.

class pulq.types.TaskRepository(*args, **kwargs)

Storage backend for pending tasks (implement with DB, Redis, etc.).

async claim_next_pending(priority, worker_id)

Atomically claim the next pending task for priority, if any.

Return type:

Task | NoPendingTask

async mark_complete(task_id, result)

Mark a task completed and return the updated record.

Return type:

Task

async schedule(task)

Enqueue task; return its id.

Return type:

str

class pulq.types.Transport(*args, **kwargs)

How a worker obtains work and reports completion (local, HTTP, …).

async request_work(worker_id)

Pull the next schedulable item for worker_id.

Return type:

Task | ManagementCommand | NoWork

async report_completion(task_id, result)

Persist completion for task_id.

Return type:

None