SQLAlchemy ORM을 활용한 데이터베이스 작업 가이드

1. 설치

pip install sqlalchemy

특정 데이터베이스 연결 시 추가 패키지 설치:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (표준 라이브러리 포함)

2. 핵심 개념

  • 엔진: 데이터베이스 통신을 담당하는 연결 객체
  • 세션: 데이터베이스 작업을 관리하는 컨텍스트
  • 모델: 테이블 구조를 정의하는 클래스
  • 쿼리: 데이터 조회 및 조작을 위한 인터페이스

3. 데이터베이스 연결

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 데이터베이스 엔진 생성
db_engine = create_engine('sqlite:///example.db', echo=True)

# 세션 팩토리 설정
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

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

4. 데이터 모델 정의

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

Base = declarative_base()

class 사용자(Base):
    __tablename__ = '사용자'
    
    id = Column(Integer, primary_key=True, index=True)
    이름 = Column(String(50), nullable=False)
    이메일 = Column(String(100), unique=True, index=True)
    
    게시물 = relationship("게시물", back_populates="작성자")
    
class 게시물(Base):
    __tablename__ = '게시물'
    
    id = Column(Integer, primary_key=True, index=True)
    제목 = Column(String(100), nullable=False)
    내용 = Column(String(500))
    작성자_id = Column(Integer, ForeignKey('사용자.id'))
    
    작성자 = relationship("사용자", back_populates="게시물")
    
    태그 = relationship("태그", secondary="게시물태그", back_populates="게시물")

class 태그(Base):
    __tablename__ = '태그'
    
    id = Column(Integer, primary_key=True, index=True)
    이름 = Column(String(30), unique=True, nullable=False)
    
    게시물 = relationship("게시물", secondary="게시물태그", back_populates="태그")

class 게시물태그(Base):
    __tablename__ = '게시물태그'
    
    게시물_id = Column(Integer, ForeignKey('게시물.id'), primary_key=True)
    태그_id = Column(Integer, ForeignKey('태그.id'), primary_key=True)

5. 테이블 생성

# 테이블 생성
Base.metadata.create_all(bind=db_engine)

# 테이블 삭제
# Base.metadata.drop_all(bind=db_engine)

6. 기본 CRUD 연산

데이터 생성

# 개별 생성
새_사용자 = 사용자(이름="장삼", 이메일="zhangsan@example.com")
db_session.add(새_사용자)
db_session.commit()

# 집합 생성
db_session.add_all([
    사용자(이름="려사", 이메일="lisi@example.com"),
    사용자(이름="왕오", 이메일="wangwu@example.com")
])
db_session.commit()

데이터 조회

# 전체 조회
사용자_목록 = db_session.query(사용자).all()

# 특정 ID 조회
사용자 = db_session.query(사용자).get(1)

데이터 수정

# 개별 수정
사용자 = db_session.query(사용자).get(1)
사용자.이름 = "장삼사"
db_session.commit()

# 집합 수정
db_session.query(사용자).filter(사용자.이름.like("장%")).update({"이름": "장씨"}, synchronize_session=False)
db_session.commit()

데이터 삭제

# 개별 삭제
사용자 = db_session.query(사용자).get(1)
db_session.delete(사용자)
db_session.commit()

# 집합 삭제
db_session.query(사용자).filter(사용자.이름 == "려사").delete(synchronize_session=False)
db_session.commit()

7. 쿼리 처리

기본 쿼리

# 전체 데이터 조회
사용자_목록 = db_session.query(사용자).all()

# 특정 필드 조회
이름_목록 = db_session.query(사용자.이름).all()

# 정렬
사용자_목록 = db_session.query(사용자).order_by(사용자.이름.desc()).all()

# 결과 제한
사용자_목록 = db_session.query(사용자).limit(10).all()

# 페이지네이션
사용자_목록 = db_session.query(사용자).offset(5).limit(10).all()

필터링 쿼리

from sqlalchemy import or_

# 정확한 값 필터링
사용자 = db_session.query(사용자).filter(사용자.이름 == "장삼").first()

# 부분 일치 검색
사용자_목록 = db_session.query(사용자).filter(사용자.이름.like("장%")).all()

# IN 조건
사용자_목록 = db_session.query(사용자).filter(사용자.이름.in_(["장삼", "려사"])).all()

# 복합 조건
사용자_목록 = db_session.query(사용자).filter(
    사용자.이름 == "장삼", 
    사용자.이메일.like("%@example.com")
).all()

# OR 조건
사용자_목록 = db_session.query(사용자).filter(
    or_(사용자.이름 == "장삼", 사용자.이름 == "려사")
).all()

8. 관계 처리

# 관계 객체 생성
사용자 = 사용자(이름="조육", 이메일="zhaoliu@example.com")
게시물 = 게시물(제목="첫 번째 글", 내용="안녕하세요!", 작성자=사용자)
db_session.add(게시물)
db_session.commit()

# 관계 조회
print(f"글 '{게시물.제목}'의 작성자는 {게시물.작성자.이름}")
print(f"사용자 {사용자.이름}의 글 목록:")
for g in 사용자.게시물:
    print(f"  - {g.제목}")

# 다대다 관계 처리
파이썬_태그 = 태그(이름="Python")
sqlalchemy_태그 = 태그(이름="SQLAlchemy")

게시물.태그.append(파이썬_태그)
게시물.태그.append(sqlalchemy_태그)
db_session.commit()

9. 트랜잭션 관리

# 자동 커밋 트랜잭션
try:
    사용자 = 사용자(이름="테스트 사용자", 이메일="test@example.com")
    db_session.add(사용자)
    db_session.commit()
except Exception as e:
    db_session.rollback()
    print(f"오류 발생: {e}")

# 컨텍스트 매니저 사용
def 사용자_생성(db_session: Session, 이름: str, 이메일: str):
    try:
        사용자 = 사용자(이름=이름, 이메일=이메일)
        db_session.add(사용자)
        db_session.commit()
        return 사용자
    except:
        db_session.rollback()
        raise

10. 최적화 전략

  • 세션 관리: 요청별 새로운 세션 생성
  • 예외 처리: 모든 예외에 대해 적절히 롤백
  • 지연 로딩: N+1 문제 방지를 위해 eager loading 적용
  • 커넥션 풀: 최적의 풀 크기와 타임아웃 설정
  • 데이터 검증: 모델 또는 애플리케이션 레벨에서 데이터 무결성 검증

태그: sqlalchemy ORM 데이터베이스 python 관계형데이터베이스

6월 15일 17:59에 게시됨

인기 태그

©2026 괴물 클럽