SQLAlchemy ORM 사용법: 데이터베이스 작업 및 관리

SQLAlchemy은 Python에서 가장 널리 사용되는 ORM(객체-관계 매핑) 프레임워크로, 효율적이고 유연한 데이터베이스 조작 기능을 제공합니다. 본문에서는 SQLAlchemy ORM을 활용한 데이터베이스 작업 방법에 대해 설명합니다.

설치

pip install sqlalchemy

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

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

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

핵심 개념

  • 엔진: 데이터베이스 연결을 처리하는 통신 인터페이스
  • 세션: 데이터베이스 작업을 관리하는 컨텍스트
  • 모델: 데이터베이스 테이블과 매핑된 클래스 구조
  • 쿼리: 데이터베이스 조회를 위한 명령어 집합

데이터베이스 연결

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

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

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

데이터 모델 정의

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)

테이블 생성

Base.metadata.create_all(bind=engine)

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

기본 CRUD 작업

데이터 생성

새_사용자 = 사용자(이름="김철수", 이메일="kimcheolsu@example.com")
session.add(새_사용자)
session.commit()

session.add_all([
    사용자(이름="박영희", 이메일="parkyeonghee@example.com"),
    사용자(이름="최민호", 이메일="choiminho@example.com")
])
session.commit()

데이터 조회

사용자_목록 = session.query(사용자).all()
첫_사용자 = session.query(사용자).first()
특정_사용자 = session.query(사용자).get(1)

데이터 수정

사용자 = session.query(사용자).get(1)
사용자.이름 = "김철수"
session.commit()

session.query(사용자).filter(사용자.이름.like("김%")).update({"이름": "김씨"}, synchronize_session=False)
session.commit()

데이터 삭제

사용자 = session.query(사용자).get(1)
session.delete(사용자)
session.commit()

session.query(사용자).filter(사용자.이름 == "박영희").delete(synchronize_session=False)
session.commit()

데이터 조회

기본 쿼리

사용자_목록 = session.query(사용자).all()
이름_목록 = session.query(사용자.이름).all()
정렬된_사용자 = session.query(사용자).order_by(사용자.이름.desc()).all()
제한된_사용자 = session.query(사용자).limit(10).all()
오프셋_사용자 = session.query(사용자).offset(5).limit(10).all()

필터링 쿼리

from sqlalchemy import or_

등가_필터 = session.query(사용자).filter(사용자.이름 == "김철수").first()
부분_매칭 = session.query(사용자).filter(사용자.이름.like("김%")).all()
IN_검색 = session.query(사용자).filter(사용자.이름.in_(["김철수", "박영희"])).all()
다중_조건 = session.query(사용자).filter(
    사용자.이름 == "김철수", 
    사용자.이메일.like("%@example.com")
).all()

OR_조건 = session.query(사용자).filter(
    or_(사용자.이름 == "김철수", 사용자.이름 == "박영희")
).all()

불일치_필터 = session.query(사용자).filter(사용자.이름 != "김철수").all()

집계 쿼리

from sqlalchemy import func

총_수량 = session.query(사용자).count()
그룹_집계 = session.query(
    사용자.이름, 
    func.count(게시물.id)
).join(게시물).group_by(사용자.이름).all()

평균_값 = session.query(func.avg(사용자.id)).scalar()

JOIN 쿼리

내부_JOIN = session.query(사용자, 게시물).join(게시물).filter(게시물.제목.like("%Python%")).all()
외부_JOIN = session.query(사용자, 게시물).outerjoin(게시물).all()
조건_지정_JOIN = session.query(사용자, 게시물).join(게시물, 사용자.id == 게시물.작성자_id).all()

관계 조작

new_user = 사용자(이름="이순신", 이메일="lee_soonshin@example.com")
post = 게시물(제목="처음 글", 내용="안녕하세요!", 작성자=new_user)
session.add(post)
session.commit()

print(f"게시물 '{post.제목}'의 작성자는 {post.작성자.이름}")
print(f"사용자 {new_user.이름}의 모든 게시물:")
for p in new_user.게시물:
    print(f"  - {p.제목}")

python_tag = 태그(이름="Python")
sqlalchemy_tag = 태그(이름="SQLAlchemy")

post.태그.append(python_tag)
post.태그.append(sqlalchemy_tag)
session.commit()

print(f"게시물 '{post.제목}'의 태그:")
for tag in post.태그:
    print(f"  - {tag.이름}")

트랜잭션 관리

try:
    user = 사용자(이름="테스트 사용자", 이메일="test@example.com")
    session.add(user)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"오류 발생: {e}")

def 사용자_생성(session: Session, 이름: str, 이메일: str):
    try:
        user = 사용자(이름=이름, 이메일=이메일)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise

with session.begin_nested():
    user = 사용자(이름="트랜잭션 사용자", 이메일="transaction@example.com")
    session.add(user)

savepoint = session.begin_nested()
try:
    user = 사용자(이름="저장점 사용자", 이메일="savepoint@example.com")
    session.add(user)
    savepoint.commit()
except:
    savepoint.rollback()

최선의 실천 방법

  1. 세션 관리: 각 요청에 새 세션 생성 후 종료
  2. 예외 처리: 예외 발생 시 트랜잭션 롤백
  3. 지연 로딩: N+1 문제 방지하기 위해 eager loading 적용
  4. 커넥션 풀: 적절한 크기와 타임아웃 설정
  5. 데이터 검증: 모델 또는 애플리케이션 레벨에서 데이터 무결성 검사
@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

with get_db() as db:
    user = 사용자(이름="컨택스트 사용자", 이메일="context@example.com")
    db.add(user)

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

7월 16일 16:28에 게시됨

인기 태그

©2026 괴물 클럽