Python Backend

# Database Patterns

SQLAlchemy and Alembic patterns for Python backends.

## Async Engine + Session

Use async engine + async session per request:

```python
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/app"

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session
```

---

## Commit/Rollback Pattern

Wrap write operations in try/except:

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

async def create_user(session: AsyncSession, user: dict) -> User:
    try:
        db_user = User(**user)
        session.add(db_user)
        await session.commit()
        await session.refresh(db_user)
        return db_user
    except Exception:
        await session.rollback()
        raise
```

### Nested Transaction with Savepoint

```python
async def transfer_funds(session: AsyncSession, from_id: int, to_id: int, amount: float):
    async with session.begin_nested():  # Creates a savepoint
        from_account = await session.get(Account, from_id)
        to_account = await session.get(Account, to_id)
        
        if from_account.balance < amount:
            raise InsufficientFunds()
        
        from_account.balance -= amount
        to_account.balance += amount
    
    await session.commit()
```

---

## Naming Conventions

Be consistent with names:
1. lower_case_snake
2. singular form (e.g. post, post_like, user_playlist)
3. group similar tables with module prefix (e.g. payment_account, payment_bill)
4. stay consistent across tables
5. _at suffix for datetime, _date suffix for date

### Constraint Naming

```python
from sqlalchemy import MetaData

NAMING_CONVENTION = {
    "ix": "%(column_0_label)s_idx",
    "uq": "%(table_name)s_%(column_0_name)s_key",
    "ck": "%(table_name)s_%(constraint_name)s_check",
    "fk": "%(table_name)s_%(column_0_name)s_fkey",
    "pk": "%(table_name)s_pkey",
}

metadata = MetaData(naming_convention=NAMING_CONVENTION)
```

---

## Alembic Migration Naming

Use human-readable file template:

```ini
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s
# Results in: 2024-08-24_add_users_table.py
```

---

## Eager Loading - Avoid N+1

### selectinload for collections

```python
from sqlalchemy import select
from sqlalchemy.orm import selectinload

# Load all users with their orders in 2 queries
stmt = select(User).options(selectinload(User.orders))
users = await session.scalars(stmt)

for user in users:
    print(user.orders)  # No additional query
```

### joinedload for single relationships

```python
from sqlalchemy.orm import joinedload

# Load orders with their user in 1 query (JOIN)
stmt = select(Order).options(joinedload(Order.user))
orders = await session.scalars(stmt)
```

### raiseload to detect N+1

```python
from sqlalchemy.orm import raiseload

stmt = select(User).options(
    selectinload(User.orders),
    raiseload("*")  # Raise on any other lazy load
)
```

---

## Cascade Delete

Configure at both ORM and database level:

```python
from sqlalchemy import ForeignKey, Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Parent(Base):
    __tablename__ = "parent"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    
    children: Mapped[list["Child"]] = relationship(
        back_populates="parent",
        cascade="all, delete",
        passive_deletes=True,
    )

class Child(Base):
    __tablename__ = "child"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    parent_id: Mapped[int] = mapped_column(
        ForeignKey("parent.id", ondelete="CASCADE")
    )
    
    parent: Mapped["Parent"] = relationship(back_populates="children")
```