SQLAlchemy는 Python 생태계에서 가장 강력하고 널리 사용되는 ORM(Object-Relational Mapping) 및 SQL 툴킷입니다. 이 가이드에서는 SQLAlchemy의 핵심 아키텍처를 이해하고, 현대적인 2.0 스타일의 쿼리 문법을 사용하여 데이터베이스 스키마 정의, 복잡한 데이터 조작, 그리고 프로덕션 환경에서의 세션 관리 패턴까지 심층적으로 다룹니다.

환경 구성 및 데이터베이스 드라이버

SQLAlchemy 코어와 ORM 기능을 사용하기 위해 기본 패키지를 설치하며, 타겟 데이터베이스에 맞는 DBAPI 드라이버를 추가해야 합니다.

# SQLAlchemy 코어 및 ORM 설치
pip install sqlalchemy

# 타겟 RDBMS에 따른 드라이버 설치
# PostgreSQL
pip install psycopg2-binary

# MySQL / MariaDB
pip install pymysql

# SQLite는 Python 표준 라이브러리에 포함되어 있어 별도 설치가 불필요합니다.

엔진 초기화 및 세션 팩토리 구성

Engine은 데이터베이스 연결 풀과 다이얼렉트를 관리하는 핵심 객체이며, Session은 객체의 영속성과 트랜잭션을 추적하는 작업 공간입니다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 데이터베이스 연결 URL (SQLite 인메모리 예시)
DATABASE_URL = "sqlite+pysqlite:///:memory:"

# 엔진 생성 및 연결 풀 설정
db_engine = create_engine(
    DATABASE_URL, 
    echo=False, 
    pool_pre_ping=True,
    pool_size=10, 
    max_overflow=20
)

# 세션 팩토리 구성
SessionFactory = sessionmaker(
    bind=db_engine, 
    autocommit=False, 
    autoflush=False,
    expire_on_commit=False
)

데이터 모델 매핑 및 스키마 생성

declarative base를 사용하여 Python 클래스를 데이터베이스 테이블로 매핑합니다. 여기서는 Department, Employee, Project, Skill 엔티티를 정의하여 1:N 및 M:N 관계를 구현합니다.

from sqlalchemy import Column, Integer, String, Float, ForeignKey, Table
from sqlalchemy.orm import DeclarativeBase, relationship, Mapped, mapped_column
from typing import List

class Base(DeclarativeBase):
    pass

# M:N 관계를 위한 association table
project_skills_association = Table(
    "project_skills",
    Base.metadata,
    Column("project_id", Integer, ForeignKey("projects.id"), primary_key=True),
    Column("skill_id", Integer, ForeignKey("skills.id"), primary_key=True)
)

class Department(Base):
    __tablename__ = "departments"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    dept_name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    
    # 1:N 관계 (양방향)
    employees: Mapped[List["Employee"]] = relationship(back_populates="department")

class Employee(Base):
    __tablename__ = "employees"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    full_name: Mapped[str] = mapped_column(String(100), nullable=False)
    hire_date: Mapped[str] = mapped_column(String(20))
    dept_id: Mapped[int] = mapped_column(ForeignKey("departments.id"))
    
    department: Mapped["Department"] = relationship(back_populates="employees")
    led_projects: Mapped[List["Project"]] = relationship(back_populates="lead_dev")

class Project(Base):
    __tablename__ = "projects"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(150), nullable=False)
    budget: Mapped[float] = mapped_column(Float, default=0.0)
    lead_id: Mapped[int] = mapped_column(ForeignKey("employees.id"))
    
    lead_dev: Mapped["Employee"] = relationship(back_populates="led_projects")
    # M:N 관계
    required_skills: Mapped[List["Skill"]] = relationship(
        secondary=project_skills_association, 
        back_populates="associated_projects"
    )

class Skill(Base):
    __tablename__ = "skills"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    tech_name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    
    associated_projects: Mapped[List["Project"]] = relationship(
        secondary=project_skills_association, 
        back_populates="required_skills"
    )

# 스키마 생성
Base.metadata.create_all(bind=db_engine)

데이터 영속화 및 상태 전이 (CRUD)

세션을 통해 객체를 생성하고 데이터베이스에 반영합니다. SQLAlchemy 2.0 스타일의 session.add() 및 session.execute()를 활용합니다.

with SessionFactory() as session:
    # 부서 및 직원 생성
    engineering = Department(dept_name="Engineering")
    dev1 = Employee(full_name="Alice Smith", hire_date="2023-01-15", department=engineering)
    dev2 = Employee(full_name="Bob Jones", hire_date="2023-03-20", department=engineering)
    
    session.add_all([engineering, dev1, dev2])
    session.flush() # ID 할당을 위한 플러시
    
    # 프로젝트 및 스킬 할당
    python_skill = Skill(tech_name="Python")
    docker_skill = Skill(tech_name="Docker")
    
    alpha_project = Project(
        title="Alpha Migration", 
        budget=50000.0, 
        lead_dev=dev1,
        required_skills=[python_skill, docker_skill]
    )
    
    session.add(alpha_project)
    session.commit()

데이터 수정 및 삭제

from sqlalchemy import update, delete

with SessionFactory() as session:
    # 조건부 일괄 업데이트 (2.0 스타일)
    stmt = (
        update(Project)
        .where(Project.title == "Alpha Migration")
        .values(budget=65000.0)
    )
    session.execute(stmt)
    
    # 조건부 일괄 삭제
    delete_stmt = delete(Employee).where(Employee.full_name == "Bob Jones")
    session.execute(delete_stmt)
    
    session.commit()

고급 쿼리 빌딩 및 집계

select() 구문을 사용하여 복잡한 필터링, 조인, 그리고 집계 함수를 구성합니다.

from sqlalchemy import select, func, or_

with SessionFactory() as session:
    # 다중 조건 필터링 및 정렬
    stmt = (
        select(Employee)
        .where(
            or_(
                Employee.full_name.like("%Smith%"),
                Employee.hire_date >= "2023-01-01"
            )
        )
        .order_by(Employee.full_name.asc())
        .limit(10)
    )
    result = session.execute(stmt).scalars().all()
    
    # 조인 및 집계 (부서별 프로젝트 예산 합계)
    agg_stmt = (
        select(
            Department.dept_name, 
            func.count(Project.id).label("project_count"),
            func.sum(Project.budget).label("total_budget")
        )
        .outerjoin(Employee, Department.id == Employee.dept_id)
        .outerjoin(Project, Employee.id == Project.lead_id)
        .group_by(Department.dept_name)
    )
    
    for row in session.execute(agg_stmt):
        print(f"Dept: {row.dept_name}, Projects: {row.project_count}, Budget: {row.total_budget}")

트랜잭션 제어 및 중첩 저장점

복잡한 비즈니스 로직에서는 부분 실패를 처리하기 위해 중첩 트랜잭션(Nested Transaction)과 저장점(Savepoint)을 활용합니다.

with SessionFactory() as session:
    try:
        # 메인 트랜잭션 시작
        new_dept = Department(dept_name="Research")
        session.add(new_dept)
        
        # 중첩 트랜잭션 (Savepoint)
        with session.begin_nested():
            temp_emp = Employee(full_name="Temp User", hire_date="2024-01-01", department=new_dept)
            session.add(temp_emp)
            # 이 블록 내에서 예외가 발생하면 Savepoint만 롤백되고 메인 트랜잭션은 유지됨
            if temp_emp.full_name == "Temp User":
                raise ValueError("Invalid employee name")
                
    except ValueError as e:
        print(f"Nested transaction rolled back: {e}")
        
    # 메인 트랜잭션 커밋 (Research 부서는 저장됨)
    session.commit()

프로덕션 환경에서의 세션 관리 패턴

웹 프레임워크나 비동기 환경에서는 세션의 수명을 요청(Request) 주기와 일치시키는 것이 중요합니다. 의존성 주입과 컨텍스트 관리자를 결합한 패턴을 사용합니다.

from contextlib import contextmanager
from typing import Generator

@contextmanager
def get_db_session() -> Generator:
    """요청 범위의 데이터베이스 세션을 관리하는 컨텍스트 매니저"""
    db_session = SessionFactory()
    try:
        yield db_session
        db_session.commit()
    except Exception:
        db_session.rollback()
        raise
    finally:
        db_session.close()

# 사용 예시
def assign_skill_to_project(project_id: int, skill_name: str):
    with get_db_session() as db:
        project = db.get(Project, project_id)
        if not project:
            raise ValueError("Project not found")
            
        skill = db.execute(
            select(Skill).where(Skill.tech_name == skill_name)
        ).scalar_one_or_none()
        
        if not skill:
            skill = Skill(tech_name=skill_name)
            db.add(skill)
            
        if skill not in project.required_skills:
            project.required_skills.append(skill)

assign_skill_to_project(1, "Kubernetes")