SQLAlchemy ORM 마스터하기: 고급 데이터베이스 연동 가이드

SQLAlchemy 개요

SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(Object-Relational Mapping) 라이브러리로, 객체 지향적인 방식으로 데이터베이스를 조작할 수 있게 해줍니다. 이 가이드에서는 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

# 엔진 생성
conn_str = 'sqlite:///database.db'
engine = create_engine(conn_str, echo=False)

# 세션 팩토리 정의
SessionFactory = sessionmaker(bind=engine)
session = SessionFactory()

데이터 모델 정의

from sqlalchemy import Column, Integer, String, ForeignKey, Text
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_address = Column(String(120), unique=True)
    
    articles = relationship("Article", back_populates="writer")

class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    headline = Column(String(200), nullable=False)
    body = Column(Text)
    author_id = Column(Integer, ForeignKey('authors.id'))
    
    writer = relationship("Author", back_populates="articles")
    keywords = relationship("Keyword", secondary="article_keywords", back_populates="articles")

class Keyword(Base):
    __tablename__ = 'keywords'
    
    id = Column(Integer, primary_key=True)
    term = Column(String(40), unique=True, nullable=False)
    
    articles = relationship("Article", secondary="article_keywords", back_populates="keywords")

class ArticleKeyword(Base):
    __tablename__ = 'article_keywords'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    keyword_id = Column(Integer, ForeignKey('keywords.id'), primary_key=True)

테이블 생성/삭제

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

# 테이블 삭제
# Base.metadata.drop_all(engine)

기본 CRUD 연산

레코드 생성

# 단일 객체 생성
new_author = Author(full_name="홍길동", email_address="hong@example.com")
session.add(new_author)
session.commit()

# 여러 객체 한 번에 추가
session.add_all([
    Author(full_name="김철수", email_address="kim@example.com"),
    Author(full_name="이영희", email_address="lee@example.com")
])
session.commit()

데이터 조회

# 전체 레코드 가져오기
all_authors = session.query(Author).all()

# 첫 번째 레코드
first_author = session.query(Author).first()

# 기본 키로 조회
author = session.query(Author).get(2)

레코드 수정

# 단일 업데이트
author = session.query(Author).get(1)
author.full_name = "홍길동(수정)"
session.commit()

# 조건부 일괄 업데이트
session.query(Author).filter(
    Author.email_address.like("%@old-domain.com")
).update({"email_address": Author.email_address.replace("@old-domain.com", "@new-domain.com")})
session.commit()

데이터 삭제

# 단일 삭제
target = session.query(Author).get(3)
session.delete(target)
session.commit()

# 조건부 일괄 삭제
session.query(Author).filter(
    Author.email_address == None
).delete()
session.commit()

고급 쿼리 패턴

필터링 조건

from sqlalchemy import or_, and_

# 정확한 일치
result = session.query(Author).filter(Author.full_name == "홍길동").first()

# 패턴 검색
results = session.query(Author).filter(Author.full_name.like("김%")).all()

# 포함 여부
results = session.query(Author).filter(
    Author.full_name.in_(["김철수", "이영희"])
).all()

# 복합 조건
results = session.query(Article).filter(
    and_(
        Article.headline.contains("Python"),
        Article.body != None
    )
).all()

# 논리 OR
results = session.query(Author).filter(
    or_(
        Author.full_name == "박지성",
        Author.email_address.like("%@premier.com")
    )
).all()

정렬 및 페이징

# 정렬
sorted_authors = session.query(Author).order_by(
    Author.full_name.asc()
).all()

# 페이징
page = 1
page_size = 10
paged_results = session.query(Article).offset(
    (page - 1) * page_size
).limit(page_size).all()

집계 함수

from sqlalchemy import func

# 총 개수
total = session.query(Article).count()

# 그룹별 통계
stats = session.query(
    Author.full_name,
    func.count(Article.id).label('article_count')
).join(Article).group_by(Author.id).all()

# 평균 및 최대값
avg_id = session.query(func.avg(Article.id)).scalar()

조인 쿼리

# 내부 조인
joined_data = session.query(Author, Article).join(Article).filter(
    Article.headline.like("%SQL%")
).all()

# 외부 조인
outer_join = session.query(Author, Article).outerjoin(Article).all()

# 명시적 조인 조건
custom_join = session.query(Author, Article).join(
    Article, Author.id == Article.author_id
).all()

관계 매핑 활용

# 관계 생성
writer = Author(full_name="최민수", email_address="choi@example.com")
post = Article(
    headline="SQLAlchemy 완벽 가이드",
    body="SQLAlchemy는 강력한 ORM입니다...",
    writer=writer
)
session.add(post)
session.commit()

# 객체 탐색
print(f"기사 '{post.headline}' 저자: {post.writer.full_name}")
print(f"{writer.full_name}의 기사 목록:")
for article in writer.articles:
    print(f"  - {article.headline}")

# 다대다 관계
python_kw = Keyword(term="Python")
orm_kw = Keyword(term="ORM")
post.keywords.append(python_kw)
post.keywords.append(orm_kw)
session.commit()

print(f"기사 '{post.headline}' 키워드:")
for kw in post.keywords:
    print(f"  - {kw.term}")

트랜잭션 관리

# 명시적 트랜잭션 처리
try:
    new_entry = Author(full_name="트랜잭션유저", email_address="tx@example.com")
    session.add(new_entry)
    session.commit()
except:
    session.rollback()
    print("트랜잭션 실패 - 롤백됨")

# 중첩 트랜잭션(저장점)
with session.begin_nested():
    temp_entry = Author(full_name="임시저장", email_address="temp@example.com")
    session.add(temp_entry)
    # 롤백 시점 설정 가능

세션 관리 컨텍스트 매니저

from contextlib import contextmanager

@contextmanager
def managed_session():
    db_session = SessionFactory()
    try:
        yield db_session
        db_session.commit()
    except:
        db_session.rollback()
        raise
    finally:
        db_session.close()

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

권장 사례

  1. 세션 범위 제한: 각 요청/작업 단위로 세션을 생성하고 종료
  2. 적극적 예외 처리: 작업 실패 시 항상 롤백 로직 포함
  3. 지연 로딩 주의: N+1 쿼리를 방지하기 위해 필요 시 joinedload() 활용
  4. 연결 풀 최적화: 애플리케이션 부하에 맞춰 풀 크기와 타임아웃 조정
  5. 데이터 검증 계층: 모델 레벨이나 애플리케이션 레벨에서 무결성 보장

핵심 요약

SQLAlchemy ORM을 통해 얻을 수 있는 이점:

  • 객체 지향적 데이터베이스 작업
  • 강력한 쿼리 빌더와 관계 매핑
  • 유연한 트랜잭션 제어
  • 데이터베이스 독립적인 코드 작성

더 깊이 있는 학습을 위해 혼합 속성(mixin), 이벤트 시스템, 맞춤형 쿼리 확장 등 고급 기능을 탐색해보세요.

태그: sqlalchemy ORM 파이썬 데이터베이스 CRUD

5월 24일 01:15에 게시됨

인기 태그

©2026 괴물 클럽