Recap — Mapped Columns, Sessions, Query Patterns

# Mapped Columns, Sessions, Query Patterns SQLAlchemy 2.0 (released 2023) modernized the API. If you see old tutorials with `Column(Integer, primary_key=True)` style, those are 1.x. New code uses Mapped + mapped_column. ## Modern declarative model ```python from datetime import datetime from sqlalchemy import String, ForeignKey from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) email: Mapped[str] = mapped_column(String(255), unique=True, index=True) name: Mapped[str] = mapped_column(String(100)) is_active: Mapped[bool] = mapped_column(default=True) created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) posts: Mapped[list["Post"]] = relationship(back_populates="author") class Post(Base): __tablename__ = "posts" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(200)) content: Mapped[str] author_id: Mapped[int] = mapped_column(ForeignKey("users.id")) author: Mapped["User"] = relationship(back_populates="posts") ``` Mapped types ARE type hints. Pyright/mypy understand them. IDE autocomplete works. ## Engine and session ```python from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session engine = create_engine("postgresql://user:pass@localhost/dbname") SessionLocal = sessionmaker(bind=engine, autoflush=False) # FastAPI dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() ``` For SQLite (dev/test): ```python engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False}) ``` ## CRUD operations ### Create ```python new_user = User(email="alice@example.com", name="Alice") db.add(new_user) db.commit() db.refresh(new_user) # populates id and other server-generated fields ``` ### Read ```python from sqlalchemy import select # By primary key user = db.get(User, 123) # Query with filter stmt = select(User).where(User.email == "alice@example.com") user = db.execute(stmt).scalar_one_or_none() # Multiple results stmt = select(User).where(User.is_active == True).order_by(User.created_at.desc()) users = db.execute(stmt).scalars().all() ``` Note: `select()` not `db.query(...)`. The old `db.query()` syntax still works but `select()` is the 2.0 style. ### Update ```python user = db.get(User, 123) user.name = "Alice Updated" db.commit() ``` Or bulk update: ```python from sqlalchemy import update stmt = update(User).where(User.is_active == False).values(deleted=True) db.execute(stmt) db.commit() ``` ### Delete ```python user = db.get(User, 123) db.delete(user) db.commit() ``` ## Joins ### Basic join ```python stmt = ( select(Post, User) .join(User, Post.author_id == User.id) .where(User.email == "alice@example.com") ) results = db.execute(stmt).all() for post, user in results: print(post.title, user.name) ``` ### Eager loading (avoid N+1) Without eager loading: querying posts then accessing `.author` triggers a query per post (N+1 problem). ```python from sqlalchemy.orm import joinedload, selectinload # joinedload: SQL JOIN; one query stmt = select(Post).options(joinedload(Post.author)) # selectinload: separate SELECT IN; two queries (often more efficient) stmt = select(Post).options(selectinload(Post.author)) ``` For lists: `selectinload` is usually preferred (avoids huge joined rows). ## Filtering patterns ```python # Equality where(User.is_active == True) # Comparison where(User.created_at > datetime(2026, 1, 1)) # In list where(User.role.in_(["admin", "moderator"])) # Like / pattern where(User.email.like("%@example.com")) # Multiple conditions (AND) where(User.is_active == True, User.role == "admin") # OR from sqlalchemy import or_ where(or_(User.role == "admin", User.role == "moderator")) # Not null where(User.email.isnot(None)) ``` ## Pagination ```python def list_users(db: Session, skip: int = 0, limit: int = 100): stmt = select(User).order_by(User.id).offset(skip).limit(limit) return db.execute(stmt).scalars().all() ``` For offset pagination. For cursor-based (better at scale): ```python def list_users_cursor(db: Session, after_id: int = 0, limit: int = 100): stmt = select(User).where(User.id > after_id).order_by(User.id).limit(limit) return db.execute(stmt).scalars().all() ``` ## Transactions ```python # Default: autocommit off; transaction spans your operations db.add(user) db.add(post) db.commit() # both succeed or fail together # Explicit transaction (context manager) with db.begin(): db.add(user) db.add(post) # Auto-commits on exit; rolls back on exception ``` For nested needs: ```python with db.begin_nested(): # SAVEPOINT db.add(maybe_failing_thing) # Commit/rollback this savepoint ``` ## Indexes ```python class User(Base): email: Mapped[st