SQLAlchemy ORM 실전 활용 가이드

SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(객체 관계 매핑) 프레임워크 중 하나로, 데이터베이스 작업을 효율적이고 유연하게 수행할 수 있는 기능을 제공합니다. 본 문서에서는 SQLAlchemy ORM을 활용한 데이터베이스 작업 방법에 대해 안내합니다.

목차

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

설치

bash

pip install sqlalchemy

특정 데이터베이스와 연동하려면 추가 드라이버 설치가 필요합니다:

bash

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

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

핵심 개념

  • 엔진 : 데이터베이스와 통신하는 연결 엔진
  • 세션 : 지속성 작업을 관리하는 세션 객체
  • 모델 : 데이터베이스 테이블과 매핑되는 클래스
  • 쿼리 : 데이터베이스 쿼리를 구성하고 실행하는 객체

데이터베이스 연결

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 연결 예시
db_engine = create_engine('sqlite:///sample.db', echo=True)

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

# MySQL 연결 예시
# db_engine = create_engine('mysql+connector://user:pass@localhost:3306/dbname')

# 세션 생성기 설정
SessionClass = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

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

데이터 모델 정의

python

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

Base = declarative_base()

class Person(Base):
    __tablename__ = 'people'
    
    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(50), nullable=False)
    email_address = Column(String(100), 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(100), nullable=False)
    content = Column(String(500))
    writer_id = Column(Integer, ForeignKey('people.id'))
    
    # N:1 관계 정의
    writer = relationship("Person", back_populates="articles")
    
    # N:N 관계 정의(중간 테이블 사용)
    tags = relationship("Tag", secondary="article_tags", back_populates="articles")

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True, index=True)
    tag_name = Column(String(30), 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)

테이블 생성

python

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

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

기본 CRUD 작업

데이터 생성

python

# 새로운 사람 추가
new_person = Person(full_name="홍길동", email_address="hong@example.com")
db_session.add(new_person)
db_session.commit()

# 병렬 추가
db_session.add_all([
    Person(full_name="김철수", email_address="kim@example.com"),
    Person(full_name="박영희", email_address="park@example.com")
])
db_session.commit()

데이터 조회

python

# 모든 사람 목록
all_people = db_session.query(Person).all()

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

# ID로 조회
person = db_session.query(Person).get(1)

데이터 수정

python

# 조회 후 수정
person = db_session.query(Person).get(1)
person.full_name = "홍길동2"
db_session.commit()

# 병렬 수정
db_session.query(Person).filter(Person.full_name.like("홍%")).update({"full_name": "홍씨"}, synchronize_session=False)
db_session.commit()

데이터 삭제

python

# 조회 후 삭제
person = db_session.query(Person).get(1)
db_session.delete(person)
db_session.commit()

# 병렬 삭제
db_session.query(Person).filter(Person.full_name == "김철수").delete(synchronize_session=False)
db_session.commit()

데이터 조회 기능

기본 쿼리

python

# 모든 기록 조회
all_records = db_session.query(Person).all()

# 특정 필드만 조회
names = db_session.query(Person.full_name).all()

# 정렬
sorted_records = db_session.query(Person).order_by(Person.full_name.desc()).all()

# 결과 제한
limited_results = db_session.query(Person).limit(10).all()

# 오프셋 설정
paged_results = db_session.query(Person).offset(5).limit(10).all()

필터링 쿼리

python

from sqlalchemy import or_

# 등가 필터
record = db_session.query(Person).filter(Person.full_name == "홍길동").first()

# 패턴 매칭
matching_records = db_session.query(Person).filter(Person.full_name.like("홍%")).all()

# IN 조건
selected_records = db_session.query(Person).filter(Person.full_name.in_(["홍길동", "김철수"])).all()

# 복합 조건
complex_conditions = db_session.query(Person).filter(
    Person.full_name == "홍길동", 
    Person.email_address.like("%example.com")
).all()

# OR 조건
or_conditions = db_session.query(Person).filter(
    or_(Person.full_name == "홍길동", Person.full_name == "김철수")
).all()

# 부등호 조건
not_equal_records = db_session.query(Person).filter(Person.full_name != "홍길동").all()

집계 쿼리

python

from sqlalchemy import func

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

# 그룹화 카운트
grouped_counts = db_session.query(
    Person.full_name, 
    func.count(Article.id)
).join(Article).group_by(Person.full_name).all()

# 합계, 평균 계산
average_id = db_session.query(func.avg(Person.id)).scalar()

조인 쿼리

python

# 내부 조인
join_results = db_session.query(Person, Article).join(Article).filter(Article.title.like("%Python%")).all()

# 외부 조인
outer_join_results = db_session.query(Person, Article).outerjoin(Article).all()

# 조인 조건 명시
custom_join = db_session.query(Person, Article).join(Article, Person.id == Article.writer_id).all()

관계 처리

python

# 관계 형성 객체 생성
person = Person(full_name="김민수", email_address="kim@example.com")
article = Article(title="처음 작성한 글", content="안녕하세요", writer=person)
db_session.add(article)
db_session.commit()

# 관계로 접근
print(f"글 '{article.title}'의 저자: {article.writer.full_name}")
print(f"{person.full_name}의 모든 글:")
for a in person.articles:
    print(f"  - {a.title}")

# N:N 관계 처리
python_tag = Tag(tag_name="Python")
sqlalchemy_tag = Tag(tag_name="SQLAlchemy")

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

print(f"글 '{article.title}'의 태그:")
for tag in article.tags:
    print(f"  - {tag.tag_name}")

트랜잭션 관리

python

# 자동 커밋
try:
    new_person = Person(full_name="테스트 사용자", email_address="test@example.com")
    db_session.add(new_person)
    db_session.commit()
except Exception as e:
    db_session.rollback()
    print(f"오류 발생: {e}")

# 트랜잭션 컨텍스트 관리
def add_user(session, name, email):
    try:
        user = Person(full_name=name, email_address=email)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise

# 중첩 트랜잭션
with db_session.begin_nested():
    user = Person(full_name="중첩 사용자", email_address="nested@example.com")
    db_session.add(user)

# 세이브포인트
savepoint = db_session.begin_nested()
try:
    user = Person(full_name="세이브포인트 사용자", email_address="savepoint@example.com")
    db_session.add(user)
    savepoint.commit()
except:
    savepoint.rollback()

최적화 팁

  1. 세션 관리: 요청별로 새로운 세션 생성 후 종료
  2. 예외 처리: 모든 트랜잭션에 롤백 로직 포함
  3. 지연 로딩: N+1 문제 주의, 즉시 로딩 사용 권장
  4. 커넥션 풀: 적절한 크기 및 타임아웃 설정
  5. 데이터 검증: 모델 레벨 또는 애플리케이션 레벨에서 검증

python

# 컨텍스트 매니저로 세션 관리
from contextlib import contextmanager

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

# 사용 예시
with get_db() as db:
    user = Person(full_name="컨텍스트 사용자", email_address="context@example.com")
    db.add(user)

태그: SQLAlchemy ORM Python 데이터베이스 연동 SQL 쿼리 최적화 객체 관계 매핑 데이터베이스 트랜잭션

6월 28일 00:38에 게시됨

인기 태그

©2026 괴물 클럽