SQLAlchemy를 활용한 데이터베이스 관리 및 운영

ORM 기반의 데이터 접근: SQLAlchemy 소개

Python 환경에서 가장 널리 사용되는 객체 관계 매핑(ORM) 도구인 SQLAlchemy는 데이터베이스 작업을 객체 지향 방식으로 간편하게 처리할 수 있도록 지원합니다. 이 문서에서는 실제 애플리케이션 개발에 적용 가능한 핵심 개념과 실용적인 예제를 중심으로 다룹니다.

설치 및 환경 구성

pip install sqlalchemy

다른 데이터베이스와 연동 시, 해당 드라이버도 함께 설치해야 합니다:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite는 기본 라이브러리 포함 → 별도 설치 불필요

핵심 구성 요소

  • 엔진(Engine): 데이터베이스 연결을 담당하는 커넥션 풀과 통신 중개자
  • 세션(Session): 데이터 변경 작업을 일괄적으로 관리하는 단위
  • 모델(Model): 데이터베이스 테이블과 매핑되는 파이썬 클래스
  • 쿼리(Query): 조건 기반의 데이터 검색 및 수정 명령을 구성하는 객체

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 기반 엔진 생성 (디버깅용 로그 활성화)
engine = create_engine('sqlite:///app_data.db', echo=True)

# 세션 팩토리 정의
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

# 세션 인스턴스 생성
db_session = SessionLocal()

모델 정의 및 관계 설정

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, index=True)
    full_name = Column(String(80), nullable=False)
    email = Column(String(120), unique=True, index=True)

    # 1:N 관계: 작성자 - 글
    articles = relationship("Article", back_populates="writer")

class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    body = Column(String(2000))
    writer_id = Column(Integer, ForeignKey('authors.id'))

    # N:1 관계: 글 - 작성자
    writer = relationship("Author", back_populates="articles")

    # M:N 관계: 글 - 태그
    tags = relationship("Tag", secondary="article_tags", back_populates="articles")

class Tag(Base):
    __tablename__ = 'tags'

    id = Column(Integer, primary_key=True, index=True)
    label = Column(String(50), 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)

기본 데이터 조작 작업

생성 (Create)

# 단일 항목 삽입
new_author = Author(full_name="김철수", email="kim@example.com")
db_session.add(new_author)
db_session.commit()

# 여러 항목 일괄 삽입
db_session.add_all([
    Author(full_name="박영희", email="park@example.com"),
    Author(full_name="최민수", email="choi@example.com")
])
db_session.commit()

조회 (Read)

# 전체 조회
all_authors = db_session.query(Author).all()

# 첫 번째 항목
first = db_session.query(Author).first()

# ID로 조회
author_by_id = db_session.query(Author).get(1)

업데이트 (Update)

# 특정 항목 수정
target = db_session.query(Author).get(1)
target.full_name = "김철수 변경됨"
db_session.commit()

# 조건에 맞는 대량 수정
db_session.query(Author).filter(
    Author.full_name.like("김%")
).update({
    "email": lambda x: f"{x}@updated.com"
}, synchronize_session=False)
db_session.commit()

삭제 (Delete)

# 특정 항목 삭제
target = db_session.query(Author).get(1)
db_session.delete(target)
db_session.commit()

# 조건에 맞는 대량 삭제
db_session.query(Author).filter(
    Author.email == "park@example.com"
).delete(synchronize_session=False)
db_session.commit()

고급 쿼리 기법

조건 필터링

from sqlalchemy import or_

# 정확한 일치
result = db_session.query(Author).filter(Author.full_name == "김철수").first()

# 와일드카드 검색
matches = db_session.query(Author).filter(Author.full_name.like("%철%")).all()

# 멀티 값 확인
selected = db_session.query(Author).filter(Author.full_name.in_(['김철수', '박영희'])).all()

# 복합 조건
complex_query = db_session.query(Author).filter(
    Author.full_name == "김철수",
    Author.email.like("%@example.com")
).all()

# OR 조건
or_result = db_session.query(Author).filter(
    or_(Author.full_name == "김철수", Author.full_name == "박영희")
).all()

집계 함수 사용

from sqlalchemy import func

# 카운트
total_count = db_session.query(Author).count()

# 그룹화 집계
grouped_stats = db_session.query(
    Author.full_name,
    func.count(Article.id)
).join(Article).group_by(Author.full_name).all()

# 평균 계산
avg_id = db_session.query(func.avg(Author.id)).scalar()

조인 쿼리

# 내부 조인
joined_results = db_session.query(Author, Article).join(Article).filter(
    Article.title.like("%파이썬%")
).all()

# 외부 조인
left_join = db_session.query(Author, Article).outerjoin(Article).all()

# 조건 기반 조인
conditional_join = db_session.query(Author, Article).join(
    Article, Author.id == Article.writer_id
).all()

관계 탐색 및 연관 데이터 처리

# 관계를 통해 객체 연결
author = Author(full_name="이지은", email="lee@example.com")
article = Article(title="초보자를 위한 튜토리얼", body="내용입니다.", writer=author)
db_session.add(article)
db_session.commit()

# 관계 속성 접근
print(f"글 '{article.title}'의 작성자는 {article.writer.full_name}입니다.")

# 다수의 글 목록 조회
for art in author.articles:
    print(f"- {art.title}")

# 다대다 관계 추가
python_tag = Tag(label="Python")
sqlalchemy_tag = Tag(label="SQLAlchemy")

article.tags.append(python_tag)
article.tags.append(sqlalchemy_tag)
db_session.commit()

# 태그 출력
for tag in article.tags:
    print(f"태그: {tag.label}")

트랜잭션 제어 및 오류 처리

# 예외 처리를 통한 롤백 보장
try:
    new_user = Author(full_name="테스트사용자", email="test@ex.com")
    db_session.add(new_user)
    db_session.commit()
except Exception as e:
    db_session.rollback()
    raise e

# 컨텍스트 관리자 활용
from contextlib import contextmanager

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

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

실무 권장 사항

  1. 세션 생명주기 관리: 요청마다 새로운 세션 생성 후 종료
  2. 예외 처리 강화: 트랜잭션 실패 시 자동 롤백 보장
  3. 지연 로딩 최적화: N+1 문제 방지를 위해 전제 로딩 또는 선택적 로딩 적용
  4. 커넥션 풀 설정: 동시성과 성능을 고려해 적절한 크기 및 타임아웃 조정
  5. 입력 검증: 모델 또는 서비스 계층에서 유효성 검사 수행

결론

SQLAlchemy는 단순한 데이터베이스 접속 도구를 넘어, 복잡한 애플리케이션 아키텍처에서도 안정적으로 작동할 수 있는 강력한 ORM 프레임워크입니다. 본 문서를 통해 다음과 같은 역량을 갖출 수 있습니다:

  • 환경 설정 및 엔진 구성 이해
  • 객체 기반 모델 설계 및 관계 정의
  • CRUD 연산 및 복잡한 쿼리 작성 능력
  • 트랜잭션 및 오류 처리 구현
  • 실제 서비스에 적용 가능한 최적화 전략 적용

추가로, 이벤트 리스너, 하이브리드 속성, 커스텀 쿼리 빌더 등 고급 기능들은 확장성을 높이는 핵심 요소입니다.

태그: sqlalchemy python ORM 데이터베이스 트랜잭션

7월 15일 01:15에 게시됨

인기 태그

©2026 괴물 클럽