도입 및 환경 준비

데이터베이스 연동 시 객체지향 프로그래밍의 유연성을 살리려면 ORM 이 필수적입니다. 파이썬 생태계에서 SQLAlchemy 는 가장 널리 쓰이는 라이브러리 중 하나로, 추상화된 인터페이스를 통해 다양한 DB 시스템을 일관되게 다룰 수 있게 해줍니다. 아래에서는 실제 프로젝트에 적용 가능한 설정부터 고급 쿼리 패턴까지 단계별로 살펴봅니다.

패키지 설치

기본 패키지뿐만 아니라 사용하려는 데이터베이스 드라이버도 함께 설치해야 연결이 가능합니다.

pip install sqlalchemy[asyncio]

# 특정 DB 드라이버 예시
pip install asyncpg         # PostgreSQL 비동기용
pip install mysqlclient     # MySQL C-bind용
# SQLite 는 내장되어 별도 설치가 불필요합니다

연결 구조 및 세션 정의

DB 엔진을 생성한 후, 이를 바탕으로 트랜잭션 단위를 관리할 세션 클래스를 정의하는 것이 표준적인 접근법입니다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 엔진 초기화 (SQLite 메모리 DB 예시)
db_engine = create_engine(
    'sqlite+aiosqlite:///:memory:', 
    echo=False, 
    future=True
)

# 세션 팩토리 생성
SessionLocalFactory = sessionmaker(bind=db_engine, expire_on_commit=False)
Base = declarative_base()

데이터 모델링 설계

테이블 구조는 파이썬 클래스로 매핑됩니다. 여기서는 사용자 계정을 중심으로 게시글과 태그 간의 관계를 정의해보겠습니다. 기존 코드와 변수명을 변경하여 가독성을 높였습니다.

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship

# 다대다 관계를 위한 중간 테이블 명시 정의
association_table = Table(
    'label_association', Base.metadata,
    Column('document_id', ForeignKey('documents.id')),
    Column('tag_id', ForeignKey('labels.id'))
)

class Account(Base):
    __tablename__ = 'accounts'
    
    uid = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False, unique=True)
    profile_email = Column(String(100))
    
    # 1:N 관계 설정
    writings = relationship("Document", back_populates="owner")

class Document(Base):
    __tablename__ = 'documents'
    
    doc_id = Column(Integer, primary_key=True)
    heading = Column(String(200), index=True)
    body_text = Column(String)
    writer_uid = Column(Integer, ForeignKey('accounts.uid'))
    
    owner = relationship("Account", back_populates="writings")
    categories = relationship(
        "Label", 
        secondary=association_table, 
        back_populates="attached_docs"
    )

class Label(Base):
    __tablename__ = 'labels'
    
    label_id = Column(Integer, primary_key=True)
    keyword = Column(String(30), unique=True)
    
    attached_docs = relationship("Document", secondary=association_table, back_populates="categories")

스키마 마이그레이션

모델 클래스를 정의한 후에는 메타데이터 기반으로 실제 테이블을 생성하거나 삭제해야 합니다.

# 모든 테이블 생성
Base.metadata.create_all(db_engine)

# 개발 환경에서 전체 삭제 시 유용
# Base.metadata.drop_all(db_engine)

기본 기록 관리 (CRUD)

세션 컨텍스트를 사용하여 데이터를 추가, 수정, 삭제하는 과정을 확인합니다.

저장 (Create)

with SessionLocalFactory() as current_session:
    # 단일 객체 저장
    new_acc = Account(username="dev_user_01", profile_email="dev@test.com")
    current_session.add(new_acc)
    
    # 여러 개 배치 처리
    batch_users = [
        Account(username=f"user_{i}", profile_email=f"{i}@example.org")
        for i in range(2, 5)
    ]
    current_session.add_all(batch_users)
    current_session.commit()

조회 (Read)

with SessionLocalFactory() as current_session:
    # 전량 읽기
    all_accounts = current_session.scalars(select(Account)).all()
    
    # 조건부 첫 번째 레코드
    target_account = current_session.scalar(select(Account).where(Account.uid == 1))
    
    # 필드만 추출
    names_only = current_session.execute(
        select(Account.username).where(Account.username.startswith("dev"))
    ).fetchall()

변경 및 제거 (Update & Delete)

with SessionLocalFactory() as current_session:
    # 개별 업데이트
    acc = current_session.get(Account, 1)
    if acc:
        acc.profile_email = "updated@email.com"
        current_session.flush()
        
    # 대량 갱신
    current_session.execute(
        update(Account).where(Account.username.like("temp%")).values(profile_email=None)
    )
    
    # 개별 삭제
    old_doc = current_session.scalar(select(Document).where(Document.doc_id == 99))
    if old_doc:
        current_session.delete(old_doc)
        current_session.commit()

고급 필터링 및 집계

복잡한 비즈니스 로직을 구현하기 위해 WHERE 조건의 조합과 집합 함수를 활용합니다.

from sqlalchemy import func, or_, desc

with SessionLocalFactory() as db:
    # 복합 필터링 (AND/OR)
    filtered_docs = db.execute(
        select(Document).where(
            or_(
                Document.heading.contains("파이썬"),
                Document.heading.contains("SQL")
            ),
            Document.writer_uid != 1
        ).order_by(desc(Document.doc_id)).limit(10)
    ).scalars().all()
    
    # 그룹별 집계
    stats_result = db.execute(
        select(Account.username, func.count(Document.doc_id)).join(Document).group_by(Account.uid)
    ).all()
    
    # 평균값 계산
    avg_score = db.execute(select(func.avg(Document.doc_id))).scalar()

관계형 데이터 조작

외래 키 대신 객체 참조를 이용해 자연스럽게 관계를 형성하고 관리합니다.

with SessionLocalFactory() as db:
    # 새 작성자 등록
    creator = Account(username="admin", profile_email="admin@sys.local")
    db.add(creator)
    db.flush()  # ID 할당
    
    # 작성자에게 연결된 문서 생성
    draft = Document(heading="시스템 설정 가이드", writer_uid=creator.uid)
    creator.writings.append(draft)
    
    # 태그 연관성 부여 (Many-to-Many)
    python_tag = Label(keyword="Programming")
    system_tag = Label(keyword="Infrastructure")
    
    db.add(python_tag)
    db.add(system_tag)
    db.flush()
    
    draft.categories.append(python_tag)
    draft.categories.append(system_tag)
    db.commit()

트랜잭션 격리 수준 관리

일관성을 보장하기 위해 롤백과 커밋 처리를 명확히 구분해야 합니다. 특히 오류 발생 시 자동 복구가 가능하도록 예외 처리가 필요합니다.

try:
    with SessionLocalFactory() as sess:
        # 작업 수행
        item = Account(username="temp_worker")
        sess.add(item)
        # 의도적인 에러 테스트
        raise ValueError("프로세스 실패 시뮬레이션")
except Exception as e:
    print(f"오류 감지: {e}")
    # 예외 처리 블록 내에서 세션은 이미 자동으로 닫히거나 복구될 수 있음
finally:
    # 자원 정리 확인 (필요 시 추가)
    pass

# 컨텍스트 관리자를 이용한 안전한 트랜잭션 래핑
def safe_operation(session: Session):
    try:
        user = Account(username="safe_user")
        session.add(user)
        session.flush()
        return True
    except Exception:
        session.rollback()
        return False

실전 적용을 위한 권고사항

• 세수 수명 주기 관리: 웹 프레임워크 등에서는 요청마다 새로운 세션을 생성하고 응답 종료 시 폐기하는 패턴 (Request-scoped Session) 을 권장합니다.
• 엔지니어링 패턴: 많은 양의 데이터를 불러올 때 성능 저하를 막기 위해 eager loading 옵션 (joinedload) 을 고려하세요.
• 유효성 검사: 모델 층보다는 API 입구에서 입력 데이터의 유효성을 먼저 검증하는 것이 좋습니다.
• 커넥션 풀 최적화: 고 부하 환경에서는 연결 풀 크기와 타임아웃 값을 서버 사양에 맞게 조정해야 안정성이 확보됩니다.

# 세션 관리 헬퍼 함수 정의
def get_db_session():
    session_factory = SessionLocalFactory()
    try:
        yield session_factory
        session_factory.commit()
    except Exception:
        session_factory.rollback()
        raise
    finally:
        session_factory.close()