sqlalchemy-orm

name sqlalchemy-orm description SQLAlchemy Python SQL toolkit and ORM with powerful query builder, relationship mapping, and database migrations via Alembic user-invocable false disable-model-invocation true progressive_disclosure {"entry_point":{"summary":"SQLAlchemy Python SQL toolkit and ORM with powerful query builder, relationship mapping, and database migrations via Alembic","when_to_use":"When working with sqlalchemy-orm or related functionality.","quick_start":"1. Review the core concepts below. 2. Apply patterns to your use case. 3. Follow best practices for implementation."}} SQLAlchemy ORM Skill progressive_disclosure: entry_point: summary: "Python SQL toolkit and ORM with powerful query builder and relationship mapping" when_to_use: - "When building Python applications with databases" - "When needing complex SQL queries with type safety" - "When working with FastAPI/Flask/Django" - "When needing database migrations (Alembic)" quick_start: - "pip install sqlalchemy" - "Define models with declarative base" - "Create engine and session" - "Query with select() and commit()" token_estimate: entry: 70-85 full: 4500-5500 Core Concepts SQLAlchemy 2.0 Modern API SQLAlchemy 2.0 introduced modern patterns with better type hints, improved query syntax, and async support. Key Changes from 1.x: select() instead of Query Mapped[T] and mapped_column() for type hints Explicit Session.execute() for queries Better async support with AsyncSession Installation # Core SQLAlchemy pip install sqlalchemy # With async support pip install sqlalchemy[asyncio] aiosqlite # SQLite pip install sqlalchemy[asyncio] asyncpg # PostgreSQL # With Alembic for migrations pip install alembic # FastAPI integration pip install fastapi sqlalchemy Declarative Models (SQLAlchemy 2.0) Basic Model Definition from datetime import datetime from typing import Optional from sqlalchemy import String, DateTime, ForeignKey, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship # Base class for all models class Base ( DeclarativeBase ): pass # User model with type hints class User ( Base ): __tablename__ = "users" # Primary key id : Mapped[ int ] = mapped_column(primary_key= True ) # Required fields email: Mapped[ str ] = mapped_column(String( 255 ), unique= True , index= True ) username: Mapped[ str ] = mapped_column(String( 50 ), unique= True ) hashed_password: Mapped[ str ] = mapped_column(String( 255 )) # Optional fields full_name: Mapped[ Optional [ str ]] = mapped_column(String( 100 )) is_active: Mapped[ bool ] = mapped_column(default= True ) # Timestamps with server defaults created_at: Mapped[datetime] = mapped_column( DateTime(timezone= True ), server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone= True ), server_default=func.now(), onupdate=func.now() ) # Relationships posts: Mapped[ list [ "Post" ]] = relationship(back_populates= "author" ) def __repr__ ( self ) -> str : return f"User(id= {self. id } , email= {self.email} )" Relationships One-to-Many: class Post ( Base ): __tablename__ = "posts" id : Mapped[ int ] = mapped_column(primary_key= True ) title: Mapped[ str ] = mapped_column(String( 200 )) content: Mapped[ str ] user_id: Mapped[ int ] = mapped_column(ForeignKey( "users.id" )) # Relationship with back_populates author: Mapped[ "User" ] = relationship(back_populates= "posts" ) tags: Mapped[ list [ "Tag" ]] = relationship( secondary= "post_tags" , back_populates= "posts" ) Many-to-Many: from sqlalchemy import Table, Column, Integer, ForeignKey # Association table post_tags = Table( "post_tags" , Base.metadata, Column( "post_id" , Integer, ForeignKey( "posts.id" ), primary_key= True ), Column( "tag_id" , Integer, ForeignKey( "tags.id" ), primary_key= True ) ) class Tag ( Base ): __tablename__ = "tags" id : Mapped[ int ] = mapped_column(primary_key= True ) name: Mapped[ str ] = mapped_column(String( 50 ), unique= True ) posts: Mapped[ list [ "Post" ]] = relationship( secondary=post_tags, back_populates= "tags" ) Database Setup Engine and Session Configuration from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from sqlalchemy.pool import QueuePool # Database URL formats # SQLite: sqlite:///./database.db # PostgreSQL: postgresql://user:pass@localhost/dbname # MySQL: mysql+pymysql://user:pass@localhost/dbname DATABASE_URL = "postgresql://user:pass@localhost/mydb" # Create engine with connection pooling engine = create_engine( DATABASE_URL, poolclass=QueuePool, pool_size= 5 , max_overflow= 10 , pool_pre_ping= True , # Check connection before using echo= False # Set True for SQL logging ) # Session factory SessionLocal = sessionmaker( bind=engine, autocommit= False , autoflush= False , expire_on_commit= False ) # Create tables Base.metadata.create_all(bind=engine) Dependency Injection (FastAPI Pattern) from typing import Generator def get_db () -> Generator[Session, None , None ]: """Database session dependency for FastAPI.""" db = SessionLocal() try : yield db finally : db.close() # Usage in FastAPI from fastapi import Depends @app.get( "/users/{user_id}" ) def get_user ( user_id: int , db: Session = Depends( get_db ) ): return db.execute( select(User).where(User. id == user_id) ).scalar_one_or_none() Query Patterns (SQLAlchemy 2.0) Select Queries from sqlalchemy import select, and_, or_, desc, func # Basic select stmt = select(User).where(User.email == "user@example.com" ) user = session.execute(stmt).scalar_one_or_none() # Multiple conditions stmt = select(User).where( and_( User.is_active == True , User.created_at > datetime( 2024 , 1 , 1 ) ) ) users = session.execute(stmt).scalars(). all () # OR conditions stmt = select(User).where( or_( User.email.like( "%@gmail.com" ), User.email.like( "%@yahoo.com" ) ) ) # Ordering and limiting stmt = ( select(User) .where(User.is_active == True ) .order_by(desc(User.created_at)) .limit( 10 ) .offset( 20 ) ) # Counting stmt = select(func.count()).select_from(User) count = session.execute(stmt).scalar() Joins # Inner join stmt = ( select(Post, User) .join(User, Post.user_id == User. id ) .where(User.is_active == True ) ) results = session.execute(stmt). all () # Left outer join stmt = ( select(User, func.count(Post. id ).label( "post_count" )) .outerjoin(Post) .group_by(User. id ) ) # Multiple joins stmt = ( select(Post) .join(Post.author) .join(Post.tags) .where(Tag.name == "python" ) ) Eager Loading (Solve N+1 Problem) from sqlalchemy.orm import selectinload, joinedload # selectinload - separate query (better for collections) stmt = select(User).options(selectinload(User.posts)) users = session.execute(stmt).scalars(). all () # Now users[0].posts won't trigger additional queries # joinedload - single query with join (better for one-to-one) stmt = select(Post).options(joinedload(Post.author)) posts = session.execute(stmt).unique().scalars(). all () # Nested eager loading stmt = select(User).options( selectinload(User.posts).selectinload(Post.tags) ) # Load only specific columns from sqlalchemy.orm import load_only stmt = select(User).options(load_only(User. id , User.email)) CRUD Operations Create def create_user ( db: Session, email: str , username: str , password: str ): """Create new user.""" user = User( email=email, username=username, hashed_password=hash_password(password) ) db.add(user) db.commit() db.refresh(user) # Get updated fields (id, timestamps) return user # Bulk insert users = [ User(email= f"user {i} @example.com" , username= f"user {i} " ) for i in range ( 100 ) ] db.add_all(users) db.commit() Read def get_user_by_email ( db: Session, email: str ) -> Optional [User]: """Get user by email.""" stmt = select(User).where(User.email == email) return db.execute(stmt).scalar_one_or_none() def get_users ( db: Session, skip: int = 0 , limit: int = 100 ) -> list [User]: """Get paginated users.""" stmt = ( select(User) .where(User.is_active == True ) .order_by(User.created_at.desc()) .offset(skip) .limit(limit) ) return db.execute(stmt).scalars(). all () Update def update_user ( db: Session, user_id: int , **kwargs ): """Update user fields.""" stmt = select(User).where(User. id == user_id) user = db.execute(stmt).scalar_one_or_none() if not user: return None for key, value in kwargs.items(): setattr (user, key, value) db.commit() db.refresh(user) return user # Bulk update from sqlalchemy import update stmt = ( update(User) .where(User.is_active == False ) .values(deleted_at=datetime.utcnow()) ) db.execute(stmt) db.commit() Delete def delete_user ( db: Session, user_id: int ) -> bool : """Delete user.""" stmt = select(User).where(User. id == user_id) user = db.execute(stmt).scalar_one_or_none() if not user: return False db.delete(user) db.commit() return True # Bulk delete from sqlalchemy import delete stmt = delete(User).where(User.is_active == False ) db.execute(stmt) db.commit() Transactions and Session Management Context Manager Pattern from contextlib import contextmanager @contextmanager def get_db_session (): """Session context manager.""" session = SessionLocal() try : yield session session.commit() except Exception: session.rollback() raise finally : session.close() # Usage with get_db_session() as db: user = create_user(db, "test@example.com" , "testuser" , "password" ) # Auto-commits on success, rollback on exception Manual Transaction Control def transfer_money ( db: Session, from_user_id: int , to_user_id: int , amount: float ): """Transfer money between users with transaction.""" try : # Begin nested transaction with db.begin_nested(): # Deduct from sender stmt = select(User).where(User. id == from_user_id).with_for_update() sender = db.execute(stmt).scalar_one() sender.balance -= amount # Add to receiver stmt = select(User).where(User. id == to_user_id).with_for_update() receiver = db.execute(stmt).scalar_one() receiver.balance += amount db.commit() except Exception as e: db.rollback() raise Async SQLAlchemy Async Setup from sqlalchemy.ext.asyncio import ( create_async_engine, AsyncSession, async_sessionmaker ) # Async engine (note: asyncpg for PostgreSQL, aiosqlite for SQLite) DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"