pgqueuer

# AGENTS.md -- Guidance for AI Code-Generation Agents

## Project Overview

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.10, async-first, MIT-licensed.

## Project Structure

```
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
```

## Build, Lint, and Test Commands

All commands use `uv` as the package manager. Install deps first: `uv sync --all-extras --frozen`

```bash
# Run ALL checks (recommended before any PR)
uv sync --all-extras --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.10)

# 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-missing
```

### Database for Tests

Tests 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:

```bash
PGUSER=pgquser PGDATABASE=pgqdb PGPASSWORD=pgqpw PGHOST=localhost PGPORT=5432
```

To use docker-compose instead: `docker compose up db populate` (and `docker compose down` after).

### Schema Management CLI

```bash
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.

### Additional Test Flags

```bash
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)
```

## Architecture

PgQueuer follows **hexagonal (ports & adapters) architecture** enforced by `import-linter` rules in `pyproject.toml`.

### Layer Structure inside `pgqueuer/`

- **`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 checks
  - `sm.py` — `SchedulerManager`: cron-based recurring tasks
  - `applications.py` — `PgQueuer`: top-level orchestrator combining QM + SM
  - `executors.py` — `AbstractEntrypointExecutor` / `AbstractScheduleExecutor` and default implementations
  - `buffers.py` — Batched async buffers for heartbeats, job status logs
  - `listeners.py` — PG NOTIFY event routing
  - `tm.py` — `TaskManager` for background task lifecycle
  - `heartbeat.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 Postgres
  - `tracing/` — Logfire, Sentry, OpenTelemetry integrations
  - `cli/` — CLI commands (run, install, dashboard, etc.)

### Import Rules (enforced by lint-imports)

1. **Domain** (`pgqueuer/domain/`) must NOT import from adapters or core
2. **Ports** (`pgqueuer/ports/`) must NOT import from adapters or core
3. **Core** (`pgqueuer/core/`) must NOT import from adapters (several temporary exceptions listed in `pyproject.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.

### Top-level Shim Modules

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.

### Key Patterns

- **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`, `HeartbeatBuffer` for batched async I/O

### Deprecation Pattern

When deprecating dataclass fields, use a module-level `_SENTINEL = object()` default with a `warnings.warn(..., DeprecationWarning, stacklevel=2)` check in `__post_init__`.

### Dependency Injection

`QueueManager` and `SchedulerManager` accept an optional `queries` argument (default=None). When omitted, they auto-create `Queries(self.connection)`. Prefer injecting a mock for tests.

## Code Style

### Formatting and Linting

- **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 = true` plus `warn_unreachable` and the error codes `truthy-bool`, `truthy-iterable`, `redundant-expr`, `unused-awaitable`, `possibly-undefined`, `ignore-without-code`; Pydantic plugin with `init_typed`; targets Python 3.10. Inside `pgqueuer/`, `tools/` and `examples/` the overrides add `disallow_any_explicit` and `disallow_any_unimported`. Run it as `uv run mypy . --no-incremental`; the incremental cache has reported green on stale signatures.

### Import Rules

1. `from __future__ import annotations` (required in every file)
2. Standard library
3. Third-party packages
4. 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.

### Type Annotations

- **Always annotate** all function/method signatures (`strict = true` implies `disallow_untyped_defs`)
- Use **native Python types**: `list[int]`, `dict[str, object]`, `int | None` (not `Optional`)
- Use `NewType` for domain primitives: `JobId = NewType("JobId", int)`
- Use `Literal` for string unions: `Literal["queued", "picked", "successful"]`
- **Exhaustive dispatch.** When branching on a `Literal` or `Enum` value, handle every member in an explicit `if`/`elif` chain and terminate with `else: assert_never(value)` (`typing_extensions.assert_never`) so mypy proves totality. No bare fallthroughs, no catch-all `else` doing real work — adding a new member must produce a type error at every dispatch site.
- Use `Protocol` for structural subtyping (port interfaces)
- Use `TypeAlias` for complex callable types
- **`Any` is banned in `pgqueuer/`, `tools/` and `examples/`** (`disallow_any_explicit`). Use `object` and narrow with `isinstance`. Driver `*args` are `object`; rows come back as `dict[str, object]` and scalars are read through `query_helpers.cell(row, key, kind)`, multi-column rows through a pydantic model. User-owned bags (`Context.resources`) are `MutableMapping[str, object]`. `Callable[..., X]` counts as `Any`; declare a `Protocol` with `__call__(self, *args: object)` instead. Untyped third-party values are assigned to an `object`-annotated name before an `isinstance` check. Test code under `test/` may use `Any`.
- **Import domain types from `pgqueuer.domain.types`**, never via `models.JobId`; `no_implicit_reexport` is on and only the top-level shims re-export.
- **Generic constructors over type-annotated assignments** for typed stdlib objects: `fut = asyncio.Future[MyType]()` not `fut: asyncio.Future[MyType] = asyncio.Future()`.
- **No bare tuples as records.** A tuple whose elements mean different things needs a `NamedTuple` at 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.

```python
# 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: ignore` in production code.** `type: ignore` comments are forbidden in `pgqueuer/`. Fix the underlying type issue instead (use `dataclasses.KW_ONLY`, protocols, generics, overloads, etc.). `# type: ignore` is acceptable in test code only.

### Dynamic Attribute Access

**`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:

```python
# 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.sqlstate
```

Three 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:

```python
# 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.

### Naming Conventions

| 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.

### String Literals

**Multi-line string content uses triple quotes.** Never build it from implicitly concatenated adjacent literals or `+`-joined pieces with embedded `\n` escapes.

```python
# 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.

### SQL Style

**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__`

`__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.

### Module-Level State

**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`, `Literal` aliases, `Protocol`/`TypeVar` definitions, `__all__`.
- Scalar literal constants (`UPPER_SNAKE_CASE` `int`/`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.

### Error Handling

- Custom exceptions inherit from `PgqException` (in `pgqueuer/domain/errors.py`)
- Hierarchy: `PgqException > RetryException > RetryRequested`; also `DuplicateJobError`, `FailingListenerError`, `SchemaDriftError`
- Use `logconfig.logger.exception(...)` for logging errors in job dispatch
- Use `contextlib.suppress(...)` for non-critical errors
- Use `pytest.raises` in tests for expected exceptions

### Docstrings

Follow [PEP 257](https://peps.python.org/pep-0257/) 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 in `pgqueuer/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 short `Usage 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.

### Comments

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 and `git grep` is 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 `xfail` reason, 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.

### Guiding Principles

- **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 the `mkdocs.yml` nav in sync.

## Testing Conventions

- **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 0` to 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 ... TEMPLATE` fixture.
- Declare async tests as `async def test_...` -- no `@pytest.mark.asyncio` decorator needed
- Annotate all test function parameters: `async def test_foo(apgdriver: db.Driver, N: int) -> None:`
- Use `@pytest.mark.parametrize` for 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

## Versioning

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.

## Issues and Pull Requests

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](CONTRIBUTING.md#agentic-contributions).

- Apply the label when creating it, e.g. `gh pr create --label Agentic` or `gh 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.

## Commit Conventions

This project follows [Conventional Commits](https://www.conventionalcommits.org/en/v1.0.0/).

### Title Format

```
<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!:` or `feat(cli)!:`)
- **PR reference**: end with the PR number in parentheses, prefixed with `#`

### Examples

```
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 and Footer

- 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).

### Rationale

- `git log --oneline` gives a clean, typed, scannable history.
- `git blame` gives immediate context.
- The `#{PR_ID}` suffix implicitly links to the PR in GitHub.
- Types enable automated changelogs and semantic versioning.

## CI Matrix

Python 3.10--3.14 x Postgres 14--18 on Ubuntu. Windows tests run only `test/windows/test_shutdown.py`.

More agent context in janbjorge/pgqueuer

2 other files this repository gives its agents.