Queue

The central class for creating and managing a task queue.

Sub-pages

The Queue API is split across several pages for readability:

Job Management — get, list, cancel, archive, replay jobs
Queue & Stats — rate limits, concurrency, pause/resume, statistics, dead letters
Workers & Hooks — run workers, lifecycle hooks, circuit breakers, async methods
Resources & Locking — resource system, distributed locks
Events & Logs — event callbacks, webhooks, structured logs
Pub/Sub — topic subscriptions, fan-out publish

Constructor

python

Queue(
    db_path: str = ".flexiq/flexiq.db",
    workers: int = 0,
    default_retry: int = 3,
    default_timeout: int = 300,
    default_priority: int = 0,
    result_ttl: int | None = None,
    retention: Retention | None = None,
    middleware: list[TaskMiddleware] | None = None,
    drain_timeout: int = 30,
    interception: str = "off",
    max_intercept_depth: int = 10,
    recipe_signing_key: str | None = None,
    max_reconstruction_timeout: int = 10,
    file_path_allowlist: list[str] | None = None,
    disabled_proxies: Sequence[BuiltInProxy | str] | None = None,
    async_concurrency: int = 100,
    event_workers: int = 4,
    scheduler_poll_interval_ms: int = 50,
    scheduler_reap_interval: int = 100,
    scheduler_cleanup_interval: int = 1200,
    namespace: str | None = None,
)

Parameter	Type	Default	Description
`db_path`	`str`	`".flexiq/flexiq.db"`	Path to SQLite database file. Parent directories are created automatically.
`workers`	`int`	`0`	Number of worker threads (`0` = auto-detect CPU count)
`default_retry`	`int`	`3`	Default max retry attempts for tasks
`default_timeout`	`int`	`300`	Default task timeout in seconds
`default_priority`	`int`	`0`	Default task priority (higher = more urgent)
`result_ttl`	`int \| None`	`None`	Legacy uniform auto-cleanup window in seconds, applied to every retention table. Superseded by `retention`.
`retention`	`Retention \| None`	`None`	Per-table retention windows. Retention is on by default — leaving both `result_ttl` and `retention` unset applies the recommended windows (`archived_jobs`/`task_metrics`/`job_errors` 7d, `task_logs` 3d, `dead_letter` 30d). Pass an empty `Retention()` to disable the table-wide windows (per-entry `result_ttl` is still honored). See Retention & auto-cleanup.
`middleware`	`list[TaskMiddleware] \| None`	`None`	Queue-level middleware applied to all tasks.
`drain_timeout`	`int`	`30`	Seconds to wait for in-flight tasks during graceful shutdown.
`interception`	`str`	`"off"`	Argument interception mode: `"strict"`, `"lenient"`, or `"off"`. See Resource System.
`max_intercept_depth`	`int`	`10`	Max recursion depth for argument walking.
`recipe_signing_key`	`str \| None`	`None`	HMAC-SHA256 key for proxy recipe integrity. Falls back to `FLEXIQ_RECIPE_SECRET` env var.
`max_reconstruction_timeout`	`int`	`10`	Max seconds allowed for proxy reconstruction.
`file_path_allowlist`	`list[str] \| None`	`None`	Allowed file path prefixes for the file proxy handler.
`disabled_proxies`	`Sequence[BuiltInProxy \| str] \| None`	`None`	Built-in proxy handlers to skip registering. An unknown id raises rather than silently leaving the handler registered.
`async_concurrency`	`int`	`100`	Maximum number of `async def` tasks running concurrently on the native async executor.
`event_workers`	`int`	`4`	Thread pool size for the event bus. Increase for high event volume.
`scheduler_poll_interval_ms`	`int`	`50`	Milliseconds between scheduler poll cycles. Lower values improve scheduling precision at the cost of CPU.
`scheduler_reap_interval`	`int`	`100`	Reap stale/timed-out jobs every N poll cycles.
`scheduler_cleanup_interval`	`int`	`1200`	Clean up old completed jobs every N poll cycles.
`namespace`	`str \| None`	`None`	Namespace for multi-tenant isolation. Jobs enqueued on this queue carry this namespace; workers only dequeue matching jobs. `None` means no namespace (default).

Parameter	Type	Default	Description
`name`	`str \| None`	Auto-generated	Explicit task name. Defaults to `module.qualname`.
`max_retries`	`int`	`3`	Max retry attempts before moving to DLQ.
`retry_backoff`	`float`	`1.0`	Base delay in seconds for exponential backoff.
`retry_delays`	`list[float] \| None`	`None`	Per-attempt delays in seconds, overrides backoff. e.g. `[1, 5, 30]`.
`max_retry_delay`	`int \| None`	`None`	Cap on backoff delay in seconds. Defaults to 300 s.
`timeout`	`int`	`300`	Hard execution time limit in seconds.
`soft_timeout`	`float \| None`	`None`	Cooperative time limit checked via `current_job.check_timeout()`.
`expires`	`float \| None`	`None`	Default expiry in seconds — jobs not started within the window are skipped. Per-call `apply_async(expires=)` overrides.
`priority`	`int`	`0`	Default priority (higher = more urgent).
`rate_limit`	`str \| None`	`None`	Rate limit string, e.g. `"100/m"`.
`queue`	`str`	`"default"`	Named queue to submit to.
`circuit_breaker`	`dict \| None`	`None`	Circuit breaker config: `{"threshold": 5, "window": 60, "cooldown": 120}`.
`middleware`	`list[TaskMiddleware] \| None`	`None`	Per-task middleware, applied in addition to queue-level middleware.
`inject`	`list[str] \| None`	`None`	Resource names to inject as keyword arguments. See Resource System.
`serializer`	`Serializer \| None`	`None`	Per-task serializer override. Falls back to queue-level serializer.
`max_concurrent`	`int \| None`	`None`	Max concurrent running instances, across the cluster. `None` = no limit.
`max_in_flight_per_task`	`int \| None`	`None`	Cap on this task's share of a single worker's dispatch slots, so one slow task can't occupy the whole pool and starve the others. In-process and free, unlike `max_concurrent`, which is cluster-wide and costs a database read. `None` = the task may use the whole pool.
`retry_budget`	`str \| None`	`None`	Cap on how fast this task may retry, across all of its jobs — same syntax as `rate_limit`, e.g. `"100/m"`. Once spent, further failures dead-letter instead of retrying (DLQ `metadata` is set to `retry_budget_exhausted`). Distinct from `max_retries`, which bounds one job rather than the rate, and from `circuit_breaker`, which trips on hard failure rather than aggregate retry rate. `None` = no cap.
`idempotent`	`bool`	`False`	Auto-derive a dedup key from `sha256(task_name\|payload)` so duplicate `.delay()`/`.apply_async()` calls while a job is pending or running return the same job ID.
`compensates`	`TaskWrapper \| str \| None`	`None`	Saga compensator for this task, enqueued in reverse order if a later workflow step fails. See Sagas.
`batch`	`bool \| dict \| None`	`None`	Enable producer-side batching (`True` for defaults, or a dict with `max_size`/`max_wait_ms`). See Batching.
`predicate`	`Predicate \| Callable \| None`	`None`	Gate execution on a runtime condition, checked at enqueue time and worker-dispatch time. See Predicates.
`on_false`	`str`	`"defer"`	What to do when `predicate` returns `False` — `"defer"` (re-schedule) or `"cancel"` (terminally skip).

predicate_extras and default_defer_seconds fine-tune predicate/on_false — see Predicates. Per-call idempotency overrides (idempotency_key, per-call idempotent=) are covered in Idempotency.

`queue.autodiscover()`

python

queue.autodiscover(package: str = "tasks") -> list[str]

Import package and every module beneath it, then register every task declared with the queue-less @flexiq.task() decorator. Returns the sorted names now registered on this queue.

Parameter	Type	Default	Description
`package`	`str`	`"tasks"`	Dotted path to the package (or single module) holding the declarations.

python

from flexiq import Queue

queue = Queue(db_path="app.db")
queue.autodiscover("myapp.tasks")  # ['myapp.tasks.invoices.send_invoice', ...]

Idempotent — a second call returns the same list rather than an empty one, and a second Queue in the same process receives the same tasks. The registry is also drained at Queue(...) construction and at run_worker(), so a worker entrypoint that imports its task modules directly never has to call this.

Raises ImportError if package or anything beneath it fails to import (never swallowed), and DuplicateTaskError if a discovered task claims a name this queue already registered for a different function. See Task discovery.

`@queue.periodic()`

python

@queue.periodic(
    cron: str,
    name: str | None = None,
    args: tuple = (),
    kwargs: dict | None = None,
    queue: str = "default",
    timezone: str | None = None,
) -> TaskWrapper

Parameter	Type	Default	Description
`cron`	`str`	—	6-field cron expression (seconds precision).
`name`	`str \| None`	Auto-generated	Explicit task name.
`args`	`tuple`	`()`	Positional arguments passed to the task on each run.
`kwargs`	`dict \| None`	`None`	Keyword arguments passed to the task on each run.
`queue`	`str`	`"default"`	Named queue to submit to.
`timezone`	`str \| None`	`None`	IANA timezone name (e.g. `"America/New_York"`). Defaults to UTC.

Periodic catalog

Schedules reach storage when run_worker() starts; these act on what is there, so they outlive the decorator that declared them.

Method	Returns	Description
`queue.list_periodic()`	`list[PeriodicInfo]`	Every registered schedule, enabled or paused.
`queue.pause_periodic(name)`	`bool`	Stop a schedule firing, keeping the registration. `False` if unknown.
`queue.resume_periodic(name)`	`bool`	Resume a paused schedule. `False` if unknown.
`queue.delete_periodic(name)`	`bool`	Unschedule. `False` if unknown.

PeriodicInfo fields: name, task_name, cron_expr, queue, enabled, last_run (None until first fire), next_run, timezone. Timestamps are Unix milliseconds. Async forms: alist_periodic, apause_periodic, aresume_periodic, adelete_periodic.

Enqueue methods

`queue.enqueue()`

python

queue.enqueue(
    task_name: str,
    args: tuple = (),
    kwargs: dict | None = None,
    priority: int | None = None,
    delay: float | None = None,
    queue: str | None = None,
    max_retries: int | None = None,
    timeout: int | None = None,
    unique_key: str | None = None,
    metadata: str | None = None,
    notes: dict[str, Any] | None = None,
    depends_on: str | list[str] | None = None,
) -> JobResult

Enqueue a task for execution. Returns a JobResult handle.

Parameter	Type	Default	Description
`metadata`	`str \| None`	`None`	Free-form JSON string blob. No size or shape constraint.
`notes`	`dict \| None`	`None`	Structured annotations, max 15 top-level keys. See Structured notes.
`depends_on`	`str \| list[str] \| None`	`None`	Job ID(s) this job depends on. See Dependencies.

`queue.enqueue_many()`

python

queue.enqueue_many(
    task_name: str,
    args_list: list[tuple],
    kwargs_list: list[dict] | None = None,
    priority: int | None = None,
    queue: str | None = None,
    max_retries: int | None = None,
    timeout: int | None = None,
    delay: float | None = None,
    delay_list: list[float | None] | None = None,
    unique_keys: list[str | None] | None = None,
    metadata: str | None = None,
    metadata_list: list[str | None] | None = None,
    notes: dict[str, Any] | None = None,
    notes_list: list[dict[str, Any] | None] | None = None,
    expires: float | None = None,
    expires_list: list[float | None] | None = None,
    result_ttl: int | None = None,
    result_ttl_list: list[int | None] | None = None,
) -> list[JobResult]

Enqueue multiple jobs in a single transaction for high throughput. Supports both uniform parameters (applied to all jobs) and per-job lists.

Parameter	Type	Default	Description
`delay`	`float \| None`	`None`	Uniform delay in seconds for all jobs
`delay_list`	`list[float \| None] \| None`	`None`	Per-job delays in seconds
`unique_keys`	`list[str \| None] \| None`	`None`	Per-job deduplication keys
`metadata`	`str \| None`	`None`	Uniform metadata JSON for all jobs
`metadata_list`	`list[str \| None] \| None`	`None`	Per-job metadata JSON
`notes`	`dict \| None`	`None`	Uniform structured notes for all jobs (≤ 15 keys)
`notes_list`	`list[dict \| None] \| None`	`None`	Per-job structured notes (takes precedence over `notes`)
`expires`	`float \| None`	`None`	Uniform expiry in seconds for all jobs
`expires_list`	`list[float \| None] \| None`	`None`	Per-job expiry in seconds
`result_ttl`	`int \| None`	`None`	Uniform result TTL in seconds
`result_ttl_list`	`list[int \| None] \| None`	`None`	Per-job result TTL in seconds

Per-job lists (*_list) take precedence over uniform values when both are provided.

Queue

Constructor

Task registration

`@queue.task()`

`queue.autodiscover()`

`@queue.periodic()`

Periodic catalog

Enqueue methods

`queue.enqueue()`

`queue.enqueue_many()`

Constructor

Task registration

`@queue.task()`

`queue.autodiscover()`

`@queue.periodic()`

Periodic catalog

Enqueue methods

`queue.enqueue()`

`queue.enqueue_many()`