SQLAlchemy ORM 기반 데이터베이스 작업 가이드

파이썬에서 가장 널리 사용되는 ORM 프레임워크 중 하나인 SQLAlchemy는 데이터베이스 작업을 효율적이고 유연하게 수행할 수 있도록 지원합니다. 본 문서에서는 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

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/dbname')

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

# 세션 생성자 생성
SessionClass = sessionmaker(autocommit=False, autoflush=False, bind=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__ = 'persons'
    
    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_name = Column(String(100), nullable=False)
    content_text = Column(String(500))
    writer_id = Column(Integer, ForeignKey('persons.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=engine)

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

기본 CRUD 연산

데이터 삽입

python

# 새 사용자 추가
new_person = Person(full_name="김철수", email_address="kim@example.com")
db_session.add(new_person)
db_session.commit()

# 대량 삽입
db_session.add_all([
    Person(full_name="박영희", email_address="park@example.com"),
    Person(full_name="최민수", email_address="choi@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 = "김철수삼"
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

# 전체 조회
people = db_session.query(Person).all()

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

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

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

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

필터링 쿼리

python

from sqlalchemy import or_

# 정확한 값 검색
person = db_session.query(Person).filter(Person.full_name == "김철수").first()

# 패턴 검색
people = db_session.query(Person).filter(Person.full_name.like("김%")).all()

# IN 조건
people = db_session.query(Person).filter(Person.full_name.in_(["김철수", "박영희"])).all()

# 다중 조건
people = db_session.query(Person).filter(
    Person.full_name == "김철수", 
    Person.email_address.like("%example.com")
).all()

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

# 부등식 조건
people = db_session.query(Person).filter(Person.full_name != "김철수").all()

집계 쿼리

python

from sqlalchemy import func

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

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

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

조인 쿼리

python

# 내부 조인
results = db_session.query(Person, Article).join(Article).filter(Article.title.like("%파이썬%")).all()

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

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

관계 처리 예시

python

# 관계를 가지는 객체 생성
person = Person(full_name="이미자", email_address="lee@example.com")
article = Article(title_name="처음 작성한 글", content_text="안녕하세요", writer=person)
db_session.add(article)
db_session.commit()

# 관계를 통한 접근
print(f"글 '{article.title_name}'의 작성자는 {article.writer.full_name}")
print(f"{person.full_name}의 모든 글:")
for a in person.articles:
    print(f"  - {a.title_name}")

# 다대다 관계 처리
python_tag = Tag(tag_name="파이썬")
sqlalchemy_tag = Tag(tag_name="SQLAlchemy")

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

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

트랜잭션 관리

python

# 자동 커밋 트랜잭션
try:
    person = Person(full_name="테스트 사용자", email_address="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 create_person(session: Session, name: str, email: str):
    try:
        person = Person(full_name=name, email_address=email)
        session.add(person)
        session.commit()
        return person
    except:
        session.rollback()
        raise

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

# 세이브포인트 사용
savepoint = db_session.begin_nested()
try:
    person = Person(full_name="세이브포인트 사용자", email_address="savepoint@example.com")
    db_session.add(person)
    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:
    person = Person(full_name="컨텍스트 사용자", email_address="context@example.com")
    db.add(person)

요약

SQLAlchemy ORM은 강력하고 유연한 데이터베이스 운영 기능을 제공합니다. 본 문서를 통해 다음을 수행할 수 있습니다:

  1. SQLAlchemy 설치 및 구성
  2. 데이터 모델 및 관계 정의
  3. 기본 CRUD 연산 수행
  4. 복잡한 쿼리 작성
  5. 트랜잭션 관리
  6. 최선의 실천 방법 적용

SQLAlchemy는 혼합 속성, 이벤트 감지, 커스텀 쿼리 등 고급 기능도 제공하므로 추가 탐구가 권장됩니다.

태그: sqlalchemy ORM python 데이터베이스 쿼리

7월 6일 19:55에 게시됨

인기 태그

©2026 괴물 클럽