환경 준비 및 필수 라이브러리 설치

Python 생태계에서 관계형 데이터베이스와의 상호작용을 원활하게 하기 위해 SQLAlchemy 라이브러리가 표준적으로 사용됩니다. 이를 사용하려면 우선 파이썬 패키지 매니저를 통해 주 모듈을 설치해야 합니다.

pip install sqlalchemy

연결하려는 데이터베이스 엔진에 따라 추가적인 드라이버가 필요할 수 있습니다. 예를 들어 PostgreSQL 의 경우 psycopg2-binary, MySQL 의 경우 pymysql 또는 mysql-connector-python 패키지를 별도로 가져와야 합니다.

핵심 아키텍처 이해

ORM 패턴을 구현하는 SQLAlchemy 는 몇 가지 주요 컴포넌트로 구성됩니다.

• Connection Engine: 데이터베이스 서버와의 물리적 연결을 담당하며 SQL 명령을 전달합니다.
• Session Object: 영속성을 관리하고 변경 사항을 추적하여 커밋하거나 롤백합니다.
• Base Model: 데이터베이스 테이블 구조를 클래스로 나타내는 추상 베이스입니다.
• Statement Builder: 복잡한 쿼리 논리를 객체 지향적으로 생성합니다.

데이터 소스 연결 설정

엔진은 URI 형식을 사용하여 연결 정보를 정의합니다. SQLite 는 파일 기반이므로 경로를 지정하고, 타겟 DB 는 호스트, 포트, 인증 정보를 포함해야 합니다.

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

# 엔진 인스턴스 생성 (SQLite 예시, Echo 옵션으로 SQL 로그 출력)
db_connect = create_engine('sqlite:///data_storage.db', echo=False)

# PostgreSQL 예시 주석
# db_connect = create_engine('postgresql://user:pass@host/dbname')

# 세션 팩토리 클래스 정의
LocalSessionFactory = sessionmaker(bind=db_connect, autocommit=False)

# 활성 세션 생성 함수
def get_active_session():
    return LocalSessionFactory()

데이터 모델 및 스키마 설계

테이블 구조를 Python 클래스로 매핑합니다. 여기서 Primary Key 와 ForeignKey 를 명시하여 관계성을 정의합니다.

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, DeclarativeBase, Mapped

class ModelBase(DeclarativeBase):
    pass

# 사용자 계정 테이블
class Member(ModelBase):
    __tablename__ = 'members'

    uid: Mapped[int] = Column(Integer, primary_key=True, autoincrement=True)
    nickname: Mapped[str] = Column(String(50), nullable=False)
    contact_email: Mapped[str] = Column(String(100), unique=True)

    # 작성한 글 목록 (일대다 관계)
    authored_entries = relationship("Article", back_populates="writer")

# 게시판 항목 테이블
class Article(ModelBase):
    __tablename__ = 'articles'

    aid: Mapped[int] = Column(Integer, primary_key=True)
    headline: Mapped[str] = Column(String(150))
    body_text: Mapped[str] = Column(String(2000))
    creator_id: Mapped[int] = Column(ForeignKey('members.uid'))

    writer = relationship("Member", back_populates="authored_entries")

# 태그 매핑 테이블 (다대다 관계를 위한 중간 테이블)
association_table = Table(
    'article_tags',
    ModelBase.metadata,
    Column('article_ref', Integer, ForeignKey('articles.aid'), primary_key=True),
    Column('tag_ref', Integer, ForeignKey('tags.id'), primary_key=True)
)

class Tag(ModelBase):
    __tablename__ = 'tags'

    id: Mapped[int] = Column(Integer, primary_key=True)
    label_name: Mapped[str] = Column(String(30), unique=True)

    linked_articles = relationship(
        "Article", 
        secondary=association_table,
        backref="applied_tags"
    )

스키마 적용 및 초기화

정의된 클래스를 바탕으로 실제 데이터베이스 내에 테이블을 생성합니다.

# 모든 모델을 기반으로 테이블 생성
ModelBase.metadata.create_all(bind=db_connect)

데이터 생명주기 관리 (CRUD)

세션을 통해 데이터를 추가, 조회, 수정, 삭제하는 작업을 수행합니다. 트랜잭션은 각 작업 단위로 분리하여 안정성을 보장합니다.

기록 생성

current_db = get_active_session()

new_member = Member(nickname="developer_kim", contact_email="kim@example.com")
current_db.add(new_member)

# 여러 건의 데이터 한 번에 로드
bulk_items = [
    Member(nickname="dev_lee", contact_email="lee@test.com"),
    Member(nickname="dev_park", contact_email="park@test.com")
]
current_db.add_all(bulk_items)
current_db.commit()

정보 조회

# 전체 목록
all_members = current_db.query(Member).all()

# 특정 조건 만족 첫 번째 레코드
target = current_db.query(Member).filter_by(nickname="developer_kim").first()

# 기본 키로 직접 접근
specific_user = current_db.get(Member, target.uid)

내역 갱신

# 개체 수정 후 저장
existing_user = current_db.get(Member, 1)
if existing_user:
    existing_user.nickname = "updated_kim"
    current_db.commit()

# 배치 업데이트 (SQL 레벨)
current_db.query(Member).filter(Member.nickname.like("dev_%")).update(
    {Member.nickname: "team_member"}, 
    synchronize_session='fetch'
)
current_db.commit()

항목 제거

# 단일 삭제
to_delete = current_db.get(Member, 1)
current_db.delete(to_delete)
current_db.commit()

# 조건부 배치 삭제
current_db.query(Member).filter(Member.contact_email.endswith("@temp.com")).delete(synchronize_session=False)
current_db.commit()

고급 쿼리 연산

데이터 필터링, 정렬, 집계 기능을 활용하여 복잡한 분석을 수행합니다.

from sqlalchemy import func, or_, desc

# 필터링 및 정렬
sorted_list = current_db.query(Member).filter(
    or_(Member.nickname == "A", Member.nickname == "B")
).order_by(desc(Member.uid)).limit(5).all()

# 집계 함수 활용 (총 인원 계산)
total_count = current_db.query(func.count(Member.uid)).scalar()

# 조인을 통한 복합 조회
joined_data = current_db.query(Member, Article).join(Article).filter(
    Article.headline.contains("Tutorial")
).all()

관계성 탐색 및 조작

정의된 외래 키와 역참조 속성을 통해 객체 그래프를 탐색하고 연결 데이터를 관리합니다.

# 새로운 게시글 등록 시 작성자 자동 연결
author_obj = current_db.get(Member, 1)
new_post = Article(headline="New Tech", body="Details...", creator_id=author_obj.uid)
# 또는 관계 속성 활용
author_obj.authored_entries.append(new_post)
current_db.add(author_obj)
current_db.commit()

# 다대다 관계 처리: 글에 태그 부여
python_tag = Tag(label_name="Python")
java_tag = Tag(label_name="Java")
new_post.applied_tags.extend([python_tag, java_tag])
current_db.commit()

# 조회 시 관계 확인
print(f"{new_post.headline} 관련 태그:")
for t in new_post.applied_tags:
    print(f"- {t.label_name}")

트랜잭션 제어 전략

실패 시 이전 상태로 복구되도록 try-except 블록이나 컨텍스트 매니저를 사용하여 상태 무결성을 보호합니다.

def safe_registration(session, name, email):
    try:
        candidate = Member(nickname=name, contact_email=email)
        session.add(candidate)
        session.commit()
        return candidate.uid
    except Exception as err:
        session.rollback()
        raise RuntimeError(f"회원가입 실패: {err}")

# 컨텍스트 매니저를 통한 자동 클로징 패턴
from contextlib import contextmanager

@contextmanager
def managed_transaction():
    sess = get_active_session()
    try:
        yield sess
        sess.commit()
    except Exception:
        sess.rollback()
        raise
    finally:
        sess.close()

# 활용 예시
with managed_transaction() as db:
    db.add(Member(nickname="safe_user", contact_email="safe@mail.com"))