SQLAlchemy ORM을 활용한 데이터베이스 운영 및 관리 기법

SQLAlchemy 설치 및 환경 구성

SQLAlchemy는 파이썬에서 데이터베이스 작업을 간편하게 수행할 수 있도록 도와주는 대표적인 객체 관계 매핑(ORM) 프레임워크입니다. 기본 설치를 위해 다음 명령어를 실행하세요.

pip install sqlalchemy
특정 데이터베이스 시스템과 연결하기 위해서는 추가 드라이버 설치가 필요합니다. 
# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (Python 표준 라이브러리 포함, 별도 설치 불필요)

핵심 구성 요소 이해

  • Engine: 데이터베이스 연결의 중앙 제어자로, 실제 통신을 담당합니다.
  • Session: 데이터베이스 작업의 일관성을 보장하는 세션 단위 처리 흐름입니다.
  • Model: 데이터베이스 테이블에 대응하는 파이썬 클래스입니다.
  • Query: SQL 쿼리를 생성하고 실행하는 객체입니다.

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 사용 예시
engine = create_engine('sqlite:///app_database.db', echo=True)

# PostgreSQL 연결 예시
# engine = create_engine('postgresql://user:pass@localhost:5432/myapp')

# MySQL 연결 예시
# engine = create_engine('mysql+mysqlconnector://user:pass@localhost:3306/myapp')

# 세션 팩토리 생성
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)
    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)
    title = Column(String(150), nullable=False)
    body = Column(String(1000))
    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)
    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)

기본적인 데이터 조작 (CRUD)

생성

# 개별 삽입
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()

조회

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

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

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

수정

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

삭제

# 항목 삭제
item_to_delete = db_session.query(Author).get(1)
if item_to_delete:
    db_session.delete(item_to_delete)
    db_session.commit()

복잡한 쿼리 작성

조건 필터링

from sqlalchemy import or_

# 정확한 값 검색
result = db_session.query(Author).filter(Author.email == "kim@example.com").first()

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

# 리스트 내 포함 여부
filtered = 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_data = db_session.query(Author, Article).join(Article).filter(
    Article.title.like("%파이썬%")
).all()

# 외부 조인
outer_data = 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 a in author.articles:
    print(f"  - {a.title}")

# 다대다 관계 추가
py_tag = Tag(label="Python")
sql_tag = Tag(label="SQLAlchemy")

article.tags.append(py_tag)
article.tags.append(sql_tag)
db_session.commit()

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

트랜잭션 제어 및 안정성 보장

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

# 트랜잭션 컨텍스트 사용
def register_author(session, name, email):
    try:
        user = Author(full_name=name, email=email)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise

# 중첩 트랜잭션
with db_session.begin_nested():
    temp_user = Author(full_name="내부 트랜잭션", email="nested@test.com")
    db_session.add(temp_user)

# 저장점 사용
savepoint = db_session.begin_nested()
try:
    save_user = Author(full_name="저장점 테스트", email="save@test.com")
    db_session.add(save_user)
    savepoint.commit()
except:
    savepoint.rollback()

권장되는 실무 팁

  1. 세션 생명주기 관리: 각 요청마다 새 세션 생성 후 종료해야 합니다.
  2. 예외 처리: 실패 시 자동 롤백을 보장하세요.
  3. 지연 로딩 최적화: N+1 문제를 피하기 위해 지연 로딩을 적절히 설정하세요.
  4. 커넥션 풀 설정: 동시성과 성능을 고려해 적절한 풀 크기 및 타임아웃을 구성하세요.
  5. 데이터 유효성 검사: 모델 또는 비즈니스 계층에서 입력 검증을 수행하세요.
from contextlib import contextmanager

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

# 사용 예시
with get_db_session() as db:
    user = Author(full_name="컨텍스트 사용자", email="context@test.com")
    db.add(user)

결론

SQLAlchemy ORM은 데이터베이스 접근을 매우 직관적이고 안정적으로 만들어줍니다. 본 문서를 통해 다음과 같은 능력을 갖출 수 있습니다:

  • SQLAlchemy 설치 및 초기 설정
  • 모델 정의 및 관계 구현
  • 기본 데이터 조작(생성, 조회, 수정, 삭제)
  • 복잡한 쿼리 및 조인 문법 사용
  • 트랜잭션 및 에러 처리 전략 적용
  • 실제 서비스 환경에서의 최적화 방법 이해

추가로, 하이브리드 속성, 이벤트 리스너, 사용자 정의 쿼리 등 고급 기능은 심화 학습을 통해 더욱 깊이 있게 탐구할 수 있습니다.

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

6월 22일 02:23에 게시됨

인기 태그

©2026 괴물 클럽