audit_trail.maintenance

Partition management, retention and health checks.

ensure_partitions creates the partitions writes will need: every severity partition of the activity table, and the monthly partitions of the transaction table and of each severity partition, from the current UTC month months_ahead months forward. There is no DEFAULT partition, so a row without a partition fails with SQLSTATE 23514; run it on a schedule (cron, worker, application start) with enough months ahead to cover a missed run.

Existing partitions are found by their bounds in the catalog, not by name: a partition created by hand or by another tool under a different name, with the same (or wider) bounds, counts as present. Names and bounds are described in audit_trail.migrations.

Locking: CREATE TABLE ... PARTITION OF takes an ACCESS EXCLUSIVE lock on the parent, so it waits for every open transaction that has touched the parent, including business transactions with uncommitted audit rows. Under PostgreSQL's documented lock queueing, later lock requests that conflict with the waiting one, such as new audit inserts, queue behind it. lock_timeout bounds that wait: past it the whole call is rolled back and PartitionLockTimeoutError is raised; retrying later is safe.

drop_expired removes monthly partitions older than a per-severity retention, with ALTER TABLE ... DETACH PARTITION ... CONCURRENTLY and then DROP TABLE. health reports how many months ahead are covered, detaches left pending and leftover tables that look like detached partitions.

Locks taken by drop_expired, as observed in pg_locks on PostgreSQL 14 and 18:

• DETACH ... CONCURRENTLY first takes SHARE UPDATE EXCLUSIVE on the parent, which does not conflict with inserts. It then waits for every transaction that has used the parent, holding no lock on any table while it waits, so new audit inserts go ahead. Last, it takes SHARE UPDATE EXCLUSIVE on the parent and ACCESS EXCLUSIVE on the partition, which waits for readers of that partition.
• DETACH ... FINALIZE takes the same locks as that last step, and waits for the same transactions.
• DROP TABLE of the detached table takes ACCESS EXCLUSIVE on that table only.

lock_timeout bounds each of these waits, including the wait for other transactions. An application transaction left open on an audit table for longer than lock_timeout therefore makes drop_expired fail with PartitionLockTimeoutError; this is expected. If that happens after a detach has started, the partition stays "pending detach" and the next call finalizes and drops it.

LOCK_NOT_AVAILABLE module-attribute

LOCK_NOT_AVAILABLE = '55P03'

SQLSTATE raised when lock_timeout expires.

PartitionError

Bases: Exception

Partition maintenance could not be done.

PartitionLockTimeoutError

Bases: PartitionError

A lock needed for partition maintenance was not granted within lock_timeout.

lock_timeout applies to each lock wait, not to the whole call, so ensure_partitions blocked behind drop_expired and then behind an open transaction can take about twice lock_timeout before this is raised. From ensure_partitions, nothing was created; months_ahead leaves room for a later retry. From drop_expired, the partitions dropped before the timeout stay dropped (each one is logged), and a detach left pending is finalized by the next call.

PartitionHealth dataclass

PartitionHealth(
    table: str | None,
    covers_now: bool,
    months_ahead: int | None,
    below: bool,
)

How far ahead one partitioned parent is covered by monthly partitions.

Attributes:

Name	Type	Description
table	str | None	Schema-qualified name of the parent (the transaction table or a severity partition), or None if the severity has no partition.
covers_now	bool	Whether the current UTC month has a partition.
months_ahead	int | None	Whole months after the current one covered without a gap; 0 when the current month is not covered, None when there is no upper bound (a MAXVALUE partition, or a severity partition that is a plain table).
below	bool	Whether the current month is not covered or months_ahead is below the threshold.

HealthReport dataclass

HealthReport(
    min_months_ahead: int,
    transaction: PartitionHealth,
    activity: dict[int, PartitionHealth],
    pending_detach: list[str],
    orphaned: list[str],
)

Result of health.

Attributes:

Name	Type	Description
min_months_ahead	int	The threshold the report was made with.
transaction	PartitionHealth	Coverage of the transaction table.
activity	dict[int, PartitionHealth]	Coverage of each requested severity, by severity value.
pending_detach	list[str]	Schema-qualified names of partitions left "pending detach" by an interrupted drop_expired; its next call finalizes them.
orphaned	list[str]	Schema-qualified names of tables in the audit schemas that are named like monthly partitions but are no partition. A crash between the detach and the drop in drop_expired leaves one behind; nothing drops them automatically. Check the name, then DROP TABLE it by hand.

ok property

ok: bool

Whether nothing is below the threshold, pending or orphaned.

PartitionManager

PartitionManager(
    engine: Engine | AsyncEngine,
    tables: AuditTables,
    severities: Iterable[int],
)

Runs partition maintenance on the library's own connections.

ensure_partitions runs in one transaction; drop_expired switches its connection to AUTOCOMMIT itself.

Parameters:

Name	Type	Description	Default
engine	Engine | AsyncEngine	Sync or async engine with DDL privileges. Must not be configured for AUTOCOMMIT.	required
tables	AuditTables	The audit tables, from audit_trail.tables.build_tables.	required
severities	Iterable[int]	A severity IntEnum class, or any iterable of ints.	required

Source code in audit_trail/maintenance.py

def __init__(
    self,
    engine: Engine | AsyncEngine,
    tables: AuditTables,
    severities: Iterable[int],
) -> None:
    self.engine = engine
    self.tables = tables
    self.severities = severity_values(severities)

ensure_partitions

ensure_partitions(
    months_ahead: int = 3,
    *,
    lock_timeout: str | None = "5s",
) -> list[str]

Create missing partitions in one transaction; see ensure_partitions.

Parameters:

Name	Type	Description	Default
months_ahead	int	Months to create after the current UTC month.	3
lock_timeout	str | None	PostgreSQL lock_timeout for the transaction.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions created.

Raises:

Type	Description
TypeError	If the manager was built with an AsyncEngine.
PartitionLockTimeoutError	If a lock was not granted in time.

Source code in audit_trail/maintenance.py

def ensure_partitions(
    self, months_ahead: int = 3, *, lock_timeout: str | None = "5s"
) -> list[str]:
    """Create missing partitions in one transaction; see ``ensure_partitions``.

    Args:
        months_ahead: Months to create after the current UTC month.
        lock_timeout: PostgreSQL ``lock_timeout`` for the transaction.

    Returns:
        Schema-qualified names of the partitions created.

    Raises:
        TypeError: If the manager was built with an ``AsyncEngine``.
        PartitionLockTimeoutError: If a lock was not granted in time.
    """
    if not isinstance(self.engine, Engine):
        raise TypeError("the engine is async; use aensure_partitions")
    with self.engine.begin() as conn:
        return ensure_partitions(
            conn,
            self.tables,
            self.severities,
            months_ahead=months_ahead,
            lock_timeout=lock_timeout,
        )

aensure_partitions async

aensure_partitions(
    months_ahead: int = 3,
    *,
    lock_timeout: str | None = "5s",
) -> list[str]

Async ensure_partitions, for a manager built with an AsyncEngine.

Parameters:

Name	Type	Description	Default
months_ahead	int	Months to create after the current UTC month.	3
lock_timeout	str | None	PostgreSQL lock_timeout for the transaction.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions created.

Raises:

Type	Description
TypeError	If the manager was built with a sync Engine.
PartitionLockTimeoutError	If a lock was not granted in time.

Source code in audit_trail/maintenance.py

async def aensure_partitions(
    self, months_ahead: int = 3, *, lock_timeout: str | None = "5s"
) -> list[str]:
    """Async ``ensure_partitions``, for a manager built with an ``AsyncEngine``.

    Args:
        months_ahead: Months to create after the current UTC month.
        lock_timeout: PostgreSQL ``lock_timeout`` for the transaction.

    Returns:
        Schema-qualified names of the partitions created.

    Raises:
        TypeError: If the manager was built with a sync ``Engine``.
        PartitionLockTimeoutError: If a lock was not granted in time.
    """
    if isinstance(self.engine, Engine):
        raise TypeError("the engine is sync; use ensure_partitions")
    async with self.engine.begin() as conn:
        return await conn.run_sync(
            lambda sync_conn: ensure_partitions(
                sync_conn,
                self.tables,
                self.severities,
                months_ahead=months_ahead,
                lock_timeout=lock_timeout,
            )
        )

drop_expired

drop_expired(
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]

Drop expired partitions on an AUTOCOMMIT connection; see drop_expired.

Parameters:

Name	Type	Description	Default
retention	Mapping[_Severity, timedelta | None]	Retention by severity value; None keeps forever.	required
transaction_retention	timedelta | None	Retention of the transaction table.	None
lock_timeout	str | None	PostgreSQL lock_timeout for each statement.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions dropped.

Raises:

Type	Description
TypeError	If the manager was built with an AsyncEngine.
PartitionLockTimeoutError	If a lock was not granted in time.

Source code in audit_trail/maintenance.py

def drop_expired(
    self,
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]:
    """Drop expired partitions on an ``AUTOCOMMIT`` connection; see ``drop_expired``.

    Args:
        retention: Retention by severity value; ``None`` keeps forever.
        transaction_retention: Retention of the transaction table.
        lock_timeout: PostgreSQL ``lock_timeout`` for each statement.

    Returns:
        Schema-qualified names of the partitions dropped.

    Raises:
        TypeError: If the manager was built with an ``AsyncEngine``.
        PartitionLockTimeoutError: If a lock was not granted in time.
    """
    if not isinstance(self.engine, Engine):
        raise TypeError("the engine is async; use adrop_expired")
    with self.engine.connect() as conn:
        return drop_expired(
            conn.execution_options(isolation_level="AUTOCOMMIT"),
            self.tables,
            retention,
            transaction_retention=transaction_retention,
            lock_timeout=lock_timeout,
        )

adrop_expired async

adrop_expired(
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]

Async drop_expired, for a manager built with an AsyncEngine.

Parameters:

Name	Type	Description	Default
retention	Mapping[_Severity, timedelta | None]	Retention by severity value; None keeps forever.	required
transaction_retention	timedelta | None	Retention of the transaction table.	None
lock_timeout	str | None	PostgreSQL lock_timeout for each statement.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions dropped.

Raises:

Type	Description
TypeError	If the manager was built with a sync Engine.
PartitionLockTimeoutError	If a lock was not granted in time.

Source code in audit_trail/maintenance.py

async def adrop_expired(
    self,
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]:
    """Async ``drop_expired``, for a manager built with an ``AsyncEngine``.

    Args:
        retention: Retention by severity value; ``None`` keeps forever.
        transaction_retention: Retention of the transaction table.
        lock_timeout: PostgreSQL ``lock_timeout`` for each statement.

    Returns:
        Schema-qualified names of the partitions dropped.

    Raises:
        TypeError: If the manager was built with a sync ``Engine``.
        PartitionLockTimeoutError: If a lock was not granted in time.
    """
    if isinstance(self.engine, Engine):
        raise TypeError("the engine is sync; use drop_expired")
    async with self.engine.connect() as conn:
        auto = await conn.execution_options(isolation_level="AUTOCOMMIT")
        return await auto.run_sync(
            lambda sync_conn: drop_expired(
                sync_conn,
                self.tables,
                retention,
                transaction_retention=transaction_retention,
                lock_timeout=lock_timeout,
            )
        )

health

health(min_months_ahead: int = 2) -> HealthReport

Report partition health; see health.

Parameters:

Name	Type	Description	Default
min_months_ahead	int	Coverage threshold in months after the current one.	2

Returns:

Type	Description
HealthReport	The report.

Raises:

Type	Description
TypeError	If the manager was built with an AsyncEngine.

Source code in audit_trail/maintenance.py

def health(self, min_months_ahead: int = 2) -> HealthReport:
    """Report partition health; see ``health``.

    Args:
        min_months_ahead: Coverage threshold in months after the current one.

    Returns:
        The report.

    Raises:
        TypeError: If the manager was built with an ``AsyncEngine``.
    """
    if not isinstance(self.engine, Engine):
        raise TypeError("the engine is async; use ahealth")
    with self.engine.connect() as conn:
        return health(
            conn, self.tables, self.severities, min_months_ahead=min_months_ahead
        )

ahealth async

ahealth(min_months_ahead: int = 2) -> HealthReport

Async health, for a manager built with an AsyncEngine.

Parameters:

Name	Type	Description	Default
min_months_ahead	int	Coverage threshold in months after the current one.	2

Returns:

Type	Description
HealthReport	The report.

Raises:

Type	Description
TypeError	If the manager was built with a sync Engine.

Source code in audit_trail/maintenance.py

async def ahealth(self, min_months_ahead: int = 2) -> HealthReport:
    """Async ``health``, for a manager built with an ``AsyncEngine``.

    Args:
        min_months_ahead: Coverage threshold in months after the current one.

    Returns:
        The report.

    Raises:
        TypeError: If the manager was built with a sync ``Engine``.
    """
    if isinstance(self.engine, Engine):
        raise TypeError("the engine is sync; use health")
    async with self.engine.connect() as conn:
        return await conn.run_sync(
            lambda sync_conn: health(
                sync_conn,
                self.tables,
                self.severities,
                min_months_ahead=min_months_ahead,
            )
        )

ensure_partitions

ensure_partitions(
    connection: Connection,
    tables: AuditTables,
    severities: Iterable[int],
    *,
    months_ahead: int = 3,
    now: datetime | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]

Create the missing severity and monthly partitions.

Covers the UTC month of now and the months_ahead months after it, for the transaction table and for every severity. Idempotent: a partition that already exists with the same or wider bounds, under any name, is left alone.

Runs in the connection's current transaction, which must not be AUTOCOMMIT; the caller commits, or rolls back on error. Concurrent calls are serialized with a transaction-level advisory lock.

Parameters:

Name	Type	Description	Default
connection	Connection	Connection with DDL privileges.	required
tables	AuditTables	The audit tables, from audit_trail.tables.build_tables.	required
severities	Iterable[int]	A severity IntEnum class, or any iterable of ints.	required
months_ahead	int	Months to create after the current one.	3
now	datetime | None	Reference time. None uses the database's now().	None
lock_timeout	str | None	PostgreSQL lock_timeout for this transaction, such as "5s". None keeps the session's setting. It bounds each lock wait, not the whole call: blocked behind drop_expired (the advisory lock) and then behind an open transaction (the CREATE), a call can take about twice lock_timeout.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions created, parents first.

Raises:

Type	Description
ValueError	If months_ahead is negative or the connection is in AUTOCOMMIT mode.
PartitionError	If a parent table does not exist, or a monthly partition name derived from an existing severity partition would be longer than PostgreSQL allows.
PartitionLockTimeoutError	If a lock was not granted within lock_timeout; the transaction must be rolled back.

Source code in audit_trail/maintenance.py

def ensure_partitions(
    connection: Connection,
    tables: AuditTables,
    severities: Iterable[int],
    *,
    months_ahead: int = 3,
    now: datetime | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]:
    """Create the missing severity and monthly partitions.

    Covers the UTC month of ``now`` and the ``months_ahead`` months after it,
    for the transaction table and for every severity. Idempotent: a partition
    that already exists with the same or wider bounds, under any name, is
    left alone.

    Runs in the connection's current transaction, which must not be
    ``AUTOCOMMIT``; the caller commits, or rolls back on error. Concurrent
    calls are serialized with a transaction-level advisory lock.

    Args:
        connection: Connection with DDL privileges.
        tables: The audit tables, from ``audit_trail.tables.build_tables``.
        severities: A severity ``IntEnum`` class, or any iterable of ints.
        months_ahead: Months to create after the current one.
        now: Reference time. ``None`` uses the database's ``now()``.
        lock_timeout: PostgreSQL ``lock_timeout`` for this transaction, such
            as ``"5s"``. ``None`` keeps the session's setting. It bounds each
            lock wait, not the whole call: blocked behind ``drop_expired``
            (the advisory lock) and then behind an open transaction (the
            ``CREATE``), a call can take about twice ``lock_timeout``.

    Returns:
        Schema-qualified names of the partitions created, parents first.

    Raises:
        ValueError: If ``months_ahead`` is negative or the connection is in
            ``AUTOCOMMIT`` mode.
        PartitionError: If a parent table does not exist, or a monthly
            partition name derived from an existing severity partition would
            be longer than PostgreSQL allows.
        PartitionLockTimeoutError: If a lock was not granted within
            ``lock_timeout``; the transaction must be rolled back.
    """
    if months_ahead < 0:
        raise ValueError("months_ahead must not be negative")
    if getattr(connection.connection.dbapi_connection, "autocommit", False):
        raise ValueError("ensure_partitions needs a transaction, not AUTOCOMMIT")
    try:
        return _ensure(
            connection,
            tables,
            severity_values(severities),
            months_ahead,
            now,
            lock_timeout,
        )
    except DBAPIError as exc:
        if getattr(exc.orig, "sqlstate", None) == LOCK_NOT_AVAILABLE:
            raise PartitionLockTimeoutError(
                f"partitions were not created: a lock was not granted within "
                f"lock_timeout={lock_timeout!r}, which applies to each lock "
                "wait, so waiting behind drop_expired and then an open "
                "transaction takes about twice as long; retry later"
            ) from exc
        raise

drop_expired

drop_expired(
    connection: Connection,
    tables: AuditTables,
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    now: datetime | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]

Detach and drop the monthly partitions past their retention.

A partition is dropped when its upper bound is earlier than now - retention; one whose upper bound is exactly that instant is kept. Each partition is detached with DETACH PARTITION ... CONCURRENTLY from its parent (the transaction table or a severity partition) and then dropped, one at a time. Detaches left pending by an interrupted call are finalized first. Concurrent calls, and ensure_partitions, are serialized with an advisory lock.

DETACH ... CONCURRENTLY cannot run inside a transaction block, so connection must be in AUTOCOMMIT mode. lock_timeout is set for the session during the call and restored afterwards. The locks taken are listed in the module documentation. Waiting for application transactions that have used an audit table counts against lock_timeout, so a long open transaction makes this call fail; that is expected, and retrying later is safe.

Parameters:

Name	Type	Description	Default
connection	Connection	AUTOCOMMIT connection with DDL privileges.	required
tables	AuditTables	The audit tables, from audit_trail.tables.build_tables.	required
retention	Mapping[_Severity, timedelta | None]	Retention by severity value. None keeps a severity forever. A severity that has a partition but no key here is also kept forever, and a warning names it, so an empty mapping drops no activity partition. A severity partition holding several values keeps the longest of their retentions.	required
transaction_retention	timedelta | None	Retention of the transaction table. It is capped at the shortest finite severity retention, with a warning if it is longer; None uses that shortest retention. When no severity has a finite retention and this is None, transaction partitions are kept.	None
now	datetime | None	Reference time, timezone-aware. None uses the database's now().	None
lock_timeout	str | None	PostgreSQL lock_timeout for each statement, such as "5s". None keeps the session's setting.	'5s'

Returns:

Type	Description
list[str]	Schema-qualified names of the partitions dropped, in order.

Raises:

Type	Description
ValueError	If the connection is not in AUTOCOMMIT mode, a retention is negative, or now is naive.
PartitionError	If a parent table does not exist.
PartitionLockTimeoutError	If a lock was not granted within lock_timeout. Partitions dropped before that stay dropped.

Source code in audit_trail/maintenance.py

def drop_expired(
    connection: Connection,
    tables: AuditTables,
    retention: Mapping[_Severity, timedelta | None],
    *,
    transaction_retention: timedelta | None = None,
    now: datetime | None = None,
    lock_timeout: str | None = "5s",
) -> list[str]:
    """Detach and drop the monthly partitions past their retention.

    A partition is dropped when its upper bound is earlier than
    ``now - retention``; one whose upper bound is exactly that instant is
    kept. Each partition is detached with ``DETACH PARTITION ...
    CONCURRENTLY`` from its parent (the transaction table or a severity
    partition) and then dropped, one at a time. Detaches left pending by an
    interrupted call are finalized first. Concurrent calls, and
    ``ensure_partitions``, are serialized with an advisory lock.

    ``DETACH ... CONCURRENTLY`` cannot run inside a transaction block, so
    ``connection`` must be in ``AUTOCOMMIT`` mode. ``lock_timeout`` is set for
    the session during the call and restored afterwards. The locks taken are
    listed in the module documentation. Waiting for application transactions
    that have used an audit table counts against ``lock_timeout``, so a long
    open transaction makes this call fail; that is expected, and retrying
    later is safe.

    Args:
        connection: ``AUTOCOMMIT`` connection with DDL privileges.
        tables: The audit tables, from ``audit_trail.tables.build_tables``.
        retention: Retention by severity value. ``None`` keeps a severity
            forever. A severity that has a partition but no key here is also
            kept forever, and a warning names it, so an empty mapping drops no
            activity partition. A severity partition holding several values
            keeps the longest of their retentions.
        transaction_retention: Retention of the transaction table. It is
            capped at the shortest finite severity retention, with a warning
            if it is longer; ``None`` uses that shortest retention. When no
            severity has a finite retention and this is ``None``, transaction
            partitions are kept.
        now: Reference time, timezone-aware. ``None`` uses the database's
            ``now()``.
        lock_timeout: PostgreSQL ``lock_timeout`` for each statement, such as
            ``"5s"``. ``None`` keeps the session's setting.

    Returns:
        Schema-qualified names of the partitions dropped, in order.

    Raises:
        ValueError: If the connection is not in ``AUTOCOMMIT`` mode, a
            retention is negative, or ``now`` is naive.
        PartitionError: If a parent table does not exist.
        PartitionLockTimeoutError: If a lock was not granted within
            ``lock_timeout``. Partitions dropped before that stay dropped.
    """
    if not getattr(connection.connection.dbapi_connection, "autocommit", False):
        raise ValueError(
            "drop_expired needs an AUTOCOMMIT connection: DETACH PARTITION "
            "... CONCURRENTLY cannot run inside a transaction block"
        )
    if any(r is not None and r < timedelta(0) for r in retention.values()) or (
        transaction_retention is not None and transaction_retention < timedelta(0)
    ):
        raise ValueError("retention must not be negative")
    _check_aware(now)
    try:
        return _drop(
            connection, tables, retention, transaction_retention, now, lock_timeout
        )
    except DBAPIError as exc:
        if getattr(exc.orig, "sqlstate", None) == LOCK_NOT_AVAILABLE:
            raise PartitionLockTimeoutError(
                f"expired partitions were not all dropped: a lock was not "
                f"granted within lock_timeout={lock_timeout!r}, which applies to "
                "each lock wait. An application "
                "transaction left open on an audit table for longer than that "
                "causes this and is expected; retry later, and a detach left "
                "pending is finalized then"
            ) from exc
        raise

health

health(
    connection: Connection,
    tables: AuditTables,
    severities: Iterable[int],
    *,
    min_months_ahead: int = 2,
    now: datetime | None = None,
) -> HealthReport

Report partition coverage, pending detaches and orphaned partitions.

Read-only; runs on any connection.

Parameters:

Name	Type	Description	Default
connection	Connection	Connection that can read the catalog.	required
tables	AuditTables	The audit tables, from audit_trail.tables.build_tables.	required
severities	Iterable[int]	A severity IntEnum class, or any iterable of ints.	required
min_months_ahead	int	Coverage below this many months after the current one is flagged. The default of 2 flags one missed run of ensure_partitions(months_ahead=3).	2
now	datetime | None	Reference time, timezone-aware. None uses the database's now().	None

Returns:

Type	Description
HealthReport	The report.

Raises:

Type	Description
ValueError	If now is naive.
PartitionError	If a parent table does not exist.

Source code in audit_trail/maintenance.py

def health(
    connection: Connection,
    tables: AuditTables,
    severities: Iterable[int],
    *,
    min_months_ahead: int = 2,
    now: datetime | None = None,
) -> HealthReport:
    """Report partition coverage, pending detaches and orphaned partitions.

    Read-only; runs on any connection.

    Args:
        connection: Connection that can read the catalog.
        tables: The audit tables, from ``audit_trail.tables.build_tables``.
        severities: A severity ``IntEnum`` class, or any iterable of ints.
        min_months_ahead: Coverage below this many months after the current
            one is flagged. The default of ``2`` flags one missed run of
            ``ensure_partitions(months_ahead=3)``.
        now: Reference time, timezone-aware. ``None`` uses the database's
            ``now()``.

    Returns:
        The report.

    Raises:
        ValueError: If ``now`` is naive.
        PartitionError: If a parent table does not exist.
    """
    _check_aware(now)
    transaction_oid = _oid(connection, _qualified(tables.transaction))
    activity_oid = _oid(connection, _qualified(tables.activity))
    if now is None:
        now = connection.execute(text("SELECT now()")).scalar_one()

    transaction_months = _children(connection, transaction_oid)
    severity_parents = _children(connection, activity_oid)
    months = {
        parent.oid: _children(connection, parent.oid)
        for parent in severity_parents
        if parent.relkind == "p"
    }
    by_severity: dict[int, _Child] = {}
    for parent in severity_parents:
        for value in parent.list_values():
            by_severity.setdefault(value, parent)

    def check(table: str | None, children: list[_Child] | None) -> PartitionHealth:
        if children is None:  # a plain table takes any date
            covers_now, months_ahead = True, None
        else:
            covers_now, months_ahead = _coverage(children, now)
        below = not covers_now or (
            months_ahead is not None and months_ahead < min_months_ahead
        )
        return PartitionHealth(table, covers_now, months_ahead, below)

    activity: dict[int, PartitionHealth] = {}
    for value in severity_values(severities):
        severity_parent = by_severity.get(value)
        if severity_parent is None:
            activity[value] = check(None, [])
        else:
            activity[value] = check(
                severity_parent.display, months.get(severity_parent.oid)
            )

    return HealthReport(
        min_months_ahead=min_months_ahead,
        transaction=check(
            _name(
                _shown_schemas(connection, tables).transaction, tables.transaction.name
            ),
            transaction_months,
        ),
        activity=activity,
        pending_detach=[
            child.display
            for children in (transaction_months, *months.values())
            for child in children
            if child.pending
        ],
        orphaned=_orphans(connection, tables, severity_parents),
    )