SQLAlchemy ORM을 활용한 데이터베이스 처리 방법

Python에서 SQLAlchemy는 인기 있는 ORM(Object-Relational Mapping) 프레임워크로, 효율적이고 유연한 데이터베이스 조작을 제공합니다. 이번 문서에서는 SQLAlchemy ORM을 사용하여 데이터베이스를 다루는 방법을 설명합니다.

목차

  1. 설치
  2. 핵심 개념
  3. 데이터베이스 연결
  4. 데이터 모델 정의
  5. 테이블 생성
  6. 기본 CRUD 작업
  7. 데이터 조회
  8. 관계 작업
  9. 트랜잭션 관리
  10. 최고의 실천 방안

설치

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

# 엔진 생성 (SQLite 예시)
db_url = 'sqlite:///test.db'
engine = create_engine(db_url, echo=True)

# 세션 공장 생성
SessionFactory = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# 세션 인스턴스 생성
session = SessionFactory()

데이터 모델 정의

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

Base = declarative_base()

class Member(Base):
    __tablename__ = 'members'

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)

    # 관계 설정
    articles = relationship("Article", back_populates="author")

class Article(Base):
    __tablename__ = 'articles'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(String(500))
    author_id = Column(Integer, ForeignKey('members.id'))

    # 관계 설정
    author = relationship("Member", back_populates="articles")
    
    categories = relationship("Category", secondary="article_categories", back_populates="articles")

class Category(Base):
    __tablename__ = 'categories'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(30), unique=True, nullable=False)

    articles = relationship("Article", secondary="article_categories", back_populates="categories")

class ArticleCategory(Base):
    __tablename__ = 'article_categories'

    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)

테이블 생성

Base.metadata.create_all(bind=engine)

기본 CRUD 작업

데이터 생성

new_member = Member(username="JohnDoe", email="john@example.com")
session.add(new_member)
session.commit()

# 일괄 추가
session.add_all([
    Member(username="JaneDoe", email="jane@example.com"),
    Member(username="Alice", email="alice@example.com")
])
session.commit()

데이터 읽기

# 모든 멤버 가져오기
members = session.query(Member).all()

# 첫 번째 멤버 가져오기
first_member = session.query(Member).first()

# ID로 멤버 가져오기
member = session.query(Member).get(1)

데이터 수정

# 특정 멤버 찾기 및 수정
member = session.query(Member).get(1)
member.username = "Johnathan"
session.commit()

# 일괄 수정
session.query(Member).filter(Member.username.like("J%")).update({"username": "UpdatedName"}, synchronize_session=False)
session.commit()

데이터 삭제

# 특정 멤버 삭제
member = session.query(Member).get(1)
session.delete(member)
session.commit()

# 일괄 삭제
session.query(Member).filter(Member.username == "Alice").delete(synchronize_session=False)
session.commit()

데이터 조회

기본 쿼리

# 모든 레코드 가져오기
members = session.query(Member).all()

# 특정 필드만 가져오기
usernames = session.query(Member.username).all()

# 정렬
members = session.query(Member).order_by(Member.username.desc()).all()

# 결과 수 제한
members = session.query(Member).limit(10).all()

# 오프셋 설정
members = session.query(Member).offset(5).limit(10).all()

필터링 쿼리

from sqlalchemy import or_

# 동일 값 필터링
member = session.query(Member).filter(Member.username == "JohnDoe").first()

# 부분 일치 검색
members = session.query(Member).filter(Member.username.like("J%")).all()

# IN 조건 검색
members = session.query(Member).filter(Member.username.in_(["JohnDoe", "JaneDoe"])).all()

# 다중 조건 검색
members = session.query(Member).filter(
    Member.username == "JohnDoe",
    Member.email.like("%@example.com")
).all()

# OR 조건 검색
members = session.query(Member).filter(
    or_(Member.username == "JohnDoe", Member.username == "JaneDoe")
).all()

# NOT 조건 검색
members = session.query(Member).filter(Member.username != "JohnDoe").all()

집계 쿼리

from sqlalchemy import func

# 레코드 수 세기
count = session.query(Member).count()

# 그룹화된 카운트
member_article_count = session.query(
    Member.username,
    func.count(Article.id)
).join(Article).group_by(Member.username).all()

# 합계 및 평균 계산
avg_id = session.query(func.avg(Member.id)).scalar()

조인 쿼리

# 내부 조인
results = session.query(Member, Article).join(Article).filter(Article.title.like("%Python%")).all()

# 외부 조인
results = session.query(Member, Article).outerjoin(Article).all()

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

관계 작업

# 관계 포함 데이터 생성
member = Member(username="Bob", email="bob@example.com")
article = Article(title="첫 번째 게시물", content="안녕하세요!", author=member)
session.add(article)
session.commit()

# 관계 통해 접근
print(f"게시물 '{article.title}'의 작성자는 {article.author.username}")
print(f"사용자 {member.username}의 모든 게시물:")
for a in member.articles:
    print(f"  - {a.title}")

# 다대다 관계 작업
python_category = Category(name="Python")
sqlalchemy_category = Category(name="SQLAlchemy")

article.categories.append(python_category)
article.categories.append(sqlalchemy_category)
session.commit()

print(f"게시물 '{article.title}'의 카테고리:")
for c in article.categories:
    print(f"  - {c.name}")

트랜잭션 관리

# 자동 커밋 트랜잭션
try:
    member = Member(username="TestUser", email="test@example.com")
    session.add(member)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"오류 발생: {e}")

# 트랜잭션 컨텍스트 관리자 사용
def create_member(session, username, email):
    try:
        member = Member(username=username, email=email)
        session.add(member)
        session.commit()
        return member
    except:
        session.rollback()
        raise

# 중첩 트랜잭션
with session.begin_nested():
    member = Member(username="NestedUser", email="nested@example.com")
    session.add(member)

# 저장점(Savepoint)
savepoint = session.begin_nested()
try:
    member = Member(username="SavepointUser", email="savepoint@example.com")
    session.add(member)
    savepoint.commit()
except:
    savepoint.rollback()

최고의 실천 방안

  1. 세션 관리: 각 요청마다 새 세션 생성 후 요청 종료 시 닫기.
  2. 예외 처리: 항상 예외 처리하고 적절히 롤백.
  3. 지연 로딩: N+1 문제 주의, eager loading으로 최적화.
  4. 커넥션 풀: 커넥션 풀 크기 및 타임아웃 설정 중요.
  5. 데이터 검증: 모델 또는 애플리케이션 레벨에서 데이터 무결성 검증.
from contextlib import contextmanager

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

# 사용 예시
with get_session() as db:
    member = Member(username="ContextUser", email="context@example.com")
    db.add(member)

요약

SQLAlchemy ORM은 강력하고 유연한 데이터베이스 작업 기능을 제공하며, 이를 통해 다음 작업을 수행할 수 있습니다:

  1. SQLAlchemy 설치 및 구성
  2. 데이터 모델 및 관계 정의
  3. 기본 CRUD 작업 수행
  4. 복잡한 쿼리 작성
  5. 데이터베이스 트랜잭션 관리
  6. 최고의 실천 방안 준수

SQLAlchemy에는 혼합 속성, 이벤트 리스너, 사용자 정의 쿼리 등 더 많은 고급 기능이 있으며, 이를 연구하면 더 나은 데이터베이스 작업을 할 수 있습니다.

태그: sqlalchemy ORM python

7월 12일 00:35에 게시됨

인기 태그

©2026 괴물 클럽