[FanMate] SQLModel로 KBO 데이터베이스 구축하기

in Dev Log on Fanmate

⚠️ 안내: 이 글은 학습 기록용입니다. 오류나 보완 의견은 댓글로 알려주세요.

배경

PostgreSQL 체크포인터를 통해 대화 상태를 관리하는 기능은 이미 구현해두었지만, 체크포인터는 에이전트 내부 상태(메시지 히스토리, 요약본 등)를 저장하는 용도에 가깝습니다. 반면 서비스에 필요한 도메인 데이터(KBO 10개 구단의 마스코트, 경기장, 응원가, 하이라이트 영상 등)를 다루려면 별도의 데이터베이스 스키마가 필요했습니다.

또한, 체크포인터에 저장되는 대화 이력은 SummarizationMiddleware에 의해 요약본으로 압축되기 때문에 원본 대화 기록을 보존하려면 채팅 메시지를 별도 테이블에 저장하는 구조도 필요했습니다. 체크포인터 데이터는 바이너리 형태로 직렬화되어 저장되므로, 대화 이력을 조회하려면 LangGraph의 체크포인터 API를 통해 역직렬화해야 합니다. 단순히 “이전 대화를 보여주는” 용도로 사용하기에는 다소 번거로운 구조였고, 이 점 역시 채팅 이력을 별도 테이블로 관리하게 된 이유였습니다.

ORM으로는 SQLModel을 선택하게 되었습니다. SQLAlchemy의 ORM 기능과 Pydantic의 데이터 검증을 하나의 클래스에서 함께 처리할 수 있어 FastAPI와 궁합이 좋았기 때문입니다. 모델 클래스 하나로 DB 테이블 정의와 API 스키마를 동시에 다룰 수 있다는 점도 선택의 이유였습니다.

ERD 설계

프로젝트 초기에 팀 전체가 함께 ERD를 설계했고, 이를 바탕으로 SQLModel 엔티티를 정의했습니다.

전체 구조는 Team을 중심으로 한 스키마 에 가깝습니다. Team 테이블이 중앙에 있고, 구단과 관련된 6개의 콘텐츠 테이블(Mascot, Stadium, Song, Uniform, Highlight, MatchTeam)이 연결됩니다. 사용자(User) 쪽에는 채팅 세션과 메시지가 1:N 관계로 이어지는 구조입니다.

                        ┌──────────┐
                        │  Users   │
                        │  (UUID)  │
                        └────┬─────┘
                   ┌─────────┼──────────┐
                   │         │          │
                   ▼         ▼          ▼
            ┌────────────┐ ┌──────────┐ ┌───────────┐
            │ChatSessions│ │MatchTeam │ │   Teams   │
            │  (1:N)     │ │  (1:1)   │ │           │
            └─────┬──────┘ └──────────┘ └─────┬─────┘
                  │                    ┌───────┼───────┬──────┬──────┐
                  ▼                    ▼       ▼       ▼      ▼      ▼
           ┌─────────────┐      ┌────────┐┌───────┐┌─────┐┌──────┐┌──────────┐
           │ChatMessages │      │Mascots ││Songs  ││Stad.││Unif. ││Highlights│
           │  (1:N)      │      └────────┘└───────┘└─────┘└──────┘└──────────┘
           └─────────────┘

erd

설계 과정에서 결정한 삭제 전략은 다음과 같습니다.

  • Team ↔ User: SET NULL — 구단이 삭제되어도 사용자 계정은 유지되어야 합니다.
  • Team ↔ 콘텐츠(Mascot, Song 등): CASCADE — 구단이 삭제되면 관련 콘텐츠도 함께 제거합니다.
  • User ↔ ChatSession ↔ ChatMessage: CASCADE — 사용자 탈퇴하면 대화 이력도 함께 삭제되도록 설계했습니다.

엔티티 정의

1. 공통 Mixin — TimeStampMixIn

모든 엔티티에 공통으로 필요한 생성 시각과 수정 시각을 Mixin으로 분리했습니다.

class TimeStampMixIn(SQLModel):
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

모든 테이블 모델이 이 Mixin을 상속받아 created_at, updated_at 필드를 공통으로 갖도록 했습니다.

2. Team — 중앙 테이블

Team은 전체 스키마의 중심이 되는 테이블입니다.

class Team(TimeStampMixIn, table=True):
    __tablename__ = "teams"

    team_id: int | None = Field(default=None, primary_key=True)
    team_name: str
    team_code: str          # LG, DOOSAN, KIA 등
    established_year: int | None

    # 1:N 관계
    users: list["User"] = Relationship(back_populates="team",
                                        sa_relationship_kwargs={"passive_deletes": True})
    uniforms: list["Uniform"] = Relationship(back_populates="team",
                                              cascade_delete=True)
    highlights: list["Highlight"] = Relationship(back_populates="team",
                                                  cascade_delete=True)
    stadiums: list["Stadium"] = Relationship(back_populates="team",
                                              cascade_delete=True)
    songs: list["Song"] = Relationship(back_populates="team",
                                        cascade_delete=True)
    mascots: list["Mascot"] = Relationship(back_populates="team",
                                            cascade_delete=True)
  • team_code는 JSON 데이터의 키로 사용되는 팀 약어입니다.
  • 시딩 과정에서 team_code → team_id 매핑을 만든 뒤, 이를 이용해 다른 엔티티의 외래 키를 연결했습니다.

3. User — 일반 로그인 + OAuth 통합

User는 일반 로그인과 OAuth 로그인을 하나의 테이블에서 함께 다룰 수 있도록 설계했습니다.

class User(TimeStampMixIn, table=True):
    __tablename__ = "users"

    user_id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    team_id: int | None = Field(default=None,
                                 foreign_key="teams.team_id",
                                 ondelete="SET NULL")

    # 일반 로그인
    email: str | None = Field(default=None, unique=True)
    password_hash: str | None = None

    # OAuth
    provider: str | None = None        # kakao, naver, google
    provider_id: str | None = None

    nickname: str
    refreshtoken: str | None = None
  • 일반 로그인 사용자는 email과 password_hash가 채워집니다.
  • OAuth 사용자는 provider와 provider_id가 채워집니다.
  • 두 방식 모두 nickname은 필수값으로 두었습니다.

4. 채팅 세션 & 메시지 — Enum 활용

class BotType(str, enum.Enum):
    JUJEOP = "JUJEOP"      # 주접 메이트
    SALES = "SALES"        # 영업왕

class ChatSessions(TimeStampMixIn, table=True):
    session_id: int | None = Field(default=None, primary_key=True)
    user_id: uuid.UUID = Field(foreign_key="users.user_id", ondelete="CASCADE")
    bot_type: BotType

class Role(str, enum.Enum):
    USER = "user"
    ASSISTANT = "assistant"

class ChatMessages(TimeStampMixIn, table=True):
    message_id: int | None = Field(default=None, primary_key=True)
    session_id: int | None = Field(foreign_key="chat_sessions.session_id",
                                    ondelete="CASCADE")
    role: Role
    content: str | None
  • 서비스에 두 종류의 챗봇(주접 메이트, 영업왕)이 있으므로 BotType Enum으로 세션을 구분했습니다.
  • 사용자 한 명이 봇 타입별로 독립적인 세션을 가질 수 있고, 각 세션에 메시지가 쌓이는 구조입니다.
  • Role Enum은 처음에 대문자(USER, ASSISTANT)로 정의했다가, LangChain의 메시지 포맷과 맞추기 위해 소문자로 변경했습니다.

이중 엔진 구성 — 동기 + 비동기

데이터베이스 엔진을 동기와 비동기 두 가지로 구성했습니다.

from sqlmodel import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

# 동기 엔진 — 테이블 생성, 시딩용
engine = create_engine(settings.DB_ENGINE_URI, echo=True)

# 비동기 엔진 — API 요청 처리용
async_engine = create_async_engine(settings.DB_ENGINE_URI)
async_session = async_sessionmaker(async_engine, expire_on_commit=False)
  • 동기 엔진 은 앱 시작 시 테이블 생성(create_db_and_tables())과 초기 데이터 시딩에 사용했습니다. 애플리케이션이 시작될 때 한 번만 수행되는 작업이기 때문에 동기로도 충분했습니다.
  • 비동기 엔진 은 FastAPI의 비동기 엔드포인트에서 채팅 메시지 저장, 대화 이력 조회 등 DB I/O를 처리하는 데 사용했습니다. 이를 통해 이벤트 루프를 블로킹하지 않고 요청을 처리할 수 있습니다.
  • DB URI를 이중으로 관리한 이유 는 SQLModel ORM과 LangGraph 체크포인터가 서로 다른 연결 문자열 형식을 사용하기 때문입니다.

장기적으로는 동기 엔진을 제거하고 비동기로 통합할 계획이었으나, 프로젝트 기간 내에는 우선순위가 더 높은 기능들이 있어 이중 엔진 구조를 유지한 채 마무리했습니다.

데이터 시딩

KBO 10개 구단의 데이터를 JSON 파일로 정리하고, 앱 시작 시 자동으로 DB에 적재하는 시딩 스크립트를 구현했습니다.

# backend/app/data/seed/main.py
def seed_all():
    with Session(engine) as session:
        team_code_to_id = seed_teams(session)        # 팀 먼저
        seed_mascots(session, team_code_to_id)       # 이후 콘텐츠
        seed_highlights(session, team_code_to_id)
        seed_songs(session, team_code_to_id)
        seed_stadiums(session, team_code_to_id)
        seed_uniforms(session, team_code_to_id)
        session.commit()

def seed_teams(session: Session) -> dict[str, int]:
    teams_file = DATA_DIR / "kbo_teams.json"
    data = load_json(teams_file)

    team_code_to_id = {}
    for team_code, team_info in data.get("meta", {}).get("teams", {}).items():
        existing = session.exec(
            select(Team).where(Team.team_code == team_code)
        ).first()

        if existing:
            team_code_to_id[team_code] = existing.team_id
            continue

        team = Team(team_name=team_info["name"], team_code=team_code)
        session.add(team)
        session.flush()      # team_id 생성 (auto-increment)
        team_code_to_id[team_code] = team.team_id

    return team_code_to_id
  • 먼저 Team 테이블을 시딩해 team_code → team_id 매핑을 만든 뒤, 이 매핑을 이용해 나머지 엔티티의 외래 키를 연결하는 방식으로 순서를 구성했습니다.
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

    with Session(engine) as session:
        team_exists = session.exec(select(Team).limit(1)).first()
        if not team_exists:
            seed_all()
        seed_mock_users(session)
        session.commit()
  • 앱의 startup 이벤트에서는 테이블 생성과 시딩을 자동으로 처리했습니다.
  • 중복 시딩 방지 로직을 넣어 앱을 재시작하더라도 데이터 중복 삽입되지 않도록 했습니다.
  • 개발 편의를 위해 Mock 유저(이메일 로그인 1명, OAuth 2명)도 자동 생성됩니다.

채팅 메시지 DB 저장 — 체크포인터와의 역할 분리

LangGraph의 체크포인터도 대화 이력을 저장하지만, 그 데이터는 어디까지나 에이전트의 내부 상태입니다. SummarizationMiddleware가 오래된 대화를 요약본으로 대체하기 때문에, 원본 메시지를 그대로 보존하려면 별도의 저장 경로가 필요했습니다.

따라서, 채팅 메시지는 ChatMessages 테이블에 별도로 저장하도록 설계했습니다.

# backend/app/services/sales/chat_utils.py

async def get_or_create_session(user_id: str, bot_type: BotType = BotType.SALES) -> int:
    """유저의 봇 타입별 세션 조회/생성"""
    async with async_session() as session:
        stmt = select(ChatSessions).where(
            ChatSessions.user_id == user_id,
            ChatSessions.bot_type == bot_type,
        )
        result = await session.execute(stmt)
        chat_session = result.scalars().first()

        if chat_session:
            return chat_session.session_id

        chat_session = ChatSessions(user_id=user_id, bot_type=bot_type)
        session.add(chat_session)
        await session.commit()
        await session.refresh(chat_session)
        return chat_session.session_id


async def save_message(session_id: int, role: Role, content: str):
    """채팅 메시지 1건 저장"""
    async with async_session() as session:
        message = ChatMessages(session_id=session_id, role=role, content=content)
        session.add(message)
        await session.commit()


async def get_chat_history_by_user(
    user_id: str, bot_type: BotType = BotType.SALES
) -> list[dict]:
    """유저의 봇 타입별 과거 대화 이력 조회"""
    async with async_session() as session:
        stmt = (
            select(ChatMessages)
            .join(ChatSessions)
            .where(
                ChatSessions.user_id == user_id,
                ChatSessions.bot_type == bot_type,
            )
            .order_by(ChatMessages.created_at)
        )
        result = await session.execute(stmt)
        messages = result.scalars().all()
        return [{"role": msg.role.value, "content": msg.content} for msg in messages]

채팅 메시지의 저장 타이밍은 다음과 같이 설계했습니다.

  • 유저 메시지: API 요청 수신 시 즉시 저장
  • AI 응답: 스트리밍 완료 후 전체 내용을 한 번에 저장 (중간 저장 시 불완전한 응답이 기록될 수 있으므로)

그 결과 데이터 흐름이 두 갈래로 분리되었습니다.

[사용자 메시지]
    ├── → 체크포인터 (에이전트 내부 상태 관리용)
    └── → ChatMessages 테이블 (원본 보존 및 대화 이력 조회용)

기존에 chat.py에 있던 get_chat_history 함수는 체크포인터에서 바이너리 데이터를 파싱하는 방식이었는데, 이를 제거하고 chat_utils.py의 DB 조회 방식으로 교체했습니다.

🔄 회고

SQLModel을 처음 사용하면서 가장 크게 느낀 점은, SQLAlchemy와 Pydantic의 장점을 한 모델 안에서 함께 가져갈 수 있다는 점이었습니다. FastAPI의 요청·응답 스키마와 DB 모델을 하나의 클래스로 다룰 수 있어 코드도 훨씬 간결해졌습니다.

물론, SQLModel 특유의 제약도 있었습니다. 예를 들어 Relationship 순환 참조 문제는 SQLAlchemy에서는 lazy 옵션으로 비교적 쉽게 다룰 수 있지만, SQLModel에서는 sa_relationship_kwargs를 통해 우회해야 했습니다. 이 부분은 공식 문서에서 명확하게 정리되어 있지 않아서, SQLAlchemy 문서까지 함께 참고해서 해결해야 했습니다.

체크포인터와 채팅 메시지 저장의 역할 분리도 설계 과정에서 했어야하는 중요한 판단이었는데, 기능에 대한 이해가 부족해 체크포인터 구현 이후에 재설계를 해야했습니다. 처음에는 체크포인터가 이미 대화를 저장하니 별도 저장이 필요 없지 않을까 생각했지만, 실제로 요약 미들웨어가 동작하면서 원본 대화가 그대로 보존되지 않는다는 점을 확인했었습니다. 그 이후 원본 메시지를 위한 별도 저장 구조가 필요하다는 판단을 내렸고, 초기 계획 단계에서의 설계를 바꿔 작업을 진행하였습니다. 이 경험은 이후 멀티 페르소나 구현에서 체크포인터를 상태를 직접 수정하는 작업의 기반이 되었던 것 같습니다.

참고

프로젝트 github 관련 github issue

© 2024. All rights reserved.

Powered by Hydejack v9.2.1