Quebec

Quebec is a simple background task queue for processing asynchronous tasks. The name is derived from the NATO phonetic alphabet for "Q", representing "Queue".

This project is inspired by Solid Queue.

Why Quebec?

• Simplified Architecture: No dependencies on Redis or message queues
• Database-Powered: Leverages RDBMS capabilities for complex task queries and management
• Rust Implementation: High performance and safety with Python compatibility
• Framework Agnostic: Works with asyncio, Trio, threading, SQLAlchemy, Django, FastAPI, etc.

Features

• Scheduled tasks
• Recurring tasks
• Concurrency control
• Per-queue concurrency limits
• Rate limiting
• Exclusive (stop-the-world) jobs
• Multi-process (fork) mode
• Memory-based worker recycling
• Web dashboard
• Automatic retries
• Signal handling & graceful restart
• Lifecycle hooks

Control Plane

Built-in web dashboard for monitoring jobs, queues, and workers in real-time.

Database Support

• SQLite
• PostgreSQL
• MySQL

Upgrading an existing PostgreSQL deployment

On PostgreSQL only, Quebec no longer creates the (key, value) and (expires_at) indexes on the semaphores table. They prevent HOT updates: every concurrency wait/signal rewrites both value and expires_at, and indexing a column that every UPDATE touches forces a new row version plus an index write each time. Dead tuples then pile up faster than autovacuum can reclaim them, and a table holding a handful of live rows can end up costing thousands of buffer hits per statement.

Databases created before this version still carry both indexes.

create_tables() does not drop indexes, so run this once against an existing database (substitute your table prefix):

DROP INDEX IF EXISTS idx_solid_queue_semaphores_key_value;
DROP INDEX IF EXISTS idx_solid_queue_semaphores_expires_at;

ALTER TABLE solid_queue_semaphores SET (
  fillfactor = 70,
  autovacuum_vacuum_scale_factor = 0,
  autovacuum_vacuum_threshold = 1000
);

-- Reclaims space already lost to bloat and applies the new fillfactor.
-- Takes an ACCESS EXCLUSIVE lock, so concurrency operations block for its
-- duration -- normally well under a second on a healthy table.
VACUUM FULL solid_queue_semaphores;

The unique index on key stays: it serves the only hot-path lookup (WHERE key = $1), and PostgreSQL applies the value predicate as a filter after it. Calling create_tables() afterwards re-applies the storage parameters but never re-creates the dropped indexes.

The expires_at index only served delete_expired, which the dispatcher runs once per concurrency_maintenance_interval (default 600s). That scan is a sequential one now -- roughly 11 ms on a 50k-row table, against an index that would otherwise be maintained on every write.

SQLite and MySQL keep both indexes and need no migration. InnoDB's undo-log MVCC and SQLite's rollback journal do not accumulate heap bloat this way, so there the indexes are a plain win -- delete_expired in particular gets to use (expires_at) instead of scanning.

Quick Start

Module Runner (Recommended)

Define jobs in a package:

# jobs/email_job.py
import quebec

class EmailJob(quebec.BaseClass):
    queue_as = "default"

    def perform(self, to, subject):
        self.logger.info(f"Sending email to {to}: {subject}")

Export them in __init__.py:

# jobs/__init__.py
from .email_job import EmailJob

Run with python -m quebec:

DATABASE_URL=sqlite:///demo.db?mode=rwc python -m quebec jobs

All configuration via QUEBEC_* environment variables — no boilerplate entry script needed.

Script Mode

For more control, use Quebec directly in a script:

import logging
from pathlib import Path
from quebec.logger import setup_logging

setup_logging(level=logging.DEBUG)

import quebec

db_path = Path('demo.db')
qc = quebec.Quebec(f'sqlite://{db_path}?mode=rwc')


@qc.register_job
class FakeJob(quebec.BaseClass):
    def perform(self, *args, **kwargs):
        self.logger.info(f"Processing job {self.id}: args={args}, kwargs={kwargs}")


if __name__ == "__main__":
    # Enqueue a job (qc is inferred from @qc.register_job)
    FakeJob.perform_later(123, foo='bar')

    # Start Quebec (handles signal, spawns workers, runs main loop)
    qc.run(
        create_tables=not db_path.exists(),
        control_plane='127.0.0.1:5006',  # Optional: web dashboard
    )

Or run the quickstart script directly:

curl -O https://raw.githubusercontent.com/ratazzi/quebec/refs/heads/master/quickstart.py
uv run quickstart.py

Auto-Discovering Jobs

If your jobs are organized in a package (e.g. app.jobs.*), call Quebec.discover_jobs() instead of decorating each class with @qc.register_job or calling qc.register_job_class(...) one by one:

# app/jobs/cleanup.py
class CleanupJob(quebec.BaseClass):
    def perform(self, *args, **kwargs): ...

# main.py
qc = quebec.Quebec(dsn)
qc.discover_jobs("app.jobs", "worker.tasks")   # recursively scans each
qc.run()

discover_jobs takes one or more dotted package paths as positional arguments (varargs) — no need to wrap a single package in a list.

discover_jobs(*packages, recursive=True, on_error="raise"):

• Registers every BaseClass subclass whose __module__ falls under one of the given packages. Classes imported from elsewhere (e.g. from some.lib import JobMixin) are ignored.
• Raises ValueError if two discovered classes share the same __qualname__, since Quebec's worker registry is keyed by qualname and the later registration would otherwise silently replace the earlier one.
• on_error="raise" (default) propagates submodule ImportError. Pass on_error="warn" to emit a RuntimeWarning and keep scanning — useful when a package contains optional-integration modules that may fail to import in some environments. The top-level package is always imported strictly.

Multiple Quebec Instances

Quebec is designed for one instance per process. Registering a job class (via @qc.register_job, qc.register_job_class, or qc.discover_jobs) binds it to that Quebec instance, so MyJob.perform_later(...) shorthand routes to the binding. If a process holds more than one Quebec instance and registers the same job class to each, the most recent registration wins — pass the target instance explicitly to disambiguate:

MyJob.perform_later(qc2, arg1)                  # route to qc2
MyJob.set(queue='critical').perform_later(qc2, arg1)

qc.run() Options

Recommended: configure worker thread count in queue.yml via workers.threads. If you need a one-off override, Quebec(..., worker_threads=3) is also supported.

Multi-Process Mode (fork supervisor)

By default qc.run() runs all components as threads in a single process. To scale across CPU cores, set QUEBEC_SUPERVISOR=1 to fork a pool of child processes instead:

QUEBEC_SUPERVISOR=1 python -m quebec your.jobs

# queue.yml (under your environment, e.g. production:)
workers:
  - queues: "*"
    threads: 5
    processes: 4        # fork 4 worker processes
dispatchers:
  - polling_interval: 1
    processes: 1        # fork 1 dispatcher process

The supervisor forks workers[].processes worker children and dispatchers[].processes dispatcher children, each taking its config from the matching yml entry, and reforks any child that dies (matching Solid Queue's process model). Fork mode is opt-in via the env var so an existing config with processes set doesn't silently switch process model on upgrade; spawn is ignored in this mode. Outside supervisor mode the processes keys are ignored and Quebec uses the single-process threaded runtime.

Force Queue Override (multi-branch development)

Set QUEBEC_FORCE_OVERRIDE_QUEUE to pin every enqueue and consumption to one queue — handy when several development branches share a single database:

QUEBEC_FORCE_OVERRIDE_QUEUE=branch_x python -m quebec your.jobs

Every enqueue path rewrites queue_name to this value (ignoring whatever the class, call site, or scheduler specified), and the worker only consumes that queue — so jobs enqueued by one branch are never picked up by another. URL-hostile characters and * in the name are sanitized to - (a literal * would otherwise be reinterpreted as a wildcard by the consuming worker).

Transactional Enqueue

This is the deliberate cost of keeping the engine fully decoupled from your ORM and connection — the upside is that Quebec drags no Python database dependencies into your app, but it means the enqueue cannot join your transaction. Two failure windows follow:

• The business transaction commits but the enqueue fails → the job is lost.
• The enqueue commits but the business transaction rolls back → the job runs against missing or stale data.

Recommendations:

• Enqueue after your business transaction commits. This removes the worse direction — a job running for a write that was rolled back.
• Make jobs idempotent and tolerant of data that may not be visible yet; lean on retries.
• If you genuinely need atomicity, use a transactional outbox: write an outbox row inside your own transaction (business + outbox commit atomically), then relay it into a real job (at-least-once delivery).

Delayed Jobs

from datetime import timedelta

# Run after 1 hour
FakeJob.set(wait=3600).perform_later(arg1)

# Run at specific time
FakeJob.set(wait_until=tomorrow_9am).perform_later(arg1)

# Override queue and priority
FakeJob.set(queue='critical', priority=1).perform_later(arg1)

Automatic Retries

from datetime import timedelta

class PaymentJob(quebec.BaseClass):
    retry_on = [
        quebec.RetryStrategy(
            (ConnectionError, TimeoutError),
            wait=timedelta(seconds=30),
            attempts=3,
        ),
        quebec.RetryStrategy(
            (ValueError,),
            wait=timedelta(seconds=5),
            attempts=1,
            # Called once retries are exhausted; receives (job, error).
            handler=lambda job, error: notify_admin(error),
        ),
    ]

    def perform(self, order_id):
        process_payment(order_id)

Multiple RetryStrategy entries can target different exception types with independent wait/attempts. The optional handler fires only when a strategy's attempts are exhausted (mirroring ActiveJob's retry_on ... do |job, error| block) and is called with (job, error). discard_on and rescue_from handlers use the same (job, error) signature.

Concurrency Control

Limit how many jobs with the same key can run simultaneously:

class ReportJob(quebec.BaseClass):
    concurrency_limit = 3          # max 3 concurrent executions per key
    concurrency_duration = 120     # semaphore TTL in seconds

    def concurrency_key(self, account_id, **kwargs):
        return str(account_id)     # final key: "ReportJob/123"

    def perform(self, account_id):
        generate_report(account_id)

The actual concurrency key is "ClassName/key" (e.g. "ReportJob/123"), so different job classes never conflict. When the limit is reached, new jobs are blocked until a slot becomes available. The concurrency_duration acts as a safety TTL — the semaphore is released automatically if a worker crashes.

Rate Limiting (experimental)

Cap how many jobs run within a sliding time window, scoped per key:

from datetime import timedelta

class ApiCallJob(quebec.BaseClass):
    rate_limit_max = 5                          # at most 5 runs...
    rate_limit_duration = timedelta(seconds=2)  # ...per rolling 2-second window
    rate_limit_on_throttle = quebec.RateLimitConflict.Reschedule  # default

    def rate_limit_key(self, region="us", **kwargs):
        return region                           # bucket key: "ApiCallJob/us"

    def perform(self, region="us"):
        call_external_api(region)

Like concurrency control, the bucket is "ClassName/key", and rate_limit_key defaults to the class name when not overridden. rate_limit_duration must be a datetime.timedelta of at least one second. When the window is exhausted, rate_limit_on_throttle decides what happens: Reschedule (the default) pushes the job to a later run, while Discard drops it.

Exclusive Jobs

Let an occasional memory-heavy job own the whole worker process while it runs:

class RebuildSearchIndexJob(quebec.BaseClass):
    exclusive = True

    def perform(self):
        rebuild_index()                         # runs alone on this worker

When an exclusive job is claimed, the worker stops claiming new jobs, waits for any in-flight siblings to finish, then runs the exclusive job by itself before resuming normal claiming. The scope is the current worker process — it does not coordinate across separate worker processes; pair it with concurrency_limit = 1 and a concurrency_key if you also need cluster-wide single-instance execution.

Graceful Restart (quiet-then-exit)

Drain in-flight work and exit on a quiet signal, for zero-downtime rolling restarts:

qc = quebec.Quebec(database_url="...", quiet_then_exit=True)
qc.run()

Sending SIGUSR1 (or SIGTSTP) puts the worker into quiet mode: it stops claiming new jobs but keeps running until every in-flight job finishes, then exits cleanly — with no time limit (unlike the SIGTERM path, which is bounded by shutdown_timeout). The usual flow is: signal the old instance quiet, start a new instance, and the old one exits once drained. Opt-in (default off), and standalone-only — under the fork supervisor a self-exited child would just be reforked, so use a supervisor-level rolling restart there instead. Also settable via QUEBEC_QUIET_THEN_EXIT=1.

Memory-Based Worker Recycling

Long-lived Python workers tend to hold onto RSS the interpreter never returns to the OS. Quebec can recycle a bloated worker by draining it and exiting with a dedicated code, leaving the actual restart to your process supervisor. It is configured by environment variables — there is no in-process restart:

QUEBEC_WORKER_MAX_RSS_MB=512                  # soft limit; unset = disabled
QUEBEC_WORKER_MEMORY_RECYCLE_CONFIRMATIONS=3  # consecutive over-limit samples before recycling (default)
QUEBEC_WORKER_MEMORY_CHECK_INTERVAL=5s        # how often RSS is sampled (default)

When a worker's RSS stays above the limit for that many consecutive samples, it enters quiet mode, stops claiming, drains its in-flight jobs (no time limit), and exits with code 75 — the planned-recycle code. The supervisor then relaunches a fresh process. Under the built-in fork supervisor (QUEBEC_SUPERVISOR=1) this refork is automatic; under systemd, Restart=on-failure relaunches the worker after the non-zero recycle exit:

# /etc/systemd/system/quebec-worker.service
[Service]
ExecStart=/usr/bin/python -m quebec your.jobs
Environment=QUEBEC_DATABASE_URL=postgresql://localhost/myapp
Environment=QUEBEC_WORKER_MAX_RSS_MB=512
Restart=on-failure

[Install]
WantedBy=multi-user.target

Exit code 75 is non-zero, so Restart=on-failure treats the planned recycle as a failure and relaunches the worker. If you'd rather not have planned recycles show up as failures (in systemctl status or the start-limit counter), add SuccessExitStatus=75 together with RestartForceExitStatus=75 — the former keeps 75 out of the failure tally, the latter still forces the restart.

Per-Job Memory Metrics (Linux)

Quebec observes two different Linux signals during perform():

• minor_faults / major_faults are native-thread activity counters from getrusage(RUSAGE_THREAD). They are useful when investigating allocation and I/O behaviour, but are not converted to bytes and are not RSS.
• Process RSS is read at job start and end and sampled every 100 ms in between. This produces process_rss_start, process_rss_peak, process_rss_end, and process_rss_peak_delta. Shorter-lived peaks may fall between samples.

RSS belongs to the process, not a thread. Quebec marks a window process_rss_single_job=true only in a supervisor-managed worker where either threads: 1 or the job is exclusive. Only those single-job windows enter the per-class RSS aggregate. Other windows remain useful as process context but are not presented as memory attributable to one job. Allocations in subprocesses are not included in the worker's RSS. Even a single-job window is a sampled process envelope: allocator reuse and worker-runtime activity can still affect it.

These are observability metrics, not enforcement. Use a separate cgroup per worker process with memory.high / memory.max when one job must not exhaust the host.

The observations appear on every job.completed log line and on execution.metric. For offline analysis, record one CSV row per finished job:

kill -USR2 <worker pid>   # start recording; send again to stop

or from code: qc.start_job_metrics(path=None), qc.stop_job_metrics(), qc.toggle_job_metrics(), qc.job_metrics_path. Under the fork supervisor the signal is forwarded to every worker child, and each child writes its own file. Columns:

ts_ms,pid,tid,jid,class,queue,status,duration_ms,minor_faults,major_faults,process_rss_start_kb,process_rss_peak_kb,process_rss_end_kb,process_rss_peak_delta_kb,process_rss_single_job,active_jobs

active_jobs shows how many jobs the process owned when the row was recorded. Aggregate attributable samples with whatever reads CSV, e.g.

select class, count(*), max(process_rss_peak_delta_kb),
       quantile_cont(process_rss_peak_delta_kb, 0.95)
from 'quebec-job-metrics-*.csv'
where process_rss_single_job = true
group by class order by 3 desc;

Environment variables:

QUEBEC_JOB_METRICS_DIR=/var/log/quebec   # output dir for SIGUSR2 recordings (default: OS temp dir)
QUEBEC_JOB_METRICS_MAX_ROWS=100000       # recording stops itself after this many rows
QUEBEC_JOB_METRICS_MAX_SECONDS=3600      # ...or after this long

Each Quebec instance also keeps per-class aggregates since startup: count, failures, duration, thread faults, and average / p50 / p95 / max of single-job process_rss_peak_delta_kb samples with the jid of the largest job. qc.job_metrics_summary(reset=False) returns them as a dict; reset=True takes and clears the current snapshot atomically. qc.log_job_metrics_summary() writes one job_metrics.summary log line per class, and stopping a recording with SIGUSR2 logs them too. Percentiles come from a log2 histogram, so they are bucket upper bounds rather than exact values.

Rows are handed to a writer thread through a bounded queue and flushed every 5 seconds. If the writer falls behind, rows are dropped rather than blocking jobs. When a row/time limit automatically ends a recording, writer draining and file flush happen on a background reaper instead of the job completion path.

USDT probes. The Linux extension module carries quebec:job_start and quebec:job_end. They are a single nop until a tracer attaches. The end probe exports the minor-fault delta and the sampled RSS peak delta; the RSS argument is -1 unless process_rss_single_job is true. job_start exports jid, class, and queue as pointer/length pairs. job_end exports jid, class, success, duration nanoseconds, minor faults, and the attributable RSS peak delta.

Per-Queue Concurrency (experimental)

Cap how many jobs run concurrently across the cluster for specific queues, independent of per-class concurrency_key:

qc = quebec.Quebec(
    database_url="...",
    experimental_queue_concurrency={"reports": 2, "exports": 1},
)
qc.run()

Each listed queue acquires a queue:<name> semaphore at claim time; queues not present are unlimited. Useful for isolating a misbehaving queue during remediation. Naming and semantics are experimental and may change.

Global Priority Across Queues (experimental, off by default)

With queues: "*" and nothing to skip, a worker polls with a single unfiltered query, so priority orders jobs across every queue. That stops being possible as soon as a queue must be skipped — paused, or with a full experimental_queue_concurrency slot — because excluding queues with NOT IN cannot use an index. Quebec then does what Solid Queue does: one query per live queue, which makes queue order override priority until the queue is resumed.

Enabling this flag keeps one global order in that situation by polling with an IN list instead:

SELECT * FROM solid_queue_ready_executions
WHERE queue_name IN ($live_queues)
ORDER BY priority, job_id
LIMIT $batch
FOR UPDATE SKIP LOCKED;

qc = quebec.Quebec(
    "postgresql://localhost/myapp",
    experimental_global_priority=True,   # or QUEBEC_EXPERIMENTAL_GLOBAL_PRIORITY=true
)

Measure before enabling — this is not universally faster. It moves the scan from the excluded side to the live side, and which one wins depends on where your backlog sits:

PostgreSQL picks between two plans for the IN form, and the choice depends on statistics: walk (queue_name, priority, job_id) per listed queue and sort the union, or walk (priority, job_id) and filter. Check which one you get on real data:

EXPLAIN (ANALYZE, BUFFERS)
SELECT * FROM solid_queue_ready_executions
WHERE queue_name IN ('live_a', 'live_b')
ORDER BY priority, job_id
LIMIT 10
FOR UPDATE SKIP LOCKED;

Look at whether the queue_name index is used, how many rows feed the Sort, Rows Removed by Filter, and shared buffers. Note that LockRows sits above Sort, so SKIP LOCKED does not shrink the sort input.

Two things it does not change:

• It only ever applies to *. Explicit queue lists (["real_time", "background"]) and wildcard prefixes ("beta*") keep Solid Queue's contract that queue order takes precedence over priority.
• Ordering and locking stay in one statement, so there is no window where the order is decided from rows another worker has already taken.

Pausing Recurring Tasks (opt-in)

Recurring tasks can be paused and resumed at runtime — from Python or from the control plane's Recurring Jobs page — without editing recurring.yml or restarting the scheduler:

qc = quebec.Quebec(db_url, recurring_pause=True)   # or QUEBEC_RECURRING_PAUSE=true

qc.pause_recurring("nightly_report")      # True; False if it was already paused
qc.recurring_paused("nightly_report")     # True
qc.paused_recurring_tasks()               # ["nightly_report"]
qc.resume_recurring("nightly_report")     # True; False if it was not paused

While paused, the scheduler skips each occurrence (nothing is enqueued, no recurring_executions row is written) and moves on to the next one. Resuming does not replay the skipped runs; the next occurrence after the resume fires as scheduled. run_recurring_now() still works on a paused task. Unknown keys raise LookupError — static tasks appear in the table once a scheduler has started.

Solid Queue has no such state, so this is the one place Quebec extends its schema: enabling recurring_pause adds a nullable paused_at column to the recurring tasks table. It is added automatically by create_tables() and when a scheduler starts, and each process checks for it on its own. It is off by default so an unmodified Solid Queue database keeps working exactly as before; with it off, the column is not added even if it exists elsewhere, and the pause API raises RuntimeError.

Sharing the database with a Rails app:

• Solid Queue reads and writes the table as usual and leaves paused_at alone, with one exception: its scheduler upserts every attribute of the static tasks when it boots, which resets paused_at. Solid Queue's scheduler also ignores the pause, so pausing only takes effect when Quebec runs the scheduler.

• If the connecting role is not allowed to ALTER TABLE, Quebec logs a warning and pausing is unavailable in that process until the column exists. Add it yourself in that case:

  add_column :solid_queue_recurring_tasks, :paused_at, :datetime
  

  No restart is needed: a process whose attempt failed keeps looking for the column (a catalog lookup, at most every 5 seconds) and starts honouring pauses as soon as it appears. Explicit calls — create_tables(), the pause API, the control-plane buttons — retry the ALTER right away.

TLS Configuration (PostgreSQL)

Quebec links sqlx against rustls + webpki-roots. Public CAs (AWS RDS, Neon, Google Cloud SQL, Supabase, etc.) are trusted out of the box — no OS trust store is consulted.

Pass libpq-style SSL options as Quebec(...) kwargs, as DSN query params, or via QUEBEC_SSL* environment variables:

qc = quebec.Quebec(
    "postgresql://user:pass@host:5432/db",
    sslmode="verify-full",             # or QUEBEC_SSLMODE
    sslrootcert="/etc/ssl/certs/ca.pem",  # internal CAs only
)

Priority is kwargs > env > DSN query. Passing any ssl* kwarg/env against a non-postgres URL raises ValueError.

For public CAs, verify-full works zero-config. Use sslrootcert for internal/self-signed CAs. sslcert + sslkey enable client certificate (mTLS) auth.

sslmode=allow is rejected with a ValueError. Upstream sqlx-postgres 0.8 treats allow identically to disable (plaintext, marked FIXME in the driver); to avoid a silent downgrade, Quebec refuses it. Use prefer for opportunistic TLS, or require/verify-* to enforce it.

Note: some managed Postgres services (e.g. Neon) terminate TLS at a proxy layer. In those cases pg_stat_ssl.ssl may report false because the backend sees plaintext from the proxy — not the client.

Lifecycle Hooks

Quebec provides several lifecycle hooks that you can use to execute code at different stages of the application lifecycle:

• @qc.on_start: Called when Quebec starts
• @qc.on_stop: Called when Quebec stops
• @qc.on_worker_start: Called when a worker starts
• @qc.on_worker_stop: Called when a worker stops
• @qc.on_shutdown: Called during graceful shutdown

These hooks are useful for:

• Initializing resources
• Cleaning up resources
• Logging application state
• Monitoring worker lifecycle
• Graceful shutdown handling