PgQueuer is a Python library that turns PostgreSQL into a job queue using LISTEN/NOTIFY for instant notifications and FOR UPDATE SKIP LOCKED for worker coordination. Supports async (asyncpg, psycopg) and sync (psycopg) drivers, with an in-memory adapter for testing. Python >=3.11, async-first, MIT-licensed.
pgqueuer/ Core library (hexagonal architecture)
domain/ Pure domain: models, types, errors, settings
ports/ Protocol interfaces (repository, driver, tracing)
core/ Business logic: QueueManager, SchedulerManager, executors
adapters/ Infrastructure: DB drivers, persistence, tracing, CLI, in-memory
*.py (top-level) Backward-compatibility shims re-exporting from canonical locations
test/ All tests (unit, integration, windows)
conftest.py Shared fixtures (testcontainers-based Postgres, per-test DB)
helpers.py Test utilities (mocked_job, wait_until_empty_queue)
integration/ FastAPI/Flask integration tests
examples/ Consumer, producer, scheduler, framework integration examples
docs/ MkDocs documentation source
tools/ Benchmarking and monitoring scripts
All commands use uv as the package manager. Install deps first: uv sync --frozen
Dev-only dependencies go in [dependency-groups] in pyproject.toml (see
Dependency groups); runtime
integrations go in [project.optional-dependencies].
# Run ALL checks (recommended before any PR)
uv sync --frozen && uv run ruff check . && uv run ruff format . --check && uv run lint-imports && uv run mypy . && uv run pytest
# Individual checks
uv run ruff check . # Lint (ruff)
uv run ruff format . --check # Format check
uv run lint-imports # Hexagonal architecture boundary validation
uv run mypy . # Type checking (strict, targets Python 3.11)
# Run all tests (parallel by default: 2 xdist workers, one Postgres container each)
uv run pytest
# Run tests serially (disable pytest-xdist)
uv run pytest -n 0
# Run a single test file
uv run pytest test/test_queries.py
# Run a single test function
uv run pytest test/test_queries.py::test_queries_put
# Run a specific parametrized variant
uv run pytest "test/test_queries.py::test_queries_put[1]"
# Run with coverage (as CI does)
uv run pytest --cov=pgqueuer --cov-report=xml --cov-report=term-missingTests use testcontainers to auto-start a Postgres container. No manual setup needed if Docker is available. Alternatively, set EXTERNAL_POSTGRES_DSN or individual PG env vars:
PGUSER=pgquser PGDATABASE=pgqdb PGPASSWORD=pgqpw PGHOST=localhost PGPORT=5432To use docker-compose instead: docker compose up db populate (and docker compose down after).
pgq install # Create tables, triggers, functions
pgq sql install # Emit SQL to stdout without connecting (pipe to psql, etc.)
pgq uninstall # Remove all PgQueuer objects
pgq upgrade --plan # Print the delta this database needs; apply nothing
pgq upgrade # Converge the schema onto the declaration
pgq verify --expect present # Check schema exists (exit 1 on mismatch)The schema is declared once, in pgqueuer/domain/schema/ — the only place to
edit when the schema changes. Install renders from it, upgrade diffs the
catalog against it, and there is no migration list to append to (ADR-0016).
Definitions use PostgreSQL's own spelling (format_type, pg_get_expr,
pg_get_indexdef, pg_proc.prosrc) so comparison is string equality. Get one
wrong and test_schema_inspect.py fails with the spelling to paste in. That
test and test_schema_convergence.py (upgrades a fixture of every old release)
keep the model honest; both must pass on PG 14–18.
The planner never emits DROP TABLE or DROP COLUMN. A column the schema no
longer declares goes in retired(): NOT NULL relaxed, drop left to the
operator as a note.
uv run pytest -vv --log-cli-level=INFO # Verbose with live log output
uv run pytest -m "not integration" # Skip integration tests (no DB needed)PgQueuer follows hexagonal (ports & adapters) architecture enforced by import-linter rules in pyproject.toml.
domain/— Pure models, types, settings, errors. No imports from other layers.ports/— Protocol definitions (Driver,RepositoryPort,SchemaManagementPort, etc.). No imports from adapters or core.core/— Business logic:qm.py—QueueManager: dequeue loop, dispatch, concurrency control, health checkssm.py—SchedulerManager: cron-based recurring tasksapplications.py—PgQueuer: top-level orchestrator combining QM + SMexecutors.py—AbstractEntrypointExecutor/AbstractScheduleExecutorand default implementationsbuffers.py— Batched async buffers for heartbeats, job status logslisteners.py— PG NOTIFY event routingtm.py—TaskManagerfor background task lifecycleheartbeat.py,helpers.py,logconfig.py
adapters/— Concrete implementations:drivers/—asyncpg.py(AsyncpgDriver, AsyncpgPoolDriver),psycopg.py(PsycopgDriver, SyncPsycopgDriver)persistence/—queries.py(SQL queries),qb.py(query builder + DBSettings),query_helpers.py,schema_ddl.py(render the model as DDL),schema_inspect.py(read the installed schema),schema_plan.py(diff the two)inmemory/— In-memory driver and queries for testing without Postgrestracing/— Logfire, Sentry, OpenTelemetry integrationscli/— CLI commands (run, install, dashboard, etc.)
- Domain (
pgqueuer/domain/) must NOT import from adapters or core - Ports (
pgqueuer/ports/) must NOT import from adapters or core - Core (
pgqueuer/core/) must NOT import from adapters (several temporary exceptions listed inpyproject.toml)
Port protocols: QueueRepositoryPort, ScheduleRepositoryPort, NotificationPort, SchemaManagementPort. The Queries class satisfies all four via structural subtyping. Core code should depend on protocols, not Queries directly.
Files like pgqueuer/qm.py, pgqueuer/queries.py, pgqueuer/models.py, etc. at the package root are re-export shims that import from the layered modules. The public API (pgqueuer/__init__.py) exposes PgQueuer, QueueManager, SchedulerManager, Queries, Job, driver classes, etc.
- Dataclasses for internal state (
QueueManager,Queries, executors) - Pydantic BaseModel for data transfer objects (
Job,Event,Schedule,Log) - Decorator registration:
@pgq.entrypoint("name"),@pgq.schedule("name", "*/5 * * * *") - Factory classmethods:
from_asyncpg_connection(),from_psycopg_connection(),in_memory() - Async-only: all entrypoints must be async (
async def); core logic is fully async - Buffer pattern:
JobStatusLogBuffer,HeartbeatBufferfor batched async I/O
When deprecating dataclass fields, use a module-level _SENTINEL = object() default with a warnings.warn(..., DeprecationWarning, stacklevel=2) check in __post_init__.
QueueManager and SchedulerManager accept an optional queries argument (default=None). When omitted, they auto-create Queries(self.connection). Prefer injecting a mock for tests.
- Formatter/linter: ruff (line-length=100)
- Lint rules: C, E, F, I, PIE, Q, RET, RSE, SIM, W, C90 (max-complexity=15)
- isort: combined-as-imports, furthest-to-closest relative imports
- Mypy
strict = truepluswarn_unreachableand the error codestruthy-bool,truthy-iterable,redundant-expr,unused-awaitable,possibly-undefined,ignore-without-code; Pydantic plugin withinit_typed; targets Python 3.11. Insidepgqueuer/,tools/andexamples/the overrides adddisallow_any_explicitanddisallow_any_unimported. Run it asuv run mypy . --no-incremental; the incremental cache has reported green on stale signatures.
from __future__ import annotations(required in every file)- Standard library
- Third-party packages
- Internal imports using absolute paths (
from pgqueuer.domain.models import Job)
No local imports. All imports must be at module top level. The only exception is if TYPE_CHECKING: blocks for avoiding circular imports at runtime.
- Always annotate all function/method signatures (
strict = trueimpliesdisallow_untyped_defs) - Use native Python types:
list[int],dict[str, object],int | None(notOptional) - Use
NewTypefor domain primitives:JobId = NewType("JobId", int) - Use
Literalfor string unions:Literal["queued", "picked", "successful"] - Exhaustive dispatch. When branching on a
LiteralorEnumvalue, handle every member in an explicitif/elifchain and terminate withelse: assert_never(value)(typing.assert_never) so mypy proves totality. No bare fallthroughs, no catch-allelsedoing real work — adding a new member must produce a type error at every dispatch site. - Use
Protocolfor structural subtyping (port interfaces) - Use
TypeAliasfor complex callable types Anyis banned inpgqueuer/,tools/andexamples/(disallow_any_explicit). Useobjectand narrow withisinstance. Driver*argsareobject; rows come back asdict[str, object]and scalars are read throughquery_helpers.cell(row, key, kind), multi-column rows through a pydantic model. User-owned bags (Context.resources) areMutableMapping[str, object].Callable[..., X]counts asAny; declare aProtocolwith__call__(self, *args: object)instead. Untyped third-party values are assigned to anobject-annotated name before anisinstancecheck. Test code undertest/may useAny.- Import domain types from
pgqueuer.domain.types, never viamodels.JobId;no_implicit_reexportis on and only the top-level shims re-export. - Generic constructors over type-annotated assignments for typed stdlib objects:
fut = asyncio.Future[MyType]()notfut: asyncio.Future[MyType] = asyncio.Future(). - No bare tuples as records. A tuple whose elements mean different things needs a
NamedTupleat minimum, a frozen dataclass when it grows behaviour.tuple[X, ...]as a homogeneous collection is fine;tuple[str, str]standing in for a record is not — the caller unpacks by position, a swap type-checks, and the meaning lives only in the variable name at the call site.
# Bad — what is [0], and what happens when a third element appears?
def column_kind_and_default(row: Row) -> tuple[ColumnKind, str | None]: ...
retired_columns: tuple[tuple[str, str], ...]
# Good
class ColumnRef(NamedTuple):
table: TableName
column: ColumnName
retired_columns: tuple[ColumnRef, ...]- No
# type: ignorein production code.type: ignorecomments are forbidden inpgqueuer/. Fix the underlying type issue instead (usedataclasses.KW_ONLY, protocols, generics, overloads, etc.).# type: ignoreis acceptable in test code only.
getattr, setattr, hasattr, and delattr are banned in pgqueuer/. They defeat the type checker: mypy cannot verify a name passed as a string, so every use is an untyped hole in an otherwise strict codebase, and a typo becomes a runtime None instead of an error.
The common temptation is probing an object for an attribute to decide what it is. Declare a runtime_checkable Protocol in pgqueuer/ports/ and use isinstance — the shape gets a name, a docstring, and a type mypy narrows on:
# Bad — the type checker learns nothing, and "sqlstate" is unverifiable
code = getattr(exc, "sqlstate", None)
# Good — the shape is declared once and checked structurally
@runtime_checkable
class SqlStateError(Protocol):
"""Driver exception carrying a PostgreSQL SQLSTATE code."""
@property
def sqlstate(self) -> str | None: ...
if isinstance(exc, SqlStateError) and isinstance(exc.sqlstate, str):
return exc.sqlstateThree grandfathered call sites remain: domain/settings.py iterating declared dataclass fields, core/executors.py probing __call__, and adapters/cli/factories.py resolving a module:attr path. Add no more; replace them when the code is next touched.
operator.attrgetter, operator.itemgetter and operator.methodcaller are banned outright. They are getattr, a subscript and a method call wearing a functional hat: the name stays a string mypy cannot check, so attrgetter("naem") type-checks and blows up at runtime. Being stdlib does not redeem them. A lambda says the same thing and is checked:
# Bad — "name" is unverifiable, and the sort key's type is unknown
tuple(sorted(tables, key=operator.attrgetter("name")))
# Good — mypy resolves the attribute and its type
tuple(sorted(tables, key=lambda table: table.name))Test code may use reflection where the test is about reflection. monkeypatch.setattr is pytest's API, not the builtin, and is unaffected.
| Element | Convention | Example |
|---|---|---|
| Classes | CamelCase | QueueManager, DuplicateJobError |
| Functions/methods | snake_case | register_executor, fetch_jobs |
| Constants/aliases | UPPER_SNAKE_CASE | JOB_STATUS, EVENT_TYPES |
| Test functions | test_ prefix |
test_queries_put |
| Fixtures | snake_case | apgdriver, dsn |
No leading-underscore prefixes. Python has no real public/private distinction. Use plain snake_case for all methods and attributes — pick a descriptive name instead of hiding it behind a prefix.
Multi-line string content uses triple quotes. Never build it from implicitly concatenated adjacent literals or +-joined pieces with embedded \n escapes.
# Bad
comment=(
"Entrypoints with free capacity; remaining is NULL for\n"
"unlimited entrypoints."
)
# Good
comment="""
Entrypoints with free capacity; remaining is NULL for
unlimited entrypoints.
"""When surrounding indentation would leak into the string, normalize with textwrap.dedent (or accept dedent at the consuming end, as SqlComposer.cte does for its comment parameter). Joining dynamic parts (", ".join(items)) is fine — the rule is about literals.
No single-letter table or CTE aliases (q, a, p, pk). The SQL here is dense enough already — use the full table/CTE name, or a descriptive alias (candidate, stale, job) when the name is parameterized or shadowed.
__all__ belongs only in re-export modules: package __init__.py files and the top-level backward-compatibility shims. Internal modules do not declare it — everything is importable, per the no-underscore rule above.
No globals. Ever. Globals are banned outright — mutable and immutable data structures alike. No module-level registries, lookup tables, caches, collections, config objects, or singletons that hold program data. Runtime state and data live on an instance and are passed explicitly; if something must be shared, thread it through a constructor or method argument, never a module attribute.
A module-level list/dict/set/tuple/dataclass instance used as a registry, table, or accumulator is a global — forbidden. Put it on a class (class attribute or instance) and pass it explicitly.
The only module-level names permitted are those the type system or a framework structurally requires, none of which hold program data:
- Type-level declarations:
NewType,TypeAlias,Literalaliases,Protocol/TypeVardefinitions,__all__. - Scalar literal constants (
UPPER_SNAKE_CASEint/str/bytes/bool) and the deprecation_SENTINEL = object(). - Framework-mandated singletons that hold no mutable state (
app = typer.Typer(...)).
The grandfathered tracing.TRACER is the sole legacy exception; add no more, and migrate it off module level when the code is next touched.
- Custom exceptions inherit from
PgqException(inpgqueuer/domain/errors.py) - Hierarchy:
PgqException > RetryException > RetryRequested; alsoDuplicateJobError,FailingListenerError,SchemaDriftError - Use
logconfig.logger.exception(...)for logging errors in job dispatch - Use
contextlib.suppress(...)for non-critical errors - Use
pytest.raisesin tests for expected exceptions
Follow PEP 257 for format. Keep them tight.
- One-liner: triple-quoted on the same line, period at the end.
"""Return the picked job.""" - Multi-line: summary line, blank line, body. Closing
"""on its own line. - Don't restate the signature. Type annotations already say what the args and return type are. Only mention a parameter when something non-obvious matters (units, constraints, ownership, side effects).
- Skip docstrings on trivial or private helpers. A descriptive name beats a docstring that repeats it.
- Test docstrings: one line describing the behavior under test. Skip when the test name is self-evident.
- Module docstrings: optional. When present, one sentence. No "This module defines..." preambles.
- External-schema docstrings exempt. Where a docstring is the canonical, externally-published description of a function -- specifically
@mcp.tool()and@mcp.prompt()handlers inpgqueuer/adapters/mcp/, whose docstrings are surfaced verbatim to MCP clients as the tool's schema -- the verbosity rules above do not apply. Document return columns, parameter semantics, and usage patterns at whatever length the consumer needs. - Usage examples recommended on public APIs. Public classes and functions exported from
pgqueuer/__init__.py(or otherwise documented for end users) should include a shortUsage example::block in the docstring -- one realistic call, placed after the summary/body. Skip the example when the API is trivial and the call site is self-evident from the signature (e.g. simple getters, one-arg helpers, obvious one-liners). Judgement call: if a new user would not need to look at the example, don't write one.
Default to no comments. Code with descriptive names is more durable than comments that rot.
Only add a comment when the why is non-obvious:
- A hidden invariant or constraint that the code relies on
- A workaround for a specific bug, library quirk, or platform behavior
- A subtle ordering or concurrency requirement
- Something that would surprise a reader who understood the code
Do not write:
- Comments that restate what the code does (
# increment counter) - Caller references (
# used by X,# called from Y) — they go stale andgit grepis faster - Task or issue refs in inline comments — link them in the PR description instead. Exception: regression tests may reference the issue ID in their docstring or
xfailreason, because the issue is the test's reason to exist. - Banner section headers (
# ----- Layer 2 -----) — let function names and module structure do the work - TODOs without an owner and a tracking link
- Step-by-step narration of a block — a comment per step retells the code in prose. Annotate only the step that would surprise a reader.
When a comment explains a non-obvious workaround, keep it short and put it adjacent to the line it explains. Prefer one line.
Two lines is the cap. Not a target to fill — a ceiling. A comment that wants a third line is a comment that has stopped explaining a surprise and started teaching a subject; the subject belongs in an ADR (docs/adr/) or the PR description, and the comment cites it in one line. This applies hardest above a SQL literal or a long function, where an essay is easy to write and rarely read.
If the comment prose in a function is a noticeable fraction of its code, that is a defect regardless of whether any single comment obeys the cap. Cut until each surviving line earns its place.
- Follow existing patterns. Before writing new code, read surrounding modules to match conventions (naming, structure, error handling, dataclass usage, etc.). Do not invent new patterns when an established one exists.
- Readability and correctness above speed. Follow the Zen of Python -- explicit is better than implicit, simple is better than complex, readability counts. Never sacrifice clarity or correctness for performance unless there is a measured, proven need.
- Every change must be proven correct by a test. Never accept a code change without an accompanying test. Tests must be narrow and precise -- test exactly the behavior being changed, not broad integration sweeps.
- Every user-facing feature must be documented. New public APIs, exceptions, executor classes, CLI commands, and behavior changes require corresponding documentation in
docs/. Update existing guides (e.g.,reliability.md,custom-executors.md,core-concepts.md) and create new guide pages when the feature warrants its own topic. Keep themkdocs.ymlnav in sync.
- pytest config:
asyncio_mode = "auto", 20s timeout per test,--durations=15,-n 2(pytest-xdist) - Tests run in parallel via pytest-xdist: session-scoped fixtures are per-worker, so each worker starts its own Postgres container and template database. Pass
-n 0to run serially, or-n <N>to change worker count. - Keep tests xdist-safe: no fixed ports, no shared files, no cross-test state. Database isolation comes from the per-test
CREATE DATABASE ... TEMPLATEfixture. - Declare async tests as
async def test_...-- no@pytest.mark.asynciodecorator needed - Annotate all test function parameters:
async def test_foo(apgdriver: db.Driver, N: int) -> None: - Use
@pytest.mark.parametrizefor multi-value testing - Fixtures create a fresh database per test via
CREATE DATABASE ... TEMPLATE - Key fixtures in
test/conftest.py:dsn(per-test DB URL),apgdriver(AsyncpgDriver),pgdriver(SyncPsycopgDriver) - In-memory tests (
test_inmemory.py) don't need Postgres
PgQueuer follows strict semantic versioning (SemVer) from v1.0.0 onward:
- Patch (1.0.x): Bug fixes only, no API changes.
- Minor (1.x.0): New features, fully backward-compatible.
- Major (x.0.0): Breaking changes — reserved for when there is no alternative.
Never introduce a breaking change in a patch or minor release. If a change would break any public API (function signatures, class fields, import paths, CLI behavior, database schema), it requires a major version bump.
Every issue or pull request an agent writes must carry the Agentic label ("Issue or PR written mainly by an AI agent. Required on all agent-authored contributions."). See CONTRIBUTING.md.
- Apply the label when creating it, e.g.
gh pr create --label Agenticorgh issue create --label Agentic. - If the label can't be set (no triage permission), put "Agentic" on the first line of the description so a maintainer can add it.
- Fill in the issue or PR template like any other contribution.
This project follows Conventional Commits.
<type>[optional scope]: <description> (#{PR_ID})
- type:
feat,fix,docs,style,refactor,perf,test,build,ci,chore - scope (optional): area of the codebase, e.g.
cli,qm,sm,schema,drivers,inmemory - description: imperative mood, lowercase, no period at the end
!after type/scope: marks a breaking change (e.g.feat!:orfeat(cli)!:)- PR reference: end with the PR number in parentheses, prefixed with
#
feat(qm): add global concurrency limit (#92)
fix: restore partial index usage in dequeue CTE (#607)
feat!: simplify CLI connection config, remove dsn() helper (#605)
docs: add custom executor guide (#88)
refactor(core): extract heartbeat into standalone module (#95)
test: add parametrized dequeue batch tests (#100)
chore: bump ruff to 0.8 (#101)
- Body: present tense, wrap at 72 characters. Explain why, not just what.
- Footer: use
BREAKING CHANGE: <description>for breaking changes (in addition to or instead of!in the title).
git log --onelinegives a clean, typed, scannable history.git blamegives immediate context.- The
#{PR_ID}suffix implicitly links to the PR in GitHub. - Types enable automated changelogs and semantic versioning.
Python 3.11--3.15 x Postgres 14--18 on Ubuntu. Windows tests run only test/windows/test_shutdown.py.