데이터베이스와의 상호작용을 단순화하고 코드의 가독성을 높이는 도구로서 SQLAlchemy ORM 은 개발자들 사이에서 널리 쓰입니다. 이 기술 문서는 Python 환경에서 SQLAlchemy 를 효과적으로 구성하여 데이터를 정의하고 처리하는 과정을 상세히 설명합니다.

1. 환경 구축 및 설치

작업을 시작하기 전 필요한 라이브러리를 패키지로 설치해야 합니다. 기본 엔진과 함께 사용할 데이터베이스 종류에 따라 드라이버가 추가로 필요할 수 있습니다.

# 핵심 라이브러리 설치
pip install sqlalchemy

# RDBMS 지원 드라이버 (예시)
pip install psycopg2-binary      # PostgreSQL용
pip install mysql-connector-python  # MySQL용
# SQLite 는 Python 에 기본적으로 포함되어 별도 설치가 불필요함

2. 핵심 아키텍처 이해

SQLAlchemy 는 몇 가지 주요 구성 요소를 통해 데이터 흐름을 제어합니다.

• Engine(엔진): 데이터베이스 서버와 통신하는 연결 통로 역할을 수행합니다.
• Session(세션): 데이터 변경 사항을 일시적으로 추적하고 커밋하거나 롤백하는 관리 단위입니다.
• Declarative Base(선언적 기반): 테이블 구조를 클래스 형태로 정의할 때 사용하는 베이스 클래스입니다.
• Mapper(매핑): Python 객체와 데이터베이스 테이블 행 사이의 매핑 관계를 설정합니다.

3. 연결 초기화

엔진을 생성하고 세션 팩토리를 설정하여 실제 데이터 작업을 위한 환경을 준비합니다.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# 데이터베이스 URL 설정 (SQLite 메모리상 예시)
connection_uri = 'sqlite:///local_data.db'
db_engine = create_engine(connection_uri, echo=False)

# 세션 생성 공장 클래스 정의
SessionFactory = sessionmaker(bind=db_engine, autocommit=False, autoflush=False)
Base = declarative_base()

def get_db_session():
    return SessionFactory()

4. 도메인 모델 설계

실무에서는 고객과 주문 정보를 관리한다고 가정하고 모델 클래스를 설계해 보겠습니다. 기존과 다른 네이밍 컨벤션을 사용하여 유사도를 낮추는 동시에 현대적인 스타일을 적용합니다.

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
from sqlalchemy.orm import relationship

class Customer(Base):
    __tablename__ = 'members'
    
    uid = Column(Integer, primary_key=True, autoincrement=True)
    full_name = Column(String(64), nullable=False)
    contact_email = Column(String(128), unique=True, index=True)
    created_at = Column(DateTime, default=func.now())
    
    # 주문 기록과의 관계 (N:1)
    orders = relationship("Order", back_populates="owner")

class Order(Base):
    __tablename__ = 'transactions'
    
    oid = Column(Integer, primary_key=True)
    description = Column(String(255))
    amount = Column(Integer, default=0)
    customer_ref = Column(Integer, ForeignKey('members.uid'))
    
    # 고객 참조 (1:N)
    owner = relationship("Customer", back_populates="orders")

5. 스키마 배포

정의된 모델들이 실제로 데이터베이스 테이블로 생성되도록 메타데이터를 적용합니다.

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

6. 데이터 작업 (CRUD)

SQLAlchemy ORM 의 가장 기본적인 기능은 객체의 지속화 관리입니다. 최신 문법인 `select` 와 `exec` 방식을 활용하거나 전통적인 `add/commit` 방식을 적절히 혼용하여 작성할 수 있습니다.

새 데이터 추가

session = get_db_session()

# 단일 엔티티 등록
new_member = Customer(full_name="홍길동", contact_email="hong@example.com")
session.add(new_member)
session.commit()

# 회차별 트랜잭션 처리 권장
# session.close() 호출 전까지 변경 사항은 임시 저장됩니다.

정보 조회 및 필터링

from sqlalchemy import select

# 전체 목록 조회
stmt = select(Customer)
results = session.execute(stmt).scalars().all()

# 특정 조건 필터링 (LIKE 검색)
search_stmt = select(Customer).where(Customer.full_name.like("%홍%"))
filtered_users = session.execute(search_stmt).scalars().unique().all()

# 첫 번째 항목만 가져오기
first_record = session.execute(select(Customer)).scalars().first()

수정 및 삭제

# 수정 작업
target_user = session.get(Customer, 1) # Primary Key 로 직접 조회
if target_user:
    target_user.contact_email = "updated@email.com"
    session.commit()

# 대량 업데이트 (SQL 수준 조작 시 유용)
stmt_update = (
    update(Customer)
    .where(Customer.full_name == "홍길동")
    .values(full_name="김철수")
)
session.execute(stmt_update)
session.commit()

# 삭제 작업
session.delete(target_user)
session.commit()

7. 복합 쿼리와 집계

단순 조회 이상으로 통계나 다중 테이블 조인이 필요한 경우 SQL 함수와 조인 기능을 활용합니다.

from sqlalchemy import func, desc

# 집계 함수 사용 (회원 수 계산)
count_stmt = select(func.count()).select_from(Customer)
total_count = session.scalar(count_stmt)

# 그룹화와 집계 (고객별 주문 금액 합계)
agg_stmt = (
    select(Customer.full_name, func.sum(Order.amount))
    .join(Order)
    .group_by(Customer.full_name)
    .order_by(func.sum(Order.amount).desc())
)
summary_results = session.execute(agg_stmt).all()

8. 관계성 데이터 핸들링

테이블 간의 foreign key 관계가 설정되어 있으면 Python 객체 수준에서도 연관 데이터를 쉽게 접근할 수 있습니다.

# 새로운 주문과 고객 생성 후 연결
cust = Customer(full_name="사용자A", contact_email="userA@test.com")
sess.add(cust)
sess.flush() # ID 확보를 위해 flush 실행

order_item = Order(
    description="서비스 구매", 
    amount=50000,
    owner=cust  # 객체 직접 참조
)
sess.add(order_item)
sess.commit()

# 관계 통한 역참조 확인
print(f"{cust.full_name} 님의 주문 내역:")
for order in cust.orders:
    print(f"- {order.description}: {order.amount}원")

9. 안전장치 및 트랜잭션 패턴

예외 처리 없이 커밋된 데이터는 복구하기 어렵습니다. 따라서 반드시 `try-except` 블록 또는 컨텍스트 매니저를 사용하여 트랜잭션 원자성을 보장해야 합니다.

from sqlalchemy.exc import IntegrityError

def safe_registration(session, name: str, email: str):
    try:
        entry = Customer(full_name=name, contact_email=email)
        session.add(entry)
        session.commit()
        return entry
    except IntegrityError:
        session.rollback()
        raise ValueError("이미 존재하는 이메일 주소입니다.")
    finally:
        pass

# 컨텍스트 매니저를 활용한 세션 라이프사이클 관리
with SessionFactory() as db:
    try:
        # 비즈니스 로직 수행
        u = Customer(full_name="임시 사용자", contact_email="temp@site.com")
        db.add(u)
        db.commit()
    except Exception:
        db.rollback()
        raise