Dead-letter queue Python API¶

The DeadLetterService handles jobs that exhausted every retry. It is reached through your Soniq instance as soniq.dead_letter - the handle is built lazily on first access and cached on the instance, so it always talks to that instance's database.

For the table schema, the CLI, and a debugging walkthrough see Dead-letter queue. This page is the Python API.

from soniq import Soniq
from soniq.features.dead_letter import DeadLetterFilter

soniq = Soniq(database_url="postgresql://localhost/myapp")

jobs = await soniq.dead_letter.list_dead_letter_jobs(
    DeadLetterFilter(job_names=["billing.charge"], limit=50)
)
for job in jobs:
    print(job.id, job.last_error)

DeadLetterFilter¶

Query filter for the list/bulk methods. A @dataclass - construct it with keyword arguments (or no arguments and assign attributes afterward). All fields are optional.

Field	Type	Default	Description
`job_names`	`list[str] \\| None`	`None`	Match any of these task names.
`queues`	`list[str] \\| None`	`None`	Match any of these queues.
`reasons`	`list[str] \\| None`	`None`	Match any of these dead-letter reasons (see `DeadLetterReason`).
`date_from`	`datetime \\| None`	`None`	Only jobs moved to the DLQ at or after this time.
`date_to`	`datetime \\| None`	`None`	Only jobs moved to the DLQ at or before this time.
`tags`	`dict[str, str] \\| None`	`None`	Match rows whose `tags` JSON contains every given key/value.
`has_been_resurrected`	`bool \\| None`	`None`	`True` = replayed at least once, `False` = never replayed, `None` = either.
`limit`	`int`	`1000`	Maximum rows to return.
`offset`	`int`	`0`	Rows to skip (pagination).

f = DeadLetterFilter(
    job_names=["myapp.tasks.sync_inventory"],
    reasons=["max_retries_exceeded"],
    limit=100,
)

DeadLetterJob¶

The record returned by list_dead_letter_jobs() and get_dead_letter_job(). A frozen-shaped @dataclass.

Attribute	Type	Description
`id`	`str`	UUID of the dead-letter row. Pass this to `replay()` / `delete_dead_letter_job()`.
`job_name`	`str`	Task name of the failed job.
`args`	`dict`	The original job arguments.
`queue`	`str`	Queue the job ran in.
`priority`	`int`	Original priority.
`max_attempts`	`int`	Attempt cap the job had.
`attempts`	`int`	How many times it actually ran.
`last_error`	`str`	Exception message from the final attempt. For a job that ran out of retries, Soniq prefixes it with `Max retries exceeded:`.
`dead_letter_reason`	`str`	Why it landed here (`max_retries_exceeded`, `permanent_failure`, `job_not_found`, `invalid_arguments`, `timeout`, `resource_exhausted`, `manual_move`).
`original_created_at`	`datetime`	When the job was first enqueued.
`moved_to_dead_letter_at`	`datetime`	When it entered the DLQ.
`resurrection_count`	`int`	Times it has been replayed (starts at 0).
`last_resurrection_at`	`datetime \\| None`	When it was last replayed.
`tags`	`dict[str, str] \\| None`	Free-form tags added via `add_tags_to_job()`.

Methods¶

All methods are async.

list_dead_letter_jobs¶

async def list_dead_letter_jobs(
    filter_criteria: DeadLetterFilter | None = None,
) -> list[DeadLetterJob]

Return dead-letter rows, newest first. With no filter, returns up to the DeadLetterFilter default of 1000 rows.

get_dead_letter_job¶

async def get_dead_letter_job(dead_letter_id: str) -> DeadLetterJob | None

Fetch a single row by id. None if it does not exist. An id that is not a valid UUID raises ValueError.

replay¶

async def replay(
    dead_letter_id: str,
    reset_attempts: bool = True,
    new_max_attempts: int | None = None,
    new_priority: int | None = None,
    new_queue: str | None = None,
) -> str | None

Mint a fresh soniq_jobs row from the dead-letter row and, in the same transaction, increment resurrection_count on the DLQ row. The DLQ row is kept as the audit trail, so the same row can be replayed more than once. Returns the new job's UUID, or None if the id is unknown or the target job is not registered on this Soniq instance - the registry lookup is how replay knows the new row's attempt cap, so the job's module must be imported first (this is why soniq dead-letter replay needs --jobs-modules). Replay needs the PostgreSQL backend. An invalid UUID raises ValueError.

Argument	Default	Effect
`reset_attempts`	`True`	Start the new job at `attempts=0`. `False` carries the old count over.
`new_max_attempts`	`None`	Override the attempt cap for the replayed job. `0` counts as unset.
`new_priority`	`None`	Override the priority.
`new_queue`	`None`	Route the replayed job to a different queue.

new_id = await soniq.dead_letter.replay(
    job.id, new_max_attempts=10, new_queue="retry-slow"
)

bulk_replay¶

async def bulk_replay(
    filter_criteria: DeadLetterFilter,
    reset_attempts: bool = True,
    new_max_attempts: int | None = None,
) -> list[str]

Replay every row matching the filter. Rows whose job is unregistered are skipped (logged, not raised). A replay that fails with an exception is also logged and skipped. The filter limit (default 1000) caps the number of rows. Returns the list of new job UUIDs.

delete_dead_letter_job¶

async def delete_dead_letter_job(dead_letter_id: str) -> bool

Permanently remove one dead-letter row. True if a row was deleted. An invalid UUID raises ValueError.

bulk_delete¶

async def bulk_delete(filter_criteria: DeadLetterFilter) -> int

Permanently remove every row matching the filter. Returns the count deleted. An empty filter deletes the whole table, so pass a real filter.

cleanup_old_dead_letter_jobs¶

async def cleanup_old_dead_letter_jobs(days: int = 30, dry_run: bool = False) -> int

Delete rows older than days (by moved_to_dead_letter_at). Returns the count deleted. With dry_run=True, it deletes nothing and returns the count that it would delete. Good for a @soniq.periodic housekeeping job.

get_dead_letter_stats¶

async def get_dead_letter_stats(hours: int | None = None) -> DeadLetterStats

Aggregate counts over the whole table, or the last hours if given. by_date always covers the last 7 days, whatever hours is. DeadLetterStats has: total_count, by_job_name, by_queue, by_reason, by_date (all dict[str, int]), oldest_job_age_hours (float), and resurrection_success_rate (float, a percentage from 0 to 100: the share of rows replayed at least once). Point your DLQ alerting at this.

add_tags_to_job¶

async def add_tags_to_job(dead_letter_id: str, tags: dict[str, str]) -> bool

Merge tags into the row's existing tags (JSON). True if the row was updated. Filter on them later with DeadLetterFilter(tags={...}).

export_dead_letter_jobs¶

async def export_dead_letter_jobs(
    filter_criteria: DeadLetterFilter | None = None,
    format: str = "json",
    output: str | None = None,
) -> str

Write matching rows to output ("json" or "csv") and return the path. Any other format raises ValueError. The filter limit caps the rows. Soniq creates the file with mode 0600 and refuses to write through a symlink. Without output, it writes a timestamped file in the current directory. An empty CSV export still gets its header row.