SQLAlchemy를 활용한 데이터베이스 관리

SQLAlchemy는 파이썬에서 가장 널리 사용되는 객체 관계 매핑(ORM) 도구로, 데이터베이스 작업을 객체 지향 방식으로 간편하게 수행할 수 있도록 지원합니다. 이 문서에서는 데이터 분석 시나리오에 최적화된 주요 기능과 실용적인 예제를 중심으로 설명합니다.

설치 및 환경 구성

pip install sqlalchemy

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

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (기본 라이브러리 포함)
# 별도 설치 불필요

핵심 구성 요소

• 엔진(Engine): 데이터베이스 연결을 담당하는 중앙 통제자
• 세션(Session): 데이터 변경 작업의 단위를 관리
• 모델(Model): 테이블 구조를 표현하는 클래스 정의
• 쿼리(Query): SQL 문장을 추상화하여 조건 기반 검색 제공

데이터베이스 연결 설정

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# SQLite 사용 예시
engine = create_engine('sqlite:///analytics.db', echo=True)

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

# 실제 세션 인스턴스 생성
db_session = SessionManager()

데이터 모델 정의

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

Base = declarative_base()

class Employee(Base):
    __tablename__ = 'employees'

    emp_id = Column(Integer, primary_key=True)
    full_name = Column(String(80), nullable=False)
    department = Column(String(50))
    email = Column(String(100), unique=True)

    # 다수의 프로젝트 참조
    projects = relationship("Project", back_populates="lead")

class Project(Base):
    __tablename__ = 'projects'

    project_id = Column(Integer, primary_key=True)
    title = Column(String(120), nullable=False)
    start_date = Column(String(20))
    lead_id = Column(Integer, ForeignKey('employees.emp_id'))

    # 리더 정보 역참조
    lead = relationship("Employee", back_populates="projects")

    # 태그 연결 (다대다 관계)
    tags = relationship("Tag", secondary="project_tags", back_populates="projects")

class Tag(Base):
    __tablename__ = 'tags'

    tag_id = Column(Integer, primary_key=True)
    label = Column(String(40), unique=True, nullable=False)

    projects = relationship("Project", secondary="project_tags", back_populates="tags")

# 다대다 관계 맵핑 테이블
class ProjectTag(Base):
    __tablename__ = 'project_tags'

    project_id = Column(Integer, ForeignKey('projects.project_id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tags.tag_id'), primary_key=True)

테이블 생성 및 관리

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

# 삭제 시 (주의: 데이터 손실 가능)
# Base.metadata.drop_all(bind=engine)

CRUD 기본 연산

생성 (Create)

# 단일 삽입
new_emp = Employee(full_name="김철수", department="개발팀", email="kim@example.com")
db_session.add(new_emp)
db_session.commit()

# 일괄 삽입
db_session.add_all([
    Employee(full_name="박영희", department="마케팅", email="park@example.com"),
    Employee(full_name="이준호", department="인사팀", email="lee@example.com")
])
db_session.commit()

조회 (Read)

# 전체 목록
all_employees = db_session.query(Employee).all()

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

# ID로 검색
emp = db_session.query(Employee).get(1)

# 조건 기반 검색
dev_team = db_session.query(Employee).filter(Employee.department == "개발팀").all()

업데이트 (Update)

# 개별 업데이트
emp = db_session.query(Employee).get(1)
emp.department = "프로덕트팀"
db_session.commit()

# 범위 업데이트
db_session.query(Employee).filter(Employee.email.like("%@example.com")).update({
    "department": "지원팀"
}, synchronize_session=False)
db_session.commit()

삭제 (Delete)

# 개별 삭제
emp = db_session.query(Employee).get(1)
db_session.delete(emp)
db_session.commit()

# 조건 삭제
db_session.query(Employee).filter(Employee.department == "퇴사자").delete(synchronize_session=False)
db_session.commit()

고급 쿼리 기법

조건 필터링

from sqlalchemy import or_

# 여러 조건 조합
results = db_session.query(Employee).filter(
    Employee.full_name.in_(["김철수", "박영희"]),
    or_(Employee.department == "개발팀", Employee.department == "프로덕트팀")
).all()

집계 함수 사용

from sqlalchemy import func

# 인원 수 집계
total_count = db_session.query(Employee).count()

# 부서별 인원 수
dept_stats = db_session.query(
    Employee.department,
    func.count(Employee.emp_id)
).group_by(Employee.department).all()

# 평균 값 계산
avg_id = db_session.query(func.avg(Employee.emp_id)).scalar()

조인 연산

# 내부 조인
joined_data = db_session.query(Employee, Project).join(Project).filter(
    Project.title.ilike("%AI%")
).all()

# 외부 조인
left_joined = db_session.query(Employee, Project).outerjoin(Project).all()

관계 기반 데이터 처리

# 관계 객체 생성
leader = Employee(full_name="최민수", department="기술팀")
project = Project(title="AI 프로젝트", start_date="2024-01-01", lead=leader)

db_session.add(project)
db_session.commit()

# 관계 탐색
print(f"프로젝트 '{project.title}'의 리더: {project.lead.full_name}")

# 태그 연결
tag_ai = Tag(label="AI")
tag_python = Tag(label="Python")

project.tags.append(tag_ai)
project.tags.append(tag_python)
db_session.commit()

트랜잭션 제어

try:
    new_user = Employee(full_name="테스트직원", department="임시팀", email="test@local.com")
    db_session.add(new_user)
    db_session.commit()
except Exception as e:
    db_session.rollback()
    raise e

# 컨텍스트 매니저 활용
from contextlib import contextmanager

@contextmanager
def get_db_session():
    session = SessionManager()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# 사용 예시
with get_db_session() as session:
    user = Employee(full_name="컨텍스트 사용자", email="context@test.com")
    session.add(user)

권장 사항

• 각 요청마다 새로운 세션 생성하고 종료 시 반드시 클로즈
• 예외 발생 시 트랜잭션 롤백 보장
• 지연 로딩 피하기, 필요한 경우 전부 로딩(즉시 로딩) 사용
• 연결 풀 크기와 타임아웃 적절히 설정
• 데이터 무결성 검증은 모델 또는 애플리케이션 레이어에서 처리