db-retry

https://github.com/modern-python/db-retry
Admin
A Python library providing robust retry mechanisms, connection utilities, and transaction helpers...
Tokens:32,994
Snippets:224
Trust Score:9
License:MIT
Update:1 week ago
Show doc for...
Tokens:
Raw
### Install db-retry with pip

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Installs the db-retry package using pip. Use this in environments not managed by uv.

```bash
pip install db-retry
```

--------------------------------

### Install db-retry with uv

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Adds the db-retry package to a project managed by uv. Run this command in the project directory.

```bash
uv add db-retry
```

--------------------------------

### build_connection_factory

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Creates a connection factory for multi-host setups.

```APIDOC
## build_connection_factory

### Description
Creates a connection factory for multi-host setups.

### Parameters
- `url` (str) - Required - The database URL.
- `timeout` (float) - Required - Connection timeout in seconds.

### Returns
A connection factory object.
```

--------------------------------

### Full configuration in one place

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/configuration.md

Shows a complete setup for db-retry: sets the DB_RETRY_RETRIES_NUMBER environment variable for the bare @postgres_retry decorator, builds a DSN with build_db_dsn, creates an async engine with a connection factory and timeout, and applies a per-callsite override with @postgres_retry(retries=8).

```python
import os

from sqlalchemy import make_url
from sqlalchemy.ext.asyncio import create_async_engine

from db_retry import build_connection_factory, build_db_dsn, postgres_retry, Transaction

# 1. Attempt budget for bare @postgres_retry (env; re-read per call)
os.environ["DB_RETRY_RETRIES_NUMBER"] = "5"

# 2. DSN configuration: database name, driver, read/write targeting
dsn = build_db_dsn(
    db_dsn="postgresql://user:pass@/db_placeholder?host=primary:5432&host=backup:5432",
    database_name="app_db",
    drivername="postgresql+asyncpg",
)

# 3. Connection factory configuration: connect timeout
engine = create_async_engine(dsn, async_creator=build_connection_factory(url=dsn, timeout=5.0))

# 4. Per-callsite overrides
@postgres_retry(retries=8)
async def critical_write(session) -> None:
    async with Transaction(session=session, isolation_level="SERIALIZABLE"):
        ...
```

--------------------------------

### Building a runtime DSN for a multi-host connection factory

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/dsn.md

Shows how to use build_db_dsn, build_connection_factory, and is_dsn_multihost to construct a DSN for a multi-host connection. The example asserts that target_session_attrs is 'read-write' for the primary and 'prefer-standby' for replicas.

```python
from db_retry import build_db_dsn, build_connection_factory, is_dsn_multihost

multi_host_dsn = "postgresql://user:password@/db_placeholder?host=primary-db:5432&host=secondary-db:5432"

if not is_dsn_multihost(multi_host_dsn):
    raise RuntimeError("expected a multi-host DSN")

dsn = build_db_dsn(
    db_dsn=multi_host_dsn,
    database_name="production_database",  # replaces db_placeholder
    drivername="postgresql+asyncpg",
)
assert dsn.query["target_session_attrs"] == "read-write"

# Same DSN for read replicas; prefer-standby routes to a standby
replica_dsn = build_db_dsn(db_dsn=multi_host_dsn, database_name="production_database", use_replica=True)
assert replica_dsn.query["target_session_attrs"] == "prefer-standby"
```

--------------------------------

### Read a pull request with comments and diff

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh pr view <number> --comments` to read a PR and its comments, and `gh pr diff <number>` to get the diff.

```bash
gh pr view <number> --comments
```

```bash
gh pr diff <number>
```

--------------------------------

### get_retries_number

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Gets the number of retries from settings.

```APIDOC
## get_retries_number

### Description
Gets the number of retries from settings.

### Module
`db_retry.settings`
```

--------------------------------

### RETRIABLE_ASYNCPG_ERRORS

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/retriable.md

A set of asyncpg exceptions that are considered retriable. Use this to inspect the taxonomy of retriable errors, for example to assert that a specific error is included.

```APIDOC
## RETRIABLE_ASYNCPG_ERRORS

### Description
A set of asyncpg exceptions that are considered retriable. Use this to inspect the taxonomy of retriable errors, for example to assert that a specific error is included.

### Type
- **set** - A set of exception classes.

### Example
```python
from db_retry.retriable import RETRIABLE_ASYNCPG_ERRORS

assert asyncpg.SerializationError in RETRIABLE_ASYNCPG_ERRORS
# 40003 (statement outcome unknown) is deliberately not retriable
assert not is_retriable(asyncpg.StatementCompletionUnknownError())
```
```

--------------------------------

### Build multi-host connection factory with build_connection_factory

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Creates an asyncpg connection factory for high availability across multiple hosts. build_db_dsn sets the database name, driver, and target_session_attrs. On a connect timeout, the factory tries each host in turn.

```python
from sqlalchemy.ext.asyncio import create_async_engine
from db_retry import build_connection_factory, build_db_dsn

# Configure multiple database hosts for high availability
multi_host_dsn = "postgresql://user:password@/myapp_db?host=primary-db:5432&host=secondary-db:5432&host=backup-db:5432"

# Set the database name, driver and target_session_attrs
dsn = build_db_dsn(db_dsn=multi_host_dsn, database_name="production_database", drivername="postgresql+asyncpg")

# Create connection factory with timeout
connection_factory = build_connection_factory(
    url=dsn,
    timeout=5.0,  # 5 second connection timeout
)

# On a connect timeout, the factory tries each host in turn
engine = create_async_engine(dsn, async_creator=connection_factory)
```

--------------------------------

### build_connection_plan / parse_connect_args

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/configuration.md

Internal helpers to build a connection plan or parse connect arguments from a SQLAlchemy URL. These are not re-exported at the top level.

```APIDOC
## build_connection_plan / parse_connect_args

### Description
Internal helpers to build a connection plan or parse connect arguments from a SQLAlchemy URL. These are not re-exported at the top level.

### Parameters
- `url` (sqlalchemy.URL) - Required - Parsed URL. `build_connection_plan` returns a `ConnectionPlan`; `parse_connect_args` returns `(connect_args_dict, [(host, port), ...])`.

### Usage
```python
from db_retry.connections import build_connection_plan, parse_connect_args

plan = build_connection_plan(url)
connect_args, hosts = parse_connect_args(url)
```
```

--------------------------------

### build_db_dsn

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Builds a DSN with the given database name and driver, and sets target_session_attrs to read-write or prefer-standby.

```APIDOC
## build_db_dsn

### Description
Builds a DSN with the given database name and driver, and sets `target_session_attrs` to `read-write`, or to `prefer-standby` when `use_replica=True`.

### Parameters
- `db_dsn` (str) - Required - The base DSN.
- `database_name` (str) - Required - The database name to use.
- `use_replica` (bool) - Optional - If True, sets `target_session_attrs` to `prefer-standby`. Defaults to False.
- `drivername` (str) - Optional - The driver name. Defaults to `"postgresql"`.

### Returns
The built DSN string.
```

--------------------------------

### Recovery of a lost pooled connection

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

On entry, Transaction rolls back a session whose transaction is dead, such as when a pooled connection died while idle (asyncpg.InterfaceError: connection is closed surfaced as DBAPIError with connection_invalidated). Re-entering Transaction on the next retry attempt starts the block on a fresh connection.

```python
@postgres_retry(retries=2)
async def select_one() -> int | None:
    nonlocal attempts
    attempts += 1
    async with transaction:  # 1st attempt: rolls back dead txn, begins on a fresh connection
        return await session.scalar(sqlalchemy.text("SELECT 1"))
```

--------------------------------

### build_connection_factory

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/configuration.md

Builds a connection factory for PostgreSQL connections. It takes a SQLAlchemy URL and a timeout, and returns a factory that creates connections with the specified timeout.

```APIDOC
## build_connection_factory

### Description
Builds a connection factory for PostgreSQL connections. It takes a SQLAlchemy URL and a timeout, and returns a factory that creates connections with the specified timeout.

### Parameters
- `url` (sqlalchemy.URL) - Required - Parsed URL (a `sqlalchemy.URL` object, not a string). Multi-host form: `postgresql+asyncpg://user:pass@/db?host=h1:5432&host=h2:5433`; single-host form is also accepted.
- `timeout` (float) - Required - Connect timeout in seconds, forwarded to every `asyncpg.connect(timeout=...)` call (bulk attempt and each host-by-host attempt).

### Usage
```python
from db_retry import build_connection_factory

factory = build_connection_factory(url, timeout=5.0)
```
```

--------------------------------

### Multi-host failover with create_async_engine

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Builds a DSN from a multi-host URL and a connection factory, then passes the factory as async_creator to create_async_engine. The factory handles failover across primary-db, secondary-db, and backup-db hosts with a 5.0 second timeout.

```python
from sqlalchemy.ext.asyncio import create_async_engine
from db_retry import build_db_dsn, build_connection_factory

multi_host_dsn = "postgresql://user:password@/myapp_db?host=primary-db:5432&host=secondary-db:5432&host=backup-db:5432"
dsn = build_db_dsn(db_dsn=multi_host_dsn, database_name="production_database", drivername="postgresql+asyncpg")

connection_factory = build_connection_factory(url=dsn, timeout=5.0)

engine = create_async_engine(dsn, async_creator=connection_factory)
```

--------------------------------

### Retry database operations with postgres_retry and Transaction

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Retries a coroutine on serialization failures, deadlocks, and lost connections. The retried coroutine must own its transaction from start to finish, e.g., by opening it with Transaction, because Postgres discards statements before the failure and SQLAlchemy raises PendingRollbackError otherwise. Total attempts come from DB_RETRY_RETRIES_NUMBER (default 3: first call plus 2 retries).

```python
import asyncio
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from db_retry import Transaction, postgres_retry


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(sa.String())
    email: Mapped[str] = mapped_column(sa.String(), index=True)


# Total attempts come from DB_RETRY_RETRIES_NUMBER (default 3: the first call plus 2 retries)
@postgres_retry
async def get_user_by_email(session: AsyncSession, email: str) -> User | None:
    async with Transaction(session=session):
        return await session.scalar(sa.select(User).where(User.email == email))


async def main():
    engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
    async with AsyncSession(engine) as session:
        # Automatically retries on connection failures, serialization errors, or deadlocks
        user = await get_user_by_email(session, "john.doe@example.com")
        if user:
            print(f"Found user: {user.name}")


asyncio.run(main())
```

--------------------------------

### build_db_dsn

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/configuration.md

Builds a SQLAlchemy URL from a DSN string, database name, and optional replica flag. It replaces the placeholder in the DSN with the given database name and sets the target_session_attrs based on the use_replica flag.

```APIDOC
## build_db_dsn

### Description
Builds a SQLAlchemy URL from a DSN string, database name, and optional replica flag. It replaces the placeholder in the DSN with the given database name and sets the target_session_attrs based on the use_replica flag.

### Parameters
- `db_dsn` (str) - Required - Input DSN string.
- `database_name` (str) - Required - Database name to write into the URL (replaces the placeholder).
- `use_replica` (bool) - Optional - `True` → `target_session_attrs=prefer-standby`; `False` → `target_session_attrs=read-write`. Always wins over any value already in the DSN.
- `drivername` (str) - Optional - Driver name on the returned URL, e.g. `"postgresql+asyncpg"`.

### Usage
```python
from db_retry import build_db_dsn

url = build_db_dsn(db_dsn, database_name, use_replica=False)
```
```

--------------------------------

### build_connection_plan function implementation

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Builds a ConnectionPlan from a SQLAlchemy URL. Pops target_session_attrs from connect args and converts it to a SessionAttribute, removes host and port when multi-host, wraps connect_args in a read-only MappingProxyType, and stores host/port pairs as the failover tuple. Raises the same exceptions as parse_connect_args.

```python
def build_connection_plan(url: sqlalchemy.URL) -> ConnectionPlan:
    connect_args, hosts_and_ports = parse_connect_args(url)
    raw_target_session_attrs: str | None = connect_args.pop("target_session_attrs", None)
    target_session_attrs: SessionAttribute | None = (
        SessionAttribute(raw_target_session_attrs) if raw_target_session_attrs else None
    )
    if hosts_and_ports:
        del connect_args["host"], connect_args["port"]
    return ConnectionPlan(
        connect_args=types.MappingProxyType(connect_args),
        target_session_attrs=target_session_attrs,
        failover=tuple(hosts_and_ports),
    )
```

--------------------------------

### Direct factory use with error handling

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Creates a connection factory directly from a make_url DSN with a 2.0 second timeout and awaits it inside a try block. Catches asyncpg.TargetServerAttributeNotMatched, which is raised when no host matches target_session_attrs requirements after all hosts are exhausted.

```python
from sqlalchemy import make_url
from db_retry.connections import build_connection_factory

factory = build_connection_factory(url=make_url("postgresql+asyncpg://user:pass@/db?host=h1:5432&host=h2:5432"), timeout=2.0)

try:
    connection = await factory()
except asyncpg.TargetServerAttributeNotMatched as exc:
    print("no host matched target_session_attrs requirements", exc)
```

--------------------------------

### Package entry point re-exports

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Shows the five symbols re-exported from the db_retry package's __init__.py, including the __all__ list. This is the public API surface for the library.

```python
from db_retry.connections import build_connection_factory
from db_retry.dsn import build_db_dsn, is_dsn_multihost
from db_retry.retry import postgres_retry
from db_retry.transaction import Transaction


__all__ = [
    "Transaction",
    "build_connection_factory",
    "build_db_dsn",
    "is_dsn_multihost",
    "postgres_retry",
]
```

--------------------------------

### Reading the configured attempt budget directly

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/settings.md

Shows how to read the attempt budget using get_retries_number, including setting the environment variable and verifying the default when unset.

```python
import os

from db_retry.settings import get_retries_number

os.environ.setdefault("DB_RETRY_RETRIES_NUMBER", "5")
assert get_retries_number() == 5

os.environ.pop("DB_RETRY_RETRIES_NUMBER", None)
assert get_retries_number() == 3  # default
```

--------------------------------

### build_connection_plan

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Builds a connection plan.

```APIDOC
## build_connection_plan

### Description
Builds a connection plan.

### Module
`db_retry.connections`
```

--------------------------------

### Create a serializable transaction with Transaction

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Use the serializable isolation level to prevent race conditions. The Transaction is created with an AsyncSession and isolation_level='SERIALIZABLE'. The session is created with create_async_engine using a PostgreSQL async URL.

```python
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from db_retry import Transaction


async def main():
    engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")

    async with AsyncSession(engine) as session:
        strict_transaction = Transaction(
            session=session,
            isolation_level="SERIALIZABLE",
        )
        # use strict_transaction where needed
```

--------------------------------

### parse_connect_args(url: sqlalchemy.URL) -> tuple[dict[str, typing.Any], list[tuple[str, int]]]

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Parses a SQLAlchemy URL and returns the asyncpg connect arguments along with a list of (host, port) pairs for multi-host DSNs. Returns an empty list for single-host DSNs.

```APIDOC
## parse_connect_args(url: sqlalchemy.URL) -> tuple[dict[str, typing.Any], list[tuple[str, int]]]

### Description
Parses a SQLAlchemy URL and returns the asyncpg connect arguments along with a list of (host, port) pairs for multi-host DSNs. Returns an empty list for single-host DSNs.

### Parameters
- **url** (sqlalchemy.URL) - Required - A parsed SQLAlchemy URL (e.g. from `sqlalchemy.make_url` or `engine.url`).

### Return
A tuple of the asyncpg connect args dict produced by `PGDialect_asyncpg().create_connect_args(url)[1]`, plus a list of `(host, port)` pairs. The pairs are built only when the asyncpg dialect produced parallel lists for `host` and `port` (the multi-host DSN forms, e.g. `?host=host1:5432&host=host2:5433` or `?host=host1,host2&port=5432,5433`); the zip is `strict=True`, so mismatched host/port list lengths fail with `ValueError`. For a single-host DSN the return is `(connect_args, [])` — the empty list is the "not multi-host" signal. Note that the first element shares the dict with `create_connect_args`; callers that mutate it change the underlying dialect output.

### Throws
A malformed multi-host DSN can raise `sqlalchemy.exc.ArgumentError` from `create_connect_args` (e.g. `?host=host1&host=host2` — repeated hosts without ports), or `ValueError` from `zip(..., strict=True)` if the host and port lists have different lengths.
```

--------------------------------

### build_connection_factory(url, timeout)

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Builds an async callable that creates an asyncpg connection with failover support. The returned factory handles single-host and multi-host DSNs, shuffling hosts and retrying on failure.

```APIDOC
## build_connection_factory(url, timeout)

### Description
Builds an async callable that creates an asyncpg connection with failover support. The returned factory handles single-host and multi-host DSNs, shuffling hosts and retrying on failure.

### Parameters
- `url` (sqlalchemy.URL) - Required - Parsed SQLAlchemy URL. Must be a `sqlalchemy.URL` object, not a string.
- `timeout` (float) - Required - Connect timeout, in seconds, passed to every `asyncpg.connect(..., timeout=timeout)` call.

### Return
An async callable (`async def _connection_factory() -> asyncpg.Connection`) meant for `AsyncEngine(..., async_creator=...)` or direct `await`-ing.

### Behavior
- **Single-host DSN** (`plan.failover` empty): connects once via `asyncpg.connect(**plan.connect_args, host=..., port=..., timeout=..., target_session_attrs=...)`. Errors propagate immediately (e.g. `TimeoutError`).
- **Multi-host DSN**, stage 1 — *bulk attempt*: all hosts are shuffled (`random.sample(failover, len(failover))`, a fresh shuffle per connection, host/port pairs kept together) and handed to one `asyncpg.connect(host=[...], port=[...])` call so asyncpg walks the list itself. Only `TimeoutError` triggers the fallback.
- **Multi-host DSN**, stage 2 — *failover*: with a fresh shuffle, each `(host, port)` is tried individually. `TimeoutError`, `OSError`, and `asyncpg.TargetServerAttributeNotMatched` are caught per host (warning logged: `Failed to fetch asyncpg connection from %s, %s`); any other exception propagates immediately.
- **Exhaustion**: if every host fails, raises `asyncpg.TargetServerAttributeNotMatched` with the message `None of the hosts match the target attribute requirement {plan.target_session_attrs}`.
```

--------------------------------

### build_connection_factory

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Builds a connection factory that returns an awaitable Connection.

```APIDOC
## build_connection_factory

### Description
Builds a connection factory that returns an awaitable Connection.

### Signature
`def build_connection_factory(url: sqlalchemy.URL, timeout: float) -> Callable[[], Awaitable[Connection]]`

### Parameters
- **url** (sqlalchemy.URL) - Required - The database URL.
- **timeout** (float) - Required - Timeout in seconds.

### Returns
- Callable that returns an Awaitable[Connection].
```

--------------------------------

### build_connection_factory

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Creates an async connection factory that attempts to connect to multiple hosts with failover. The factory is used as the `async_creator` for SQLAlchemy's `create_async_engine` or can be called directly to obtain a connection.

```APIDOC
## build_connection_factory

### Description
Creates an async connection factory that attempts to connect to multiple hosts with failover. The factory is used as the `async_creator` for SQLAlchemy's `create_async_engine` or can be called directly to obtain a connection.

### Parameters
- **url** (URL) - Required - The SQLAlchemy URL object (e.g., from `make_url`).
- **timeout** (float) - Required - The connection timeout in seconds.

### Returns
- **Callable** - An async callable that returns a connection.

### Throws
- `TimeoutError` - Single-host connect timeout; or any error outside the per-host catch list during failover stage 2.
- `asyncpg.TargetServerAttributeNotMatched` - Raised by asyncpg when a host's server attributes do not match the requirement; re-raised by the factory when all hosts are exhausted.
- `OSError` - Per-host network-level failures during stage 2 (caught and tried on the next host; only the final exhaustion error escapes).
- `sqlalchemy.exc.ArgumentError` / `ValueError` - Malformed DSN, from `build_connection_plan` at factory construction time.

### Example (with create_async_engine)
```python
from sqlalchemy.ext.asyncio import create_async_engine
from db_retry import build_db_dsn, build_connection_factory

multi_host_dsn = "postgresql://user:password@/myapp_db?host=primary-db:5432&host=secondary-db:5432&host=backup-db:5432"
dsn = build_db_dsn(db_dsn=multi_host_dsn, database_name="production_database", drivername="postgresql+asyncpg")

connection_factory = build_connection_factory(url=dsn, timeout=5.0)

engine = create_async_engine(dsn, async_creator=connection_factory)
```

### Example (direct use with error handling)
```python
from sqlalchemy import make_url
from db_retry.connections import build_connection_factory

factory = build_connection_factory(url=make_url("postgresql+asyncpg://user:pass@/db?host=h1:5432&host=h2:5432"), timeout=2.0)

try:
    connection = await factory()
except asyncpg.TargetServerAttributeNotMatched as exc:
    print("no host matched target_session_attrs requirements", exc)
```
```

--------------------------------

### build_db_dsn

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Builds a SQLAlchemy URL from a DSN string and database name.

```APIDOC
## build_db_dsn

### Description
Builds a SQLAlchemy URL from a DSN string and database name.

### Signature
`def build_db_dsn(db_dsn: str, database_name: str, use_replica: bool = False, drivername: str = "postgresql") -> sa.URL`

### Parameters
- **db_dsn** (str) - Required - The DSN string.
- **database_name** (str) - Required - The database name.
- **use_replica** (bool) - Optional - Whether to use replica.
- **drivername** (str) - Optional - The driver name.

### Returns
- sa.URL - The SQLAlchemy URL.
```

--------------------------------

### parse_connect_args

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Parses connection arguments.

```APIDOC
## parse_connect_args

### Description
Parses connection arguments.

### Module
`db_retry.connections`
```

--------------------------------

### Call site in postgres_retry

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/settings.md

Shows how the decorator uses get_retries_number to supply the default attempt budget when retries is not provided.

```python
attempts = retries if retries is not None else settings.get_retries_number()
```

--------------------------------

### async def __aenter__ - Transaction context manager entry

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

Enters the async context manager. Rolls back a dead session, applies isolation_level via execution options if set, begins a transaction if not already in one, and returns self. Recovers sessions with invalidated pooled connections.

```python
    async def __aenter__(self) -> typing.Self:
        if _is_dead(self.session):
            await self.session.rollback()

        if self.isolation_level:
            await self.session.connection(execution_options={"isolation_level": self.isolation_level})

        if not self.session.in_transaction():
            await self.session.begin()
        return self
```

--------------------------------

### Set DB_RETRY_RETRIES_NUMBER environment variable

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Configures the total number of attempts (first call plus retries) for the retry decorator. The default is 3.

```bash
export DB_RETRY_RETRIES_NUMBER=5
```

--------------------------------

### build_connection_factory function definition

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Builds an async connection factory that handles single-host and multi-host failover. For multi-host DSNs, it first attempts a bulk connection with shuffled hosts, then falls back to trying each host individually on TimeoutError, OSError, or TargetServerAttributeNotMatched. Raises TargetServerAttributeNotMatched if all hosts fail.

```python
def build_connection_factory(
    url: sqlalchemy.URL,
    timeout: float,
) -> typing.Callable[[], typing.Awaitable["ConnectionType"]]:
    plan: typing.Final = build_connection_plan(url)

    async def _connection_factory() -> "ConnectionType":
        if not plan.failover:
            return await _connect(plan, timeout)

        bulk_order: typing.Final = _shuffled(plan.failover)
        try:
            return await _connect(
                plan,
                timeout,
                host=list(map(itemgetter(0), bulk_order)),
                port=list(map(itemgetter(1), bulk_order)),
            )
        except TimeoutError:
            logger.warning("Failed to fetch asyncpg connection. Trying host by host.")

        for host, port in _shuffled(plan.failover):
            try:
                return await _connect(plan, timeout, host=host, port=port)
            except (TimeoutError, OSError, asyncpg.TargetServerAttributeNotMatched) as exc:
                logger.warning("Failed to fetch asyncpg connection from %s, %s", host, exc)
        msg: typing.Final = f"None of the hosts match the target attribute requirement {plan.target_session_attrs}"
        raise asyncpg.TargetServerAttributeNotMatched(msg)

    return _connection_factory
```

--------------------------------

### ConnectionPlan frozen dataclass definition

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Defines an immutable connection plan with connect arguments, target session attributes, and failover host/port pairs. The dataclass is frozen with slots, and connect_args is a mapping proxy at runtime.

```python
@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
class ConnectionPlan:
    connect_args: typing.Mapping[str, typing.Any]
    target_session_attrs: SessionAttribute | None
    failover: tuple[tuple[str, int], ...]
```

--------------------------------

### List open GitHub issues with details

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh issue list` with `--state open` and JSON output to list issues. The `--jq` filter extracts number, title, body, labels, and comments. Apply `--label` and `--state` filters as needed.

```bash
gh issue list --state open --json number,title,body,labels,comments --jq '[.[] | {number, title, body, labels: [.labels[].name], comments: [.comments[].body]}]'
```

--------------------------------

### build_db_dsn

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Builds a database DSN from a multi-host DSN, database name, and driver name. This is a utility function for constructing the DSN used with the connection factory.

```APIDOC
## build_db_dsn

### Description
Builds a database DSN from a multi-host DSN, database name, and driver name.

### Parameters
- **db_dsn** (str) - Required - The multi-host DSN.
- **database_name** (str) - Required - The name of the database.
- **drivername** (str) - Required - The driver name (e.g., 'postgresql+asyncpg').

### Returns
- **str** - The constructed DSN.

### Example
```python
from db_retry import build_db_dsn

dsn = build_db_dsn(db_dsn=multi_host_dsn, database_name="production_database", drivername="postgresql+asyncpg")
```
```

--------------------------------

### parse_connect_args function definition

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/connections.md

Parses a SQLAlchemy URL into asyncpg connect arguments and a list of (host, port) pairs. Returns an empty list for single-host DSNs. Raises ValueError if host and port lists have mismatched lengths.

```python
def parse_connect_args(url: sqlalchemy.URL) -> tuple[dict[str, typing.Any], list[tuple[str, int]]]:
    connect_args: typing.Final[dict[str, typing.Any]] = PGDialect_asyncpg().create_connect_args(url)[1]
    hosts: typing.Final = connect_args.get("host")
    ports: typing.Final = connect_args.get("port")
    if isinstance(hosts, list) and isinstance(ports, list):
        return connect_args, list(zip(hosts, ports, strict=True))
    return connect_args, []
```

--------------------------------

### List open issues for frontier query

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

List the map's open children with `gh issue list --state open`, scoped to the map's sub-issues or task list. Drop any with an open blocker or an assignee.

```bash
gh issue list --state open
```

--------------------------------

### Resolve a ticket with answer and close

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Comment the answer on the issue, then close it. After closing, append a context pointer (gist + link) to the map's Decisions-so-far.

```bash
gh issue comment <n> --body "<answer>"
```

```bash
gh issue close <n>
```

--------------------------------

### Error handling with ArgumentError

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/dsn.md

Catches sqlalchemy.exc.ArgumentError when build_db_dsn receives an unparseable DSN string.

```python
from sqlalchemy.exc import ArgumentError
from db_retry import build_db_dsn

try:
    dsn = build_db_dsn(db_dsn="not a url", database_name="db")
except ArgumentError as exc:
    print(f"unparseable DSN: {exc}")
```

--------------------------------

### postgres_retry overload signatures

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/postgres-retry.md

Shows the exact overload signatures for the postgres_retry decorator. Supports both direct decoration and parameterized usage with an optional retries argument.

```python
@typing.overload
def postgres_retry(func: _Func, *, retries: int | None = ...) -> _Func: ...


@typing.overload
def postgres_retry(func: None = ..., *, retries: int | None = ...) -> _Decorator: ...


def postgres_retry(
    func: _Func | None = None,
    *,
    retries: int | None = None,
) -> _Func | _Decorator:
```

--------------------------------

### ConnectionPlan

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Type representing a connection plan.

```APIDOC
## ConnectionPlan

### Description
Type representing a connection plan.

### Module
`db_retry.connections`
```

--------------------------------

### Read a GitHub issue with comments

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh issue view <number> --comments` to read an issue and its comments. Filter comments with `jq` and also fetch labels.

```bash
gh issue view <number> --comments
```

--------------------------------

### Create a wayfinder map issue

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Create a single issue labelled `wayfinder:map` to hold the map body.

```bash
gh issue create --label wayfinder:map
```

--------------------------------

### Create a GitHub issue with gh CLI

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh issue create` to create a new issue. Use a heredoc for multi-line bodies. The repo is inferred from `git remote -v` automatically when run inside a clone.

```bash
gh issue create --title "..." --body "..."
```

--------------------------------

### @postgres_retry

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Decorator for async functions that should retry on database errors. Retries serialization failures, deadlocks, connection errors, and invalidated connections, with configurable total attempts.

```APIDOC
## @postgres_retry

### Description
Decorator for async functions that should retry on database errors. Retries serialization failure (40001), detected deadlock (40P01), connection error (class 08), and any DBAPIError marked with connection_invalidated. The retried coroutine must own its whole transaction.

### Usage
- `@postgres_retry` - Uses total attempts from `DB_RETRY_RETRIES_NUMBER` (default 3)
- `@postgres_retry(retries=N)` - Overrides total attempts per callsite

### Parameters
- `retries` (int) - Optional - Total attempts (first call plus retries). Defaults to `DB_RETRY_RETRIES_NUMBER`.

### Notes
- Retries also fire when the retriable asyncpg error is wrapped by advanced-alchemy's `wrap_sqlalchemy_exception()`.
- The retried coroutine must own its whole transaction; otherwise, `PendingRollbackError` is raised.
```

--------------------------------

### build_db_dsn

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/dsn.md

Parses a DSN string and returns a new SQLAlchemy URL with the database name, driver name, and target_session_attrs query parameter updated. The input DSN is not mutated; all other query parameters are preserved.

```APIDOC
## build_db_dsn

### Description
Parses a DSN string and returns a new SQLAlchemy URL with the database name, driver name, and target_session_attrs query parameter updated. The input DSN is not mutated; all other query parameters are preserved.

### Signature
```python
def build_db_dsn(
    db_dsn: str,
    database_name: str,
    use_replica: bool = False,
    drivername: str = "postgresql",
) -> sa.URL:
```

### Parameters
| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| `db_dsn` | `str` | Yes | — | DSN string, typically a multi-host fallback DSN of the form `postgresql://login:password@/db_placeholder?host=host1:5432&host=host2:5432`. |
| `database_name` | `str` | Yes | — | Replacement database name; replaces the placeholder (`db_placeholder` in the docs) via the URL's `database` component. |
| `use_replica` | `bool` | No | `False` | When `True`, sets `target_session_attrs=prefer-standby` (a replica may serve the query); when `False`, sets `target_session_attrs=read-write` (a writable node is required). |
| `drivername` | `str` | No | `"postgresql"` | Driver name to set on the returned URL, e.g. `"postgresql+asyncpg"`. |

### Return
`sqlalchemy.URL` — a new URL object (the input is not mutated) with `database` and `drivername` replaced and the query updated. Every other query parameter of the input DSN is carried through unchanged.

### Behavior Notes
- **`use_replica` always wins over the DSN**: the dict union order is `db_dsn_query | {"target_session_attrs": ...}`, and a value in the DSN never overrides the argument — `use_replica` alone decides the attribute. A DSN pinned to `read-write` still produces `prefer-standby` when `use_replica=True`, and vice versa.
- **All other query params pass through** — verified by tests for e.g. `application_name`.

### Throws
| Error type | Condition |
|-----------|-----------|
| `sqlalchemy.exc.ArgumentError` | The `db_dsn` string cannot be parsed by `sqlalchemy.make_url`. |
```

--------------------------------

### async def commit - Commit transaction

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

Commits the session's transaction by delegating to AsyncSession.commit.

```python
    async def commit(self) -> None:
        await self.session.commit()
```

--------------------------------

### Claim a ticket by assigning to self

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh issue edit <n> --add-assignee @me` to claim a ticket, which is the session's first write.

```bash
gh issue edit <n> --add-assignee @me
```

--------------------------------

### RETRIABLE_ASYNCPG_ERRORS

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Constant containing retriable asyncpg errors.

```APIDOC
## RETRIABLE_ASYNCPG_ERRORS

### Description
Constant containing retriable asyncpg errors.

### Module
`db_retry.retriable`
```

--------------------------------

### async def rollback - Rollback transaction

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

Rolls back the session's transaction by delegating to AsyncSession.rollback.

```python
    async def rollback(self) -> None:
        await self.session.rollback()
```

--------------------------------

### Error handling for an invalid value

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/settings.md

Demonstrates catching ValueError when DB_RETRY_RETRIES_NUMBER is set to a non-integer string, printing an error message.

```python
import os

os.environ["DB_RETRY_RETRIES_NUMBER"] = "many"

try:
    get_retries_number()
except ValueError as exc:
    print("DB_RETRY_RETRIES_NUMBER is not an integer", exc)
```

--------------------------------

### Transaction

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Context manager for transaction handling. On entry, rolls back a transaction that can no longer continue, otherwise joins or begins a transaction. On exit, rolls back if still in a transaction and closes the session.

```APIDOC
## Transaction

### Description
Context manager for transaction handling. On entry, rolls back a transaction that can no longer continue (a failed flush or a lost connection), otherwise joins the session's open transaction or begins one. On exit, rolls back if still in a transaction, then closes the session.

### Parameters
- `session` (Session) - Required - The SQLAlchemy session.
- `isolation_level` (str) - Optional - The isolation level to use.

### Usage
```python
with Transaction(session) as tx:
    # perform operations
```
```

--------------------------------

### Basic committed work with Transaction

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

Use Transaction as an async context manager to wrap a unit of work. Explicit commit is required; exiting the block rolls back otherwise.

```python
from sqlalchemy.ext.asyncio import AsyncSession
from db_retry import Transaction


async def transfer(session: AsyncSession, amount: int) -> None:
    async with Transaction(session=session):
        await session.execute(...)
        await transaction.commit()  # explicit commit required; exit rolls back otherwise
```

--------------------------------

### build_db_dsn function definition

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/dsn.md

Parses a DSN string and returns a new SQLAlchemy URL with the database name, driver name, and target_session_attrs query parameter updated. The input DSN is not mutated; all other query parameters are preserved. The use_replica argument always overrides any target_session_attrs in the DSN.

```python
def build_db_dsn(
    db_dsn: str,
    database_name: str,
    use_replica: bool = False,
    drivername: str = "postgresql",
) -> sa.URL:
    """Parse DSN variable and replace some parts.

    - DSN stored in format postgresql://login:password@/db_placeholder?host=host1:5432&host=host2:5432
    https://docs.sqlalchemy.org/en/20/dialects/postgresql.html#specifying-multiple-fallback-hosts
    - 'db_placeholder' is replaced here with service database name
    - `target_session_attrs` is chosen based on `use_replica` arg
    """
    parsed_db_dsn: typing.Final = sa.make_url(db_dsn)
    db_dsn_query: typing.Final[dict[str, typing.Any]] = dict(parsed_db_dsn.query or {})
    return parsed_db_dsn.set(
        database=database_name,
        drivername=drivername,
        query=db_dsn_query
        | {
            "target_session_attrs": "prefer-standby" if use_replica else "read-write",
        },
    )
```

--------------------------------

### Override retry attempts with postgres_retry(retries=5)

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Per-callsite override of the total number of attempts. The retries parameter replaces the default from DB_RETRY_RETRIES_NUMBER.

```python
@postgres_retry(retries=5)
async def create_order(session: AsyncSession, order: Order) -> Order: ...
```

--------------------------------

### List external pull requests for triage

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh pr list` with JSON output to list open PRs. Filter by `authorAssociation` to keep only `CONTRIBUTOR`, `FIRST_TIME_CONTRIBUTOR`, or `NONE`, dropping `OWNER`/`MEMBER`/`COLLABORATOR`.

```bash
gh pr list --state open --json number,title,body,labels,author,authorAssociation,comments
```

--------------------------------

### get_retries_number()

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/settings.md

Reads the configured attempt budget from the environment variable DB_RETRY_RETRIES_NUMBER. Returns the value as an integer, or 3 if the variable is unset. Raises ValueError if the variable is set to a non-integer string.

```APIDOC
## get_retries_number()

### Description
Reads the configured attempt budget from the environment variable DB_RETRY_RETRIES_NUMBER. Returns the value as an integer, or 3 if the variable is unset. Raises ValueError if the variable is set to a non-integer string.

### Method
Function call

### Endpoint
`db_retry.settings.get_retries_number`

### Parameters
None.

### Return
`int` — the value of the environment variable `DB_RETRY_RETRIES_NUMBER` parsed as an int, or `3` when the variable is unset. It represents the **total number of attempts** (first call plus retries) for a bare `@postgres_retry`.

### Throws
| Error type | Condition |
|-----------|-----------|
| `ValueError` | The variable is set to a value that is not a valid integer string (e.g. `DB_RETRY_RETRIES_NUMBER=abc`). Raised at the call site of `int(os.getenv(...))` — i.e. on the first invocation of a bare-decorated function. |

### Usage Example
```python
import os

from db_retry.settings import get_retries_number

os.environ.setdefault("DB_RETRY_RETRIES_NUMBER", "5")
assert get_retries_number() == 5

os.environ.pop("DB_RETRY_RETRIES_NUMBER", None)
assert get_retries_number() == 3  # default
```

### Error Handling Example
```python
import os

os.environ["DB_RETRY_RETRIES_NUMBER"] = "many"

try:
    get_retries_number()
except ValueError as exc:
    print("DB_RETRY_RETRIES_NUMBER is not an integer", exc)
```
```

--------------------------------

### ConnectionType TYPE_CHECKING-only alias

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/types.md

Private alias in db_retry.connections (lines 15-16). Defined only for type checkers; at runtime annotations referencing it are strings. Names asyncpg.Connection generic with unconstrained item type.

```python
if typing.TYPE_CHECKING:
    ConnectionType = asyncpg.Connection[typing.Any]
```

--------------------------------

### is_dsn_multihost

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/index.md

Checks if a DSN string indicates multiple hosts.

```APIDOC
## is_dsn_multihost

### Description
Checks if a DSN string indicates multiple hosts.

### Signature
`def is_dsn_multihost(db_dsn: str) -> bool`

### Parameters
- **db_dsn** (str) - Required - The DSN string.

### Returns
- bool - True if the DSN is multihost.
```

--------------------------------

### is_dsn_multihost

Source: https://github.com/modern-python/db-retry/blob/main/README.md

Checks if a DSN contains multiple hosts.

```APIDOC
## is_dsn_multihost

### Description
Checks if a DSN contains multiple hosts.

### Parameters
- `db_dsn` (str) - Required - The DSN to check.

### Returns
Boolean indicating whether the DSN contains multiple hosts.
```

--------------------------------

### async def __aexit__ - Transaction context manager exit

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/transaction.md

Exits the async context manager. Rolls back any open transaction (including on normal exit) and always closes the session. Returns None; exceptions from the block propagate unchanged.

```python
    async def __aexit__(self, exc_type: object, *args: object, **kwargs: object) -> None:
        if self.session.in_transaction():
            await self.session.rollback()
        await self.session.close()
```

--------------------------------

### is_dsn_multihost function definition

Source: https://github.com/modern-python/db-retry/blob/main/_autodocs/api-reference/dsn.md

Returns True when the DSN names multiple hosts, i.e. when the asyncpg dialect produced parallel host/port lists. Throws sqlalchemy.exc.ArgumentError for malformed multi-host DSNs.

```python
def is_dsn_multihost(db_dsn: str) -> bool:
    return bool(parse_connect_args(sa.make_url(db_dsn))[1])
```

--------------------------------

### Comment on a GitHub issue

Source: https://github.com/modern-python/db-retry/blob/main/docs/agents/issue-tracker.md

Use `gh issue comment <number> --body "..."` to add a comment to an issue.

```bash
gh issue comment <number> --body "..."
```
Tokens:
Raw