SQLAlchemy ORM 데이터베이스 작업 완벽 가이드

SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(객체 관계 매핑) 프레임워크 중 하나로, 효율적이고 유연한 데이터베이스 작업을 제공합니다. 본문에서는 SQLAlchemy ORM을 사용하여 데이터베이스 작업을 수행하는 방법을 자세히 설명합니다.

목차

  1. SQLAlchemy 설치
  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 (Python 표준 라이브러리에 포함되어 있어 별도 설치 불필요)

핵심 개념

  • Engine: 데이터베이스 연결 엔진으로, 데이터베이스와의 통신을 담당합니다
  • Session: 데이터베이스 세션으로, 모든 지속화 작업을 관리합니다
  • Model: 데이터 모델 클래스로, 데이터베이스의 테이블에 해당합니다
  • Query: 쿼리 객체로, 데이터베이스 쿼리 구축 및 실행에 사용됩니다

데이터베이스 연결

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 데이터베이스 연결 엔진 생성
# SQLite 예시
engine = create_engine('sqlite:///sample.db', echo=True)

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

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

# 세션 팩토리 생성
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

데이터 모델 정의

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

# 기본 클래스 생성
DataModel = declarative_base()

class Person(DataModel):
    __tablename__ = 'persons'
    
    id = Column(Integer, primary_key=True, index=True)
    fullname = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # 일대다 관계 정의
    articles = relationship("Article", back_populates="writer")
    
class Article(DataModel):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True, index=True)
    heading = Column(String(100), nullable=False)
    content = Column(String(500))
    writer_id = Column(Integer, ForeignKey('persons.id'))
    
    # 다대일 관계 정의
    writer = relationship("Person", back_populates="articles")
    
    # 다대다 관계 정의 (연결 테이블을 통해)
    categories = relationship("Category", secondary="article_categories", back_populates="articles")

class Category(DataModel):
    __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(DataModel):
    __tablename__ = 'article_categories'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)

데이터베이스 테이블 생성

# 모든 테이블 생성
DataModel.metadata.create_all(bind=engine)

# 모든 테이블 삭제
# DataModel.metadata.drop_all(bind=engine)

기본 CRUD 작업

데이터 생성

# 새로운 사용자 생성
new_person = Person(fullname="김철수", email="chulsoo@example.com")
db_session.add(new_person)
db_session.commit()

# 일괄 생성
db_session.add_all([
    Person(fullname="이영희", email="younghee@example.com"),
    Person(fullname="박민준", email="minjun@example.com")
])
db_session.commit()

데이터 읽기

# 모든 사용자 조회
people = db_session.query(Person).all()

# 첫 번째 사용자 조회
first_person = db_session.query(Person).first()

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

데이터 업데이트

# 조회 후 업데이트
person = db_session.query(Person).get(1)
person.fullname = "김철수 modified"
db_session.commit()

# 일괄 업데이트
db_session.query(Person).filter(Person.fullname.like("김%")).update({"fullname": "김씨"}, synchronize_session=False)
db_session.commit()

데이터 삭제

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

# 일괄 삭제
db_session.query(Person).filter(Person.fullname == "이영희").delete(synchronize_session=False)
db_session.commit()

데이터 조회

기본 조회

# 모든 레코드 조회
people = db_session.query(Person).all()

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

# 정렬
people = db_session.query(Person).order_by(Person.fullname.desc()).all()

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

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

필터링 조회

from sqlalchemy import or_

# 동등 필터링
person = db_session.query(Person).filter(Person.fullname == "김철수").first()

# 패턴 매칭
people = db_session.query(Person).filter(Person.fullname.like("김%")).all()

# IN 쿼리
people = db_session.query(Person).filter(Person.fullname.in_(["김철수", "이영희"])).all()

# 다중 조건 쿼리
people = db_session.query(Person).filter(
    Person.fullname == "김철수", 
    Person.email.like("%@example.com")
).all()

# OR 조건
people = db_session.query(Person).filter(
    or_(Person.fullname == "김철수", Person.fullname == "이영희")
).all()

# 같지 않음
people = db_session.query(Person).filter(Person.fullname != "김철수").all()

집계 조회

from sqlalchemy import func

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

# 그룹별 카운트
person_article_count = db_session.query(
    Person.fullname, 
    func.count(Article.id)
).join(Article).group_by(Person.fullname).all()

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

조인 조회

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

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

# 조인 조건 지정
results = db_session.query(Person, Article).join(Article, Person.id == Article.writer_id).all()

관계형 작업

# 관계가 있는 객체 생성
writer = Person(fullname="조현우", email="hyunwoo@example.com")
article = Article(heading="첫 번째 블로그 글", content="안녕하세요, 세계!", writer=writer)
db_session.add(article)
db_session.commit()

# 관계를 통해 접근
print(f"'{article.heading}' 글의 작성자: {article.writer.fullname}")
print(f"{writer.fullname}의 모든 글:")
for a in writer.articles:
    print(f"  - {a.heading}")

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

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

print(f"'{article.heading}' 글의 카테고리:")
for category in article.categories:
    print(f"  - {category.name}")

트랜잭션 관리

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

# 트랜잭션 컨텍스트 관리기 사용
from sqlalchemy.orm import Session

def add_person(session: Session, fullname: str, email: str):
    try:
        person = Person(fullname=fullname, email=email)
        session.add(person)
        session.commit()
        return person
    except:
        session.rollback()
        raise

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

# 저장점
savepoint = db_session.begin_nested()
try:
    person = Person(fullname="저장점 사용자", email="savepoint@example.com")
    db_session.add(person)
    savepoint.commit()
except:
    savepoint.rollback()

모범 사례

  1. 세션 관리: 각 요청에 대해 새 세션을 생성하고 요청 종료 후 닫습니다
  2. 예외 처리: 항상 예외를 처리하고 적절하게 트랜잭션을 롤백합니다
  3. 지연 로딩: N+1 쿼리 문제에 주의하고, eager loading으로 최적화합니다
  4. 연결 풀: 연결 풀 크기 및 시간 초과 설정을 합리적으로 구성합니다
  5. 데이터 검증: 모델 레이어 또는 애플리케이션 레이어에서 데이터 무결성을 검증합니다
# 컨텍스트 관리기를 사용한 세션 관리
from contextlib import contextmanager

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

# 사용 예시
with get_database_session() as db:
    person = Person(fullname="컨텍스트 사용자", email="context@example.com")
    db.add(person)

결론

SQLAlchemy ORM은 강력하고 유연한 데이터베이스 작업 방식을 제공하며, 본 가이드를 통해 다음을 할 수 있게 되었습니다:

  1. SQLAlchemy 설치 및 구성
  2. 데이터 모델과 관계 정의
  3. 기본 CRUD 작업 수행
  4. 복잡한 쿼리 구축
  5. 데이터베이스 트랜잭션 관리
  6. 모범 사례 따르기

SQLAlchemy에는 혼합 속성, 이벤트 리스너, 사용자 정의 쿼리 등 더 많은 고급 기능이 있으며, 추가 탐구 학습이 가치가 있습니다.

태그: sqlalchemy ORM python 데이터베이스 CRUD

7월 1일 02:06에 게시됨

인기 태그

©2026 괴물 클럽