SQLAlchemy를 통한 고성능 데이터 접근 방법

SQLAlchemy는 파이썬에서 가장 널리 사용되는 객체 관계 매핑(ORM) 프레임워크로, 데이터베이스 작업을 간결하고 안정적으로 수행할 수 있도록 지원합니다. 본 문서에서는 실용적인 예제를 중심으로 데이터 모델 정의, 쿼리 최적화, 트랜잭션 처리 등 핵심 기능을 다룹니다.

설치 및 환경 구성

pip install sqlalchemy

다른 데이터베이스와 연동 시 추가 패키지 설치 필요:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (기본 라이브러리 포함)

핵심 구성 요소

• Engine: 데이터베이스 연결을 관리하는 커널 역할
• Session: 영속성 작업의 단위, 일련의 변경 사항을 추적
• Model: 테이블 구조를 클래스로 표현
• Query: SQL 문을 생성하고 실행하는 인터페이스

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///app.db", echo=False)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

모델 정의: 관계형 데이터 구조 설계

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

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    
    id = Column(Integer, primary_key=True)
    full_name = Column(String(80), nullable=False)
    email = Column(String(120), unique=True)

    # 1:N 관계
    articles = relationship("Article", back_populates="writer")

class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(150), nullable=False)
    body = Column(String(2000))
    writer_id = Column(Integer, ForeignKey('authors.id'))

    # N:1 관계
    writer = relationship("Author", back_populates="articles")

    # 다대다 관계 준비
    tags = relationship("Tag", secondary="article_tags", back_populates="articles")

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    label = Column(String(40), unique=True, nullable=False)

    articles = relationship("Article", secondary="article_tags", back_populates="tags")

class ArticleTag(Base):
    __tablename__ = 'article_tags'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

테이블 생성 및 관리

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

# 특정 테이블만 삭제
# Base.metadata.drop_all(bind=engine, tables=[Article.__table__, Author.__table__])

CRUD 작업: 데이터 조작 기본

session = SessionLocal()

# 삽입
new_author = Author(full_name="김철수", email="kim@example.com")
session.add(new_author)
session.commit()

# 일괄 삽입
session.add_all([
    Author(full_name="박영희", email="park@example.com"),
    Author(full_name="이민수", email="lee@example.com")
])
session.commit()

# 조회
first_article = session.query(Article).filter_by(title="첫 번째 글").first()
all_authors = session.query(Author).all()

# 업데이트
target = session.query(Author).get(1)
if target:
    target.full_name = "김철수(수정됨)"
    session.commit()

# 삭제
to_delete = session.query(Article).get(5)
if to_delete:
    session.delete(to_delete)
    session.commit()

고급 쿼리 작성 기법

from sqlalchemy import func, or_

# 필드 선택
titles = session.query(Article.title).all()

# 정렬 및 제한
recent_articles = (
    session.query(Article)
    .order_by(Article.id.desc())
    .limit(5)
    .all()
)

# 조건 검색
matching_authors = (
    session.query(Author)
    .filter(
        or_(
            Author.full_name.like("%김%"),
            Author.email.like("%@example.com")
        )
    )
    .all()
)

# 그룹화 집계
author_stats = (
    session.query(
        Author.full_name,
        func.count(Article.id).label('article_count')
    )
    .join(Article)
    .group_by(Author.full_name)
    .all()
)

# 조인 쿼리
joined_results = (
    session.query(Article, Author)
    .join(Author, Article.writer_id == Author.id)
    .filter(Article.title.contains("Python"))
    .all()
)

관계 탐색 및 성능 최적화

# 관계 기반 데이터 접근
author = session.query(Author).get(1)
print(f"작가: {author.full_name}")
for art in author.articles:
    print(f"글: {art.title}")

# 다대다 관계 처리
article = session.query(Article).get(1)
python_tag = session.query(Tag).filter_by(label="Python").first()
article.tags.append(python_tag)
session.commit()

트랜잭션 제어 및 오류 처리

try:
    new_user = Author(full_name="테스트 유저", email="test@local.com")
    session.add(new_user)
    session.commit()
except Exception as e:
    session.rollback()
    raise e

# 컨텍스트 매니저 활용
from contextlib import contextmanager

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

# 사용 예시
with get_session() as db:
    user = Author(full_name="컨텍스트 유저", email="ctx@example.com")
    db.add(user)

권장 사항 및 실전 팁

• 세션은 요청 단위로 생성하고 반드시 종료
• 복잡한 관계 조회 시 joinedload 사용으로 부하 감소
• 대량 데이터 처리 시 yield_per 옵션 적용
• 연결 풀 크기를 서버 리소스에 맞게 조절
• 입력값 검증은 모델 또는 서비스 계층에서 수행