• 목차
• 🟥 Epic 1 — Task 4
  • 🗄️ PostgreSQL 연결 및 SQLAlchemy 모델 생성 (설명형 Markdown)
    • 📌 작업 개요
    • 🎯 작업 목적
    • 🛠️ 진행 내용
      • 🗂️ 1. PostgreSQL 연결 설정 파일 생성 (app/db.py)
      • 🗂️ 2. Django 모델 → SQLAlchemy 모델 변환
      • 🗂️ 3. JSONField → JSONB 변환
      • 🗂️ 4. 관계(OneToOne, ForeignKey) 변환
      • 🗂️ 5. DB 테이블 생성 스크립트 작성
      • 🗂️ 6. DB 연결 테스트
    • 📌 산출물

🟥 Epic 1 — Task 4¶

🗄️ PostgreSQL 연결 및 SQLAlchemy 모델 생성 (설명형 Markdown)¶

📌 작업 개요¶

FastAPI mgmt 서버가 실제 데이터를 저장하고 조회할 수 있도록
PostgreSQL 데이터베이스 연결을 설정하고,
기존 Django ORM 모델을 기반으로 SQLAlchemy 모델을 생성하는 단계이다.

이 작업은 이후 모든 API 개발, MQTT worker 데이터 처리, reverse SSH 상태 저장 등
전체 시스템의 데이터 흐름을 책임지는 핵심 기반이 된다.

🎯 작업 목적¶

• FastAPI 서버에서 PostgreSQL에 안정적으로 연결할 수 있도록 한다.

• SQLAlchemy ORM을 사용해 Django 모델 구조를 그대로 재현한다.

• JSONField → JSONB 변환을 포함하여 PostgreSQL의 고급 기능을 활용한다.

• DB 연결 모듈(db.py)을 표준화하여 모든 라우터/worker에서 재사용 가능하게 한다.

• 향후 마이그레이션 및 스키마 확장에 대비한 구조를 마련한다.

🛠️ 진행 내용¶

🗂️ 1. PostgreSQL 연결 설정 파일 생성 (app/db.py)¶

FastAPI에서 DB 연결을 담당하는 모듈을 생성한다.

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

DATABASE_URL = "postgresql://postgres:password@localhost:5432/mgmt"

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    echo=False
)

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

Base = declarative_base()

📌 설계 포인트

• pool_pre_ping=True → DB 연결 끊김 방지

• autocommit=False → 트랜잭션 안정성 확보

• declarative_base() → SQLAlchemy 모델의 기반 클래스

🗂️ 2. Django 모델 → SQLAlchemy 모델 변환¶

기존 Django 모델을 분석하여 SQLAlchemy 모델로 변환한다.

아래는 Park의 ServerMaster 모델을 SQLAlchemy로 옮긴 예시이다.

from sqlalchemy import Column, String, DateTime
from datetime import datetime
from app.db import Base

class ServerMaster(Base):
    __tablename__ = "server_master"

    hostname = Column(String(100), primary_key=True)
    alias = Column(String(100))
    version = Column(String(20))
    created_at = Column(DateTime, default=datetime.utcnow)

📌 설계 포인트

• Django의 auto_now_add=True → SQLAlchemy에서는 default=datetime.utcnow

• PK, max_length, null 옵션 그대로 유지

• 테이블명은 Django와 동일하게 유지하여 마이그레이션 용이

🗂️ 3. JSONField → JSONB 변환¶

PostgreSQL의 강력한 JSONB 타입을 사용한다.

예: ServerStatus의 CPU/RAM 히스토리

from sqlalchemy.dialects.postgresql import JSONB

cpu_history = Column(JSONB, default=list)
ram_history = Column(JSONB, default=list)

📌 설계 포인트

• JSONB는 인덱싱 가능

• 배열 검색 및 부분 업데이트 가능

• Django JSONField보다 훨씬 빠르고 안정적

🗂️ 4. 관계(OneToOne, ForeignKey) 변환¶

Django의 OneToOneField → SQLAlchemy에서는 ForeignKey + relationship 조합으로 구현한다.

예:

from sqlalchemy import Column, Integer, Float, ForeignKey
from sqlalchemy.orm import relationship

server = Column(String(100), ForeignKey("server_master.hostname"), primary_key=True)
master = relationship("ServerMaster", back_populates="status")

그리고 ServerMaster에도 관계를 추가한다:

status = relationship("ServerStatus", uselist=False, back_populates="master")

📌 설계 포인트

• uselist=False → OneToOne 관계

• Django의 related_name → SQLAlchemy의 back_populates

🗂️ 5. DB 테이블 생성 스크립트 작성¶

초기 테이블 생성을 위한 스크립트를 작성한다.

scripts/migrate.sh:

#!/bin/bash
python -c "from app.db import Base, engine; Base.metadata.create_all(bind=engine)"

🗂️ 6. DB 연결 테스트¶

FastAPI 라우터에서 DB 연결이 정상인지 테스트한다.

from fastapi import APIRouter, Depends
from app.db import SessionLocal

router = APIRouter()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@router.get("/db-test")
def db_test(db=Depends(get_db)):
    return {"status": "ok"}

📌 산출물¶

• PostgreSQL 연결 모듈 (app/db.py)

• SQLAlchemy 모델 4종 (ServerMaster, ServerStatus, ServerControl, ServerDetail)

• JSONB 필드 적용

• 관계(OneToOne) 변환 완료

• DB 테이블 생성 스크립트 작성

• /db-test API로 연결 검증 완료