### Install Dependencies

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

Installs and synchronizes project dependencies using uv.

```bash
just install       # uv lock --upgrade && uv sync
```

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

### Install db-retry with uv

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

Install the db-retry library using the uv package manager.

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

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

### Install db-retry with pip

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

Install the db-retry library using pip.

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

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

### Full Pipeline Execution

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

Executes the full pipeline including installation, linting, building, and testing.

```bash
just               # install lint build test (full pipeline)
```

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

### CI DB_DSN Format

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

Example format for the DB_DSN connection string used in CI environments.

```bash
postgresql+asyncpg://postgres:postgres@localhost:5432/postgres
```

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

### Serializable Transactions for Consistency

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

Employ serializable isolation level to prevent race conditions in your database operations. This example demonstrates setting up a transaction with this isolation level.

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

```

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

### Build Docker Image

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

Builds the Docker image for the project.

```bash
just build         # build docker image
```

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

### High Availability Database Connections

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

Set up resilient database connections with multiple fallback hosts using a connection factory. The engine will automatically try different hosts on failure. A connection timeout can also be configured.

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

# Build production-ready DSN
ds = 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=ds,
    timeout=5.0  # 5 second connection timeout
)

# Engine will automatically try different hosts on failure
engine = create_async_engine(ds, async_creator=connection_factory)
```

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

### Format Code

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

Formats code automatically using ruff.

```bash
just lint          # ruff format + eof-fixer (auto-fix)
```

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

### Simplified Transaction Management with ORM

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

Use this snippet for handling database transactions with automatic cleanup. It requires importing Transaction, postgres_retry, and other necessary modules.

```python
import dataclasses
import datetime
import typing

from schemas import AnalyticsEventCreate, AnalyticsEvent
from db_retry import Transaction, postgres_retry

from your_service_name.database.tables import EventsTable
from your_service_name.producers.analytics_service_events_producer import AnalyticsEventsProducer
from your_service_name.repositories.events_repository import EventsRepository
from your_service_name.settings import settings


@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
class CreateEventUseCase:
    events_repository: EventsRepository
    transaction: Transaction
    analytics_events_producer: AnalyticsEventsProducer

    @postgres_retry
    async def __call__(
            self,
            event_create_data: AnalyticsEventCreate,
    ) -> AnalyticsEvent:
        async with self.transaction:
            model: typing.Final = EventsTable(
                **event_create_data.model_dump(),
                created_at=datetime.datetime.now(tz=settings.common.default_timezone),
            )
            saved_event: typing.Final[EventsTable] = await self.events_repository.create(model)
            event: typing.Final = AnalyticsEvent.model_validate(saved_event)
            await self.analytics_events_producer.send_message(event)
            await self.transaction.commit()
            return event

```

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

### Run Tests

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

Runs pytest within a Docker container. Requires a PostgreSQL instance.

```bash
just test          # run pytest inside docker (requires postgres)
```

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

### Run Pytest Locally

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

Executes pytest locally. Ensure DB_DSN environment variable is set.

```bash
uv run pytest
```

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

### Run Single Pytest Locally

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

Executes a specific pytest test function locally. Ensure DB_DSN environment variable is set.

```bash
uv run pytest tests/test_retry.py::test_postgres_retry  # single test
```

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

### Check Code for Linting Errors

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

Checks code for linting errors without applying fixes.

```bash
just lint-ci       # ruff check only (no fixes)
```

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

### Configure DB Retry Retries Number

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

Set the number of retry attempts for database operations using the DB_RETRY_RETRIES_NUMBER environment variable. The default is 3.

```bash
export DB_RETRY_RETRIES_NUMBER=5

```

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

### ORM Operations with Automatic Retry

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

Apply retry logic to ORM operations using the @postgres_retry decorator. This automatically retries on connection failures or serialization errors. The default retry count is 3, but can be overridden.

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


class User(DeclarativeBase):
    __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)


# Apply retry logic to ORM operations (uses DB_RETRY_RETRIES_NUMBER, default 3)
@postgres_retry
async def get_user_by_email(session: AsyncSession, email: str) -> User:
    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 or serialization errors
        user = await get_user_by_email(session, "john.doe@example.com")
        if user:
            print(f"Found user: {user.name}")


asyncio.run(main())
```

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

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.