Python과 SQLAlchemy를 활용한 데이터베이스 ORM 실전 가이드

시작하며

SQLAlchemy는 파이썬에서 가장 널리 사용되는 ORM(Object-Relational Mapping) 프레임워크로서, 데이터베이스 작업을 효율적이고 유연하게 처리할 수 있게 해줍니다. 이 가이드에서는 SQLAlchemy ORM을 사용하여 데이터베이스를 다루는 핵심 방법들을 살펴보겠습니다.

목차

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

1. SQLAlchemy 설치

가장 먼저 pip를 사용하여 SQLAlchemy를 설치합니다.

pip install sqlalchemy

연결하려는 데이터베이스 종류에 따라 적절한 드라이버를 추가로 설치해야 합니다.

# PostgreSQL 연결 시
pip install psycopg2-binary

# MySQL 연결 시
pip install mysql-connector-python

# SQLite의 경우 파이썬 표준 라이브러리에 포함되어 있음

2. 핵심 개념 이해

SQLAlchemy의 아키텍처를 이해하기 위해 핵심 구성요소를 살펴보겠습니다:

  • Engine: 데이터베이스와 통신을 담당하는 연결 엔진
  • Session: 영속성 연산을 관리하는 데이터베이스 세션
  • Model: 데이터베이스 테이블에 대응하는 파이썬 클래스
  • Query: 데이터베이스 查询을 구성하고 실행하는 객체

3. 데이터베이스 연결

데이터베이스 연결을 설정하는 방법을 알아봅시다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 연결 엔진 생성
engine = create_engine('sqlite:///example.db', echo=True)

# PostgreSQL 연결 예시
# engine = create_engine('postgresql://username:password@localhost:5432/mydatabase')

# MySQL 연결 예시
# engine = create_engine('mysql+mysqlconnector://username:password@localhost:3306/mydatabase')

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

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

4. 데이터 모델 정의

테이블 구조를 파이썬 클래스로 정의하는 방법을 보여줍니다.

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

# 베이스 클래스 생성
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # 일대다 관계 정의
    posts = relationship("Post", back_populates="author")
    
class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(String(500))
    author_id = Column(Integer, ForeignKey('users.id'))
    
    # 다대일 관계 정의
    author = relationship("User", back_populates="posts")
    
    # 다대다 관계 정의 (중개 테이블 사용)
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(30), unique=True, nullable=False)
    
    posts = relationship("Post", secondary="post_tags", back_populates="tags")

# 다대다 관계를 위한 중개 테이블
class PostTag(Base):
    __tablename__ = 'post_tags'
    
    post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

5. 테이블 생성

정의된 모델을 바탕으로 실제 데이터베이스 테이블을 생성합니다.

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

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

6. 기본 CRUD 연산

데이터 생성

# 단일 사용자 생성
new_user = User(name="홍길동", email="hong@example.com")
session.add(new_user)
session.commit()

# 여러 사용자批量 생성
session.add_all([
    User(name="김철수", email="kim@example.com"),
    User(name="이영희", email="lee@example.com")
])
session.commit()

데이터 조회

# 모든 사용자 조회
users = session.query(User).all()

# 첫 번째 사용자 조회
first_user = session.query(User).first()

# ID로 특정 사용자 조회
user = session.query(User).get(1)

데이터 수정

# 조회 후 수정
user = session.query(User).get(1)
user.name = "홍길동수정"
session.commit()

#批量 수정
session.query(User).filter(User.name.like("홍%")).update({"name": "홍氏"}, synchronize_session=False)
session.commit()

데이터 삭제

# 조회 후 삭제
user = session.query(User).get(1)
session.delete(user)
session.commit()

#批量 삭제
session.query(User).filter(User.name == "김철수").delete(synchronize_session=False)
session.commit()

7. 데이터 조회 심화

기본 조회

# 전체 레코드 조회
users = session.query(User).all()

# 특정 필드만 조회
names = session.query(User.name).all()

# 정렬
users = session.query(User).order_by(User.name.desc()).all()

# 결과 수 제한
users = session.query(User).limit(10).all()

# 오프셋 지정
users = session.query(User).offset(5).limit(10).all()

필터 조회

from sqlalchemy import or_

# 정확히 일치하는 조건
user = session.query(User).filter(User.name == "홍길동").first()

# 유사 검색
users = session.query(User).filter(User.name.like("홍%")).all()

# IN 조건
users = session.query(User).filter(User.name.in_(["홍길동", "김철수"])).all()

#复合 조건
users = session.query(User).filter(
    User.name == "홍길동", 
    User.email.like("%@example.com")
).all()

# OR 조건
users = session.query(User).filter(
    or_(User.name == "홍길동", User.name == "김철수")
).all()

# 같지 않음
users = session.query(User).filter(User.name != "홍길동").all()

집계 조회

from sqlalchemy import func

# 레코드 수 계산
count = session.query(User).count()

# 그룹별 카운트
user_post_count = session.query(
    User.name, 
    func.count(Post.id)
).join(Post).group_by(User.name).all()

# 평균값 조회
avg_id = session.query(func.avg(User.id)).scalar()

조인 조회

# 내부 조인
results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all()

# 왼쪽 외부 조인
results = session.query(User, Post).outerjoin(Post).all()

# 조인 조건 명시
results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

8. 관계형 데이터 처리

# 관계를 가진 객체 생성
user = User(name="박지민", email="park@example.com")
post = Post(title="내 첫 번째 글", content="Hello World!", author=user)
session.add(post)
session.commit()

# 관계를 통해 데이터 접근
print(f"글 '{post.title}'의 작성자: {post.author.name}")
print(f"사용자 {user.name}의 모든 글:")
for p in user.posts:
    print(f"  - {p.title}")

# 다대다 관계 조작
python_tag = Tag(name="Python")
sqlalchemy_tag = Tag(name="SQLAlchemy")

post.tags.append(python_tag)
post.tags.append(sqlalchemy_tag)
session.commit()

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

9. 트랜잭션 관리

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

# 컨텍스트 매니저를 사용한 트랜잭션
from sqlalchemy.orm import Session

def create_user(session: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise

# 중첩 트랜잭션
with session.begin_nested():
    user = User(name="트랜잭션사용자", email="transaction@example.com")
    session.add(user)

# 세이브포인트
savepoint = session.begin_nested()
try:
    user = User(name="세이브포인트사용자", email="savepoint@example.com")
    session.add(user)
    savepoint.commit()
except:
    savepoint.rollback()

10. 최적화 기법

효율적인 데이터베이스 연산을 위한 모범 사례를 알아봅시다:

  1. 세션 관리: 각 요청마다 새 세션을 생성하고 완료 후 반드시 닫아야 합니다
  2. 예외 처리: 예외 상황을 적절히 처리하고 트랜잭션을 롤백해야 합니다
  3. 지연 로딩 주의: N+1 查询 문제를 인식하고 eager loading으로 최적화해야 합니다
  4. 연결 풀: 연결 풀 크기와超时 시간을 적절히 설정해야 합니다
  5. 데이터 검증: 모델 계층 또는 애플리케이션 계층에서 데이터 무결성을 검증해야 합니다
# 컨텍스트 매니저로 세션 관리
from contextlib import contextmanager

@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 = User(name="컨텍스트사용자", email="context@example.com")
    db.add(user)

마치며

SQLAlchemy ORM은 강력하고 유연한 데이터베이스 操作 수단을 제공합니다. 이 가이드의 내용을 통해 다음을 수행할 수 있게 되었습니다:

  1. SQLAlchemy 설치 및 설정
  2. 데이터 모델과 관계 정의
  3. 기본 CRUD 연산 수행
  4. 복잡한 查询 구성
  5. 데이터베이스 트랜잭션 관리
  6. 최적화 기법 적용

SQLAlchemy에는 혼합 속성, 이벤트 리스너, 사용자 정의 查询 등 더 많은 고급 기능이 있으며, 이를 더욱 탐구해 보시기 바랍니다.

태그: python sqlalchemy ORM Database PostgreSQL

7월 16일 06:07에 게시됨

인기 태그

©2026 괴물 클럽