Python SQLAlchemy ORM 활용 가이드

SQLAlchemy는 Python에서 가장 널리 사용되는 ORM(Object-Relational Mapping) 라이브러리 중 하나로, 데이터베이스 작업을 효율적이고 유연하게 수행할 수 있도록 지원합니다. 이 문서에서는 SQLAlchemy ORM을 사용하여 데이터베이스를 조작하는 방법에 대해 설명합니다.

설치

SQLAlchemy를 설치하려면 pip를 사용합니다.


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 예시
db_engine = create_engine('sqlite:///my_database.db', echo=True)

# PostgreSQL 예시
# db_engine = create_engine('postgresql://user:password@host:port/dbname')

# MySQL 예시
# db_engine = create_engine('mysql+mysqlconnector://user:password@host:port/dbname')

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

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

데이터 모델 정의

데이터베이스 테이블 구조를 Python 클래스로 정의합니다.


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

# 모델의 기반 클래스
BaseModel = declarative_base()

class Customer(BaseModel):
    __tablename__ = 'customers'

    customer_id = Column('id', Integer, primary_key=True, index=True)
    full_name = Column(String(100), nullable=False)
    email_address = Column(String(150), unique=True, index=True)

    # 고객과 주문 간의 일대다 관계 정의
    orders = relationship("Order", back_populates="customer")

class Product(BaseModel):
    __tablename__ = 'products'

    product_id = Column('id', Integer, primary_key=True, index=True)
    product_name = Column(String(120), nullable=False)
    price = Column(Integer) # 가격은 정수로 표현 (예: 센트 단위)

class Order(BaseModel):
    __tablename__ = 'orders'

    order_id = Column('id', Integer, primary_key=True, index=True)
    order_date = Column(String) # 날짜 형식은 필요에 따라 조정
    customer_id = Column(Integer, ForeignKey('customers.id'))

    # 주문과 고객 간의 다대일 관계 정의
    customer = relationship("Customer", back_populates="orders")

    # 주문과 상품 간의 다대다 관계 (Order_Item을 통한 연결)
    order_items = relationship("OrderItem", back_populates="order")

class OrderItem(BaseModel):
    __tablename__ = 'order_items'

    item_id = Column('id', Integer, primary_key=True, index=True)
    order_id = Column(Integer, ForeignKey('orders.id'), primary_key=True)
    product_id = Column(Integer, ForeignKey('products.id'), primary_key=True)
    quantity = Column(Integer)

    order = relationship("Order", back_populates="order_items")
    product = relationship("Product")

데이터베이스 테이블 생성

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


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

# 모든 테이블 삭제 (주의해서 사용)
# BaseModel.metadata.drop_all(bind=db_engine)

기본 CRUD 작업

데이터 생성 (Create)


# 새 고객 추가
new_customer = Customer(full_name="김철수", email_address="chulsoo.kim@example.com")
db_session.add(new_customer)
db_session.commit()

# 여러 레코드 동시 추가
db_session.add_all([
    Customer(full_name="박영희", email_address="younghee.park@example.com"),
    Customer(full_name="이민준", email_address="minjun.lee@example.com")
])
db_session.commit()

데이터 읽기 (Read)


# 모든 고객 조회
all_customers = db_session.query(Customer).all()

# 첫 번째 고객 조회
first_customer = db_session.query(Customer).first()

# ID로 특정 고객 조회
customer_by_id = db_session.query(Customer).get(1)

데이터 수정 (Update)


# ID가 1인 고객의 이메일 주소 변경
target_customer = db_session.query(Customer).get(1)
if target_customer:
    target_customer.email_address = "new.email@example.com"
    db_session.commit()

# 특정 조건을 만족하는 고객 이름 일괄 변경
db_session.query(Customer).filter(Customer.full_name.like("박%")).update({"full_name": "박OO"}, synchronize_session=False)
db_session.commit()

데이터 삭제 (Delete)


# ID가 1인 고객 삭제
customer_to_delete = db_session.query(Customer).get(1)
if customer_to_delete:
    db_session.delete(customer_to_delete)
    db_session.commit()

# 이름이 "이민준"인 고객 일괄 삭제
db_session.query(Customer).filter(Customer.full_name == "이민준").delete(synchronize_session=False)
db_session.commit()

데이터 조회 (Query)

기본 조회


# 모든 레코드 조회
customers = db_session.query(Customer).all()

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

# 이름 기준 내림차순 정렬
sorted_customers = db_session.query(Customer).order_by(Customer.full_name.desc()).all()

# 결과 수 제한
limited_customers = db_session.query(Customer).limit(5).all()

# 특정 범위의 결과 조회 (페이징)
paged_customers = db_session.query(Customer).offset(10).limit(5).all()

필터링 조건


from sqlalchemy import or_

# 이름이 "김철수"인 고객 조회
customer = db_session.query(Customer).filter(Customer.full_name == "김철수").first()

# 이름이 "김"으로 시작하는 고객 조회
customers = db_session.query(Customer).filter(Customer.full_name.like("김%")).all()

# 이름이 특정 목록에 포함된 고객 조회
customers = db_session.query(Customer).filter(Customer.full_name.in_(["김철수", "박영희"])).all()

# 복합 조건 (AND)
customers = db_session.query(Customer).filter(
    Customer.full_name == "김철수",
    Customer.email_address.like("%@example.com")
).all()

# 복합 조건 (OR)
customers = db_session.query(Customer).filter(
    or_(Customer.full_name == "김철수", Customer.full_name == "이민준")
).all()

# 같지 않음
customers = db_session.query(Customer).filter(Customer.full_name != "김철수").all()

집계 함수 사용


from sqlalchemy import func

# 고객 수 계산
customer_count = db_session.query(Customer).count()

# 고객별 주문 수 계산 (Join 및 Group By 사용)
customer_order_counts = db_session.query(
    Customer.full_name,
    func.count(Order.order_id)
).join(Order, Customer.customer_id == Order.customer_id, isouter=True).group_by(Customer.full_name).all()

# 평균 ID 값 계산
avg_customer_id = db_session.query(func.avg(Customer.customer_id)).scalar()

JOIN을 사용한 조회


# 고객과 주문 정보 INNER JOIN
results = db_session.query(Customer, Order).join(Order).filter(Order.order_date.like("2023-%")).all()

# 고객과 주문 정보 LEFT OUTER JOIN
results = db_session.query(Customer, Order).outerjoin(Order).all()

# 명시적 JOIN 조건 사용
results = db_session.query(Customer, Order).join(Order, Customer.customer_id == Order.customer_id).all()

관계(Relationship)를 활용한 작업

모델 간의 관계를 설정하고 활용합니다.


# 관계를 가진 객체 생성
customer = Customer(full_name="유관순", email_address="yks@example.com")
order = Order(order_date="2024-01-15", customer=customer) # customer 객체를 직접 할당
db_session.add(order)
db_session.commit()

# 관계를 통해 접근
print(f"주문 ID {order.order_id}의 고객: {order.customer.full_name}")
print(f"고객 {customer.full_name}의 주문 목록:")
for o in customer.orders:
    print(f"  - 주문 ID: {o.order_id}, 날짜: {o.order_date}")

# 다대다 관계 (OrderItem을 통해)
# Product와 OrderItem, Order 간의 관계 설정 필요 (위 모델 정의 참고)
# 예시: 특정 주문에 상품 추가
product_a = Product(product_name="노트북", price=120000)
order_item_a = OrderItem(order=order, product=product_a, quantity=1)
db_session.add(order_item_a)
db_session.commit()

# 주문에 포함된 상품 조회
print(f"주문 ID {order.order_id}에 포함된 상품:")
for item in order.order_items:
    print(f"  - 상품: {item.product.product_name}, 수량: {item.quantity}")

트랜잭션 관리

데이터베이스 트랜잭션을 안전하게 관리합니다.


# 자동 커밋/롤백 (try-except 구문 활용)
try:
    new_customer = Customer(full_name="임시사용자", email_address="temp@example.com")
    db_session.add(new_customer)
    db_session.commit() # 성공 시 커밋
except Exception as e:
    db_session.rollback() # 오류 발생 시 롤백
    print(f"트랜잭션 오류 발생: {e}")

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

def add_new_customer(db: Session, name: str, email: str):
    try:
        customer = Customer(full_name=name, email_address=email)
        db.add(customer)
        db.commit()
        return customer
    except:
        db.rollback()
        raise

# 중첩 트랜잭션
with db_session.begin_nested():
    customer1 = Customer(full_name="중첩1", email_address="nested1@example.com")
    db_session.add(customer1)
    # 필요 시 여기서 추가 작업 수행

# 세이브 포인트 (Savepoint)
save_point = db_session.begin_nested()
try:
    customer2 = Customer(full_name="세이브포인트", email_address="savepoint@example.com")
    db_session.add(customer2)
    save_point.commit() # 세이브 포인트 커밋
except:
    save_point.rollback() # 세이브 포인트 롤백

모범 사례

  1. 세션 관리: 각 요청마다 새로운 세션을 생성하고, 요청 완료 후 반드시 세션을 닫습니다.
  2. 예외 처리: 모든 데이터베이스 작업에서 발생할 수 있는 예외를 처리하고, 필요시 트랜잭션을 롤백합니다.
  3. 지연 로딩 최적화: N+1 쿼리 문제를 방지하기 위해 `joinedload` 또는 `selectinload`와 같은 Eager Loading 기법을 적절히 사용합니다.
  4. 연결 풀링: 데이터베이스 연결 풀의 크기와 타임아웃 설정을 최적화하여 성능을 향상시킵니다.
  5. 데이터 유효성 검사: 모델 정의 단계 또는 애플리케이션 로직에서 데이터의 유효성을 검사합니다.

컨텍스트 관리자를 활용한 세션 관리 예시:


from contextlib import contextmanager

@contextmanager
def manage_db_session():
    session = SessionFactory()
    try:
        yield session
        session.commit() # 성공 시 커밋
    except Exception:
        session.rollback() # 오류 시 롤백
        raise
    finally:
        session.close() # 세션 종료

# 사용 예시
with manage_db_session() as current_session:
    new_customer = Customer(full_name="컨텍스트사용자", email_address="context@example.com")
    current_session.add(new_customer)

SQLAlchemy ORM은 강력하고 유연한 데이터베이스 작업 기능을 제공합니다. 이 가이드에서는 기본적인 설치, 모델 정의, CRUD 작업, 쿼리 작성, 관계 설정 및 트랜잭션 관리 방법을 다루었습니다.

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

7월 24일 06:27에 게시됨

인기 태그

©2026 괴물 클럽