sqlalchemy

name	sqlalchemy
description	[Applies to: **/*.py] This guide enforces modern SQLAlchemy 2.x best practices for Python applications, ensuring type-safe, performant, and maintainable database interactions.
source	cursor_mdc

SQLAlchemy Best Practices (2.x Style)

This document outlines the definitive best practices for using SQLAlchemy 2.x in our projects. We exclusively adopt the "2-style" API, leveraging its fully typed, declarative features for robust and maintainable code.

1. Code Organization and Data Modeling

Always define your ORM models using the modern DeclarativeBase and Mapped annotations. Keep model definitions in a dedicated models.py or orm_models/ directory.

1.1 Declarative Models with Type Annotations

Use DeclarativeBase as your base class and Mapped for all ORM-mapped attributes. This enables static type checking and clear schema definition.

❌ BAD: Legacy declarative_base() function and untyped Column.

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String) # No type hint

✅ GOOD: Modern DeclarativeBase and Mapped for explicit typing.

from typing import List, Optional
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey, func, DateTime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    created_at: Mapped[DateTime] = mapped_column(DateTime, insert_default=func.now())

    addresses: Mapped[List["Address"]] = relationship(back_populates="user", cascade="all, delete-orphan")

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))

    user: Mapped["User"] = relationship(back_populates="addresses")

1.2 Mixins for Common Fields

Implement common fields like timestamps, soft-delete flags, or version counters using mixins.

from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import DateTime, func

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), onupdate=func.now())

class SoftDeleteMixin:
    deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime, default=None)

class User(Base, TimestampMixin, SoftDeleteMixin):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    # ... other fields

2. Session Management

The Session is your unit of work. Always use it as a context manager to ensure proper transaction handling and resource cleanup.

2.1 Context Manager for Sessions

Wrap all database operations within a with Session(engine) as session: block. This guarantees commit() on success and rollback() on error, followed by close().

❌ BAD: Manual session handling, prone to leaks and uncommitted transactions.

session = Session(engine)
try:
    user = User(name="Alice")
    session.add(user)
    session.commit()
except Exception:
    session.rollback()
finally:
    session.close()

✅ GOOD: Session as a context manager.

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

engine = create_engine("sqlite:///./test.db")
Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def create_user(name: str):
    with Session() as session:
        user = User(name=name)
        session.add(user)
        session.commit() # Flushes changes and commits transaction
        session.refresh(user) # Refresh to get auto-generated IDs
        return user

2.2 Explicit Transaction Blocks

For complex operations requiring multiple steps within a single transaction, use session.begin().

def transfer_funds(from_account_id: int, to_account_id: int, amount: float):
    with Session() as session:
        with session.begin(): # Explicit transaction block
            from_account = session.get(Account, from_account_id)
            to_account = session.get(Account, to_account_id)

            if not from_account or not to_account:
                raise ValueError("Account not found")
            if from_account.balance < amount:
                raise ValueError("Insufficient funds")

            from_account.balance -= amount
            to_account.balance += amount
        # Transaction commits here if no exceptions, or rolls back if an error occurs
    print(f"Transferred {amount} from {from_account_id} to {to_account_id}")

3. Querying and Optimization

Always use the select() construct for queries and prioritize eager loading for relationships to avoid N+1 problems.

3.1 Use select() for All Queries

The Query object is deprecated in 2.x. Use select() for all ORM and Core queries.

❌ BAD: Using the legacy session.query() API.

users = session.query(User).filter(User.name == "Alice").all()

✅ GOOD: Using select() with session.execute().

from sqlalchemy import select

def get_user_by_name(session: Session, name: str) -> Optional[User]:
    stmt = select(User).where(User.name == name)
    return session.scalar(stmt) # Use scalar for single result, one_or_none for one, all for list

3.2 Eager Loading Relationships

Prevent N+1 query issues by eagerly loading related objects using selectinload or joinedload. selectinload is generally preferred for collections.

❌ BAD: Lazy loading in a loop, leading to N+1 queries.

users = session.scalars(select(User)).all()
for user in users:
    print(f"{user.name} has {len(user.addresses)} addresses") # Each access triggers a new query

✅ GOOD: Eager loading with selectinload.

from sqlalchemy.orm import selectinload

def get_users_with_addresses(session: Session) -> List[User]:
    stmt = select(User).options(selectinload(User.addresses))
    return session.scalars(stmt).all()

# Now, accessing user.addresses won't trigger additional queries
users = get_users_with_addresses(session)
for user in users:
    print(f"{user.name} has {len(user.addresses)} addresses")

4. Data Manipulation

The ORM Unit of Work pattern handles inserts, updates, and deletes efficiently.

4.1 Adding and Updating Objects

Add new objects with session.add(). Modifications to existing objects are tracked automatically.

def update_user_name(session: Session, user_id: int, new_name: str):
    user = session.get(User, user_id) # Efficiently get by primary key
    if user:
        user.name = new_name # Change is tracked
        session.commit()
    return user

def add_new_user_with_address(session: Session, name: str, email: str):
    user = User(name=name, fullname=name)
    address = Address(email_address=email, user=user) # Relationship automatically links
    session.add(user) # Adding user also adds address due to cascade
    session.commit()
    return user

4.2 Deleting Objects

Delete objects using session.delete().

def delete_user(session: Session, user_id: int):
    user = session.get(User, user_id)
    if user:
        session.delete(user)
        session.commit()
        print(f"User {user_id} deleted.")
    else:
        print(f"User {user_id} not found.")

4.3 Bulk Operations (Use Sparingly)

Only use bulk operations (session.bulk_insert_mappings, session.execute(insert/update/delete)) when performance is paramount and ORM event handling (e.g., onupdate hooks) is not required.

from sqlalchemy import insert

def bulk_insert_users(session: Session, user_data: List[dict]):
    # This bypasses ORM object creation and event hooks.
    # Use when you have raw data and need speed.
    session.execute(insert(User), user_data)
    session.commit()

# Example user_data: [{"name": "Charlie", "fullname": "Charlie Brown"}, ...]

5. Migrations and Type Checking

Integrate Alembic for schema migrations and ensure static type checking with Mypy or Ruff.

5.1 Alembic for Schema Evolution

Always use Alembic to manage database schema changes. Never manually alter the database schema.

# Initialize Alembic (once per project)
alembic init -t async migrations

# Generate a new migration script
alembic revision --autogenerate -m "Add new_column to user_account"

# Apply migrations
alembic upgrade head

5.2 Static Type Checking

Leverage Python's typing and SQLAlchemy's Mapped for comprehensive type checking with tools like Mypy or Ruff. This catches schema mismatches and API misuses early.

# mypy.ini or pyproject.toml configuration for mypy
[mypy]
plugins = sqlalchemy.ext.mypy.plugin