SQLAlchemy ORM 소개

SQLAlchemy는 Python에서 널리 사용되는 ORM(Object-Relational Mapping) 프레임워크로, 데이터베이스 작업을 효율적이고 유연하게 처리할 수 있습니다.

설치 방법

pip install sqlalchemy

데이터베이스별 드라이버 추가 설치:

# PostgreSQL
pip install psycopg2

# MySQL
pip install pymysql

# SQLite (별도 설치 불필요)

핵심 구성 요소

• Engine: 데이터베이스 연결 관리
• Session: 트랜잭션 및 영속성 제어
• Model: 데이터베이스 테이블 매핑 클래스
• Query: 데이터 조회 객체

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 연결 예제
db_engine = create_engine('sqlite:///app_data.db', echo=False)

# PostgreSQL 연결 예제
# db_engine = create_engine('postgresql://user:pass@localhost:5432/appdb')

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

# 세션 인스턴스화
db_session = DatabaseSession()

데이터 모델 정의

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

ModelBase = declarative_base()

class Customer(ModelBase):
    __tablename__ = 'customers'
    
    customer_id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(50), nullable=False)
    contact_email = Column(String(100), unique=True)
    
    orders = relationship("Order", back_populates="customer")

class Order(ModelBase):
    __tablename__ = 'orders'
    
    order_id = Column(Integer, primary_key=True)
    product_name = Column(String(100))
    quantity = Column(Integer)
    customer_ref = Column(Integer, ForeignKey('customers.customer_id'))
    
    customer = relationship("Customer", back_populates="orders")
    categories = relationship("Category", secondary="order_categories")

class Category(ModelBase):
    __tablename__ = 'categories'
    
    category_id = Column(Integer, primary_key=True)
    category_name = Column(String(30), unique=True)
    
    orders = relationship("Order", secondary="order_categories")

# 다대다 관계 연결 테이블
class OrderCategory(ModelBase):
    __tablename__ = 'order_categories'
    
    order_ref = Column(Integer, ForeignKey('orders.order_id'), primary_key=True)
    category_ref = Column(Integer, ForeignKey('categories.category_id'), primary_key=True)

테이블 생성

ModelBase.metadata.create_all(bind=db_engine)

CRUD 작업

데이터 생성

new_customer = Customer(full_name="김철수", contact_email="chulsoo@domain.com")
db_session.add(new_customer)
db_session.commit()

데이터 조회

customer = db_session.query(Customer).filter_by(full_name="김철수").first()

데이터 수정

customer.contact_email = "new_email@domain.com"
db_session.commit()

데이터 삭제

db_session.delete(customer)
db_session.commit()

고급 쿼리

from sqlalchemy import func, or_

# 조건 필터링
results = db_session.query(Customer).filter(
    or_(Customer.full_name.startswith("김"), 
        Customer.contact_email.contains("domain"))
).all()

# 집계 연산
order_count = db_session.query(
    Customer.full_name, 
    func.count(Order.order_id)
).join(Order).group_by(Customer.full_name).all()

관계형 데이터 작업

customer = Customer(full_name="이영희", contact_email="younghee@company.com")
new_order = Order(product_name="노트북", quantity=1, customer=customer)
db_session.add(new_order)
db_session.commit()

# 관계 접근
for order in customer.orders:
    print(f"{customer.full_name}의 주문: {order.product_name}")

트랜잭션 관리

try:
    new_data = Customer(full_name="트랜잭션 테스트", contact_email="tx_test@mail.com")
    db_session.add(new_data)
    db_session.commit()
except Exception:
    db_session.rollback()

권장 사항

• 요청 단위 세션 관리
• 예외 발생 시 롤백 처리
• N+1 문제 해결을 위한 적절한 로딩 전략
• 커넥션 풀 설정 최적화

from contextlib import contextmanager

@contextmanager
def db_session_scope():
    session = DatabaseSession()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

# 사용 예시
with db_session_scope() as session:
    session.add(Customer(full_name="홍길동", contact_email="gildong@hong.com"))