SQLAlchemy 2.0

SQLAlchemy는 Python의 가장 강력한 ORM 라이브러리입니다. 2.0 버전부터 선언형 매핑과 async 지원이 크게 개선되었습니다.

---

설치와 기본 설정

pip install sqlalchemy psycopg2-binary  # PostgreSQL
# 또는
pip install sqlalchemy aiosqlite         # SQLite (비동기)
pip install sqlalchemy asyncpg           # PostgreSQL (비동기)

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# 동기 엔진
engine = create_engine(
    "postgresql://user:password@localhost/dbname",
    pool_size=10,
    max_overflow=20,
    echo=False,  # SQL 로그 출력 여부
)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)


class Base(DeclarativeBase):
    pass

---

모델 선언형 매핑

from sqlalchemy import String, Integer, ForeignKey, Text, Boolean, DateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from datetime import datetime


class Category(Base):
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    slug: Mapped[str] = mapped_column(String(100), unique=True)

    products: Mapped[list["Product"]] = relationship(back_populates="category")

    def __repr__(self):
        return f"<Category(id={self.id}, name={self.name})>"


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    price: Mapped[float] = mapped_column(nullable=False)
    stock: Mapped[int] = mapped_column(default=0)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

    category_id: Mapped[int | None] = mapped_column(ForeignKey("categories.id"))
    category: Mapped["Category | None"] = relationship(back_populates="products")

    def __repr__(self):
        return f"<Product(id={self.id}, name={self.name})>"

---

테이블 생성 및 세션 관리

# 테이블 생성
Base.metadata.create_all(bind=engine)

# 세션 컨텍스트 매니저 (FastAPI / 일반 앱 공통)
from contextlib import contextmanager


@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

---

CRUD 패턴

from sqlalchemy.orm import Session
from sqlalchemy import select, update, delete


# CREATE
def create_product(db: Session, name: str, price: float, category_id: int) -> Product:
    product = Product(name=name, price=price, category_id=category_id)
    db.add(product)
    db.flush()      # DB에 반영 (commit 전)
    db.refresh(product)  # auto-generated 값(id, created_at) 가져오기
    return product


# READ — select() 스타일 (2.0 권장)
def get_product(db: Session, product_id: int) -> Product | None:
    stmt = select(Product).where(Product.id == product_id)
    return db.scalar(stmt)


def get_products(
    db: Session,
    skip: int = 0,
    limit: int = 20,
    is_active: bool = True,
) -> list[Product]:
    stmt = (
        select(Product)
        .where(Product.is_active == is_active)
        .order_by(Product.created_at.desc())
        .offset(skip)
        .limit(limit)
    )
    return list(db.scalars(stmt))


# 관계 데이터 함께 조회 (N+1 방지)
from sqlalchemy.orm import selectinload, joinedload


def get_products_with_category(db: Session) -> list[Product]:
    stmt = (
        select(Product)
        .options(joinedload(Product.category))  # JOIN으로 한 번에
        .where(Product.is_active == True)
    )
    return list(db.scalars(stmt).unique())


# UPDATE
def update_product(db: Session, product_id: int, **kwargs) -> Product | None:
    product = get_product(db, product_id)
    if not product:
        return None
    for key, value in kwargs.items():
        setattr(product, key, value)
    db.flush()
    db.refresh(product)
    return product


# 벌크 UPDATE
def deactivate_products(db: Session, category_id: int) -> int:
    stmt = (
        update(Product)
        .where(Product.category_id == category_id)
        .values(is_active=False)
        .execution_options(synchronize_session="fetch")
    )
    result = db.execute(stmt)
    return result.rowcount


# DELETE
def delete_product(db: Session, product_id: int) -> bool:
    product = get_product(db, product_id)
    if not product:
        return False
    db.delete(product)
    return True

---

집계와 서브쿼리

from sqlalchemy import func, and_, or_


# 집계
def get_category_stats(db: Session) -> list[dict]:
    stmt = (
        select(
            Category.name,
            func.count(Product.id).label("product_count"),
            func.avg(Product.price).label("avg_price"),
            func.sum(Product.stock).label("total_stock"),
        )
        .join(Product, isouter=True)
        .group_by(Category.id)
        .having(func.count(Product.id) > 0)
        .order_by(func.count(Product.id).desc())
    )
    rows = db.execute(stmt).all()
    return [row._asdict() for row in rows]


# 서브쿼리
def get_expensive_products(db: Session) -> list[Product]:
    avg_price_subq = select(func.avg(Product.price)).scalar_subquery()
    stmt = select(Product).where(Product.price > avg_price_subq)
    return list(db.scalars(stmt))

---

비동기 SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

# 비동기 엔진
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine, expire_on_commit=False
)


# FastAPI 의존성
async def get_async_db():
    async with AsyncSessionLocal() as db:
        try:
            yield db
            await db.commit()
        except Exception:
            await db.rollback()
            raise


# 비동기 쿼리
async def get_product_async(db: AsyncSession, product_id: int) -> Product | None:
    stmt = (
        select(Product)
        .options(selectinload(Product.category))  # 비동기에선 selectinload 권장
        .where(Product.id == product_id)
    )
    return await db.scalar(stmt)

---

정리

기능	방법
모델 정의	Mapped[type] = mapped_column(...)
단일 조회	db.scalar(select(Model).where(...))
복수 조회	db.scalars(stmt)
관계 즉시 로딩	joinedload() / selectinload()
벌크 업데이트	update(Model).where(...).values(...)
집계	func.count(), func.avg(), group_by()
비동기	AsyncSession + create_async_engine

SQLAlchemy 2.0의 select() 스타일이 타입 안전성과 IDE 자동 완성이 우수하여 표준으로 권장됩니다.