FastAPI와 PostgreSQL 연동하기 (Docker 환경 실전 예제)

UGREEN DXP4800PLUS NAS에서 FastAPI 기반 쇼츠 트렌드 수집 시스템을 구성한 사례입니다.
사례의 DB에는 영상 700개 이상과 스냅샷 40,000개 이상이 기록됐고, 수집·저장은 30분 주기로 구성했습니다.
이 구조를 만들면서 FastAPI와 PostgreSQL 연동에서 막혔던 부분들을 정리합니다.

단순한 예제 코드가 아니라 Docker Compose 환경에서 실제로 동작하는 구성을 기준으로 설명합니다.

사례의 운영 환경

NAS: UGREEN DXP4800PLUS (Debian 12 기반)
FastAPI: Docker 컨테이너로 실행
PostgreSQL: 동일 Docker Compose 내 별도 컨테이너
ORM: SQLAlchemy 2.0 (비동기, asyncpg 드라이버)
DB명: shorts / 주요 테이블: videos, snapshots

1. Docker Compose 구성

FastAPI와 PostgreSQL을 같은 Compose 파일 안에 두면 컨테이너 이름으로 서로 통신할 수 있습니다.

services:
  db:
    image: postgres:15
    environment:
      POSTGRES_DB: shorts
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d shorts"]
      interval: 5s
      retries: 5

  api:
    build: .
    ports:
      - "8105:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://user:password@db:5432/shorts
    depends_on:
      db:
        condition: service_healthy

volumes:
  postgres_data:

api에 build: .를 썼다면 같은 디렉터리에 최소한 Dockerfile과 의존성 파일이 있어야 합니다. 아래는 이 글의 예제를 실행할 수 있게 만드는 최소 구성입니다.

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# requirements.txt
fastapi
uvicorn[standard]
sqlalchemy>=2.0
asyncpg

실제 프로젝트에서 추가 패키지를 import한다면 requirements.txt에도 함께 넣어야 합니다. 비밀번호는 설명용 문자열을 그대로 운영에 쓰지 말고 환경변수나 secret으로 분리합니다.

핵심 포인트: depends_on만 쓰면 PostgreSQL이 완전히 준비되기 전에 FastAPI가 먼저 뜰 수 있습니다.
healthcheck + condition: service_healthy를 같이 써야 순서가 보장됩니다.
이 내용은 이전 글 Docker Compose depends_on이 제대로 동작하지 않는 이유와 해결법에서 자세히 다뤘습니다.

2. SQLAlchemy 비동기 설정

FastAPI는 동기·비동기 DB 접근을 모두 사용할 수 있습니다. 이 예제는 async def 경로에서 SQLAlchemy 2.0의 비동기 엔진을 선택했기 때문에 asyncpg 드라이버를 사용합니다.

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, DeclarativeBase
import os

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_async_engine(DATABASE_URL, echo=False)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

class Base(DeclarativeBase):
    pass

이 비동기 엔진 예제에서는 postgresql+asyncpg:// 접두어가 필요합니다. 동기 SQLAlchemy 엔진을 선택한다면 psycopg 같은 동기 드라이버와 그에 맞는 URL을 사용해야 합니다.

3. 테이블 모델 정의

# models.py
from sqlalchemy import Column, Integer, String, DateTime, Float
from database import Base
from datetime import datetime

class Video(Base):
    __tablename__ = "videos"

    id = Column(Integer, primary_key=True, index=True)
    video_id = Column(String, unique=True, index=True)
    title = Column(String)
    channel_name = Column(String)
    view_count = Column(Integer, default=0)
    content_cluster = Column(String)
    format_type = Column(String)
    collected_at = Column(DateTime, default=datetime.utcnow)

4. FastAPI에서 DB 세션 주입

# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from database import AsyncSessionLocal, engine, Base
from models import Video

app = FastAPI()

# 앱 시작 시 테이블 자동 생성
@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# DB 세션 의존성
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# 영상 목록 조회
@app.get("/api/videos")
async def get_videos(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Video).limit(20))
    videos = result.scalars().all()
    return [{"id": v.id, "title": v.title, "view_count": v.view_count} for v in videos]

# 영상 저장
@app.post("/api/videos")
async def create_video(video_id: str, title: str, db: AsyncSession = Depends(get_db)):
    video = Video(video_id=video_id, title=title)
    db.add(video)
    await db.commit()
    return {"status": "ok"}

5. 실제로 겪은 오류와 해결

오류 1: connection refused (db 컨테이너가 준비되기 전에 api가 뜨는 경우)

sqlalchemy.exc.OperationalError: connection refused

원인: depends_on만 쓰고 healthcheck 없이 구성했을 때 발생합니다.
해결: 위 Compose 예제처럼 pg_isready healthcheck 추가.

오류 2: asyncpg.exceptions.UndefinedTableError

asyncpg.exceptions.UndefinedTableError: relation "videos" does not exist

원인: startup 이벤트에서 create_all을 하지 않았거나, 모델 import가 누락된 경우.
해결: main.py에서 모든 모델을 import한 뒤 create_all 실행.

오류 3: 컨테이너를 재생성한 뒤 이전 데이터가 보이지 않음

단순한 docker restart가 named volume이나 bind mount 데이터를 지우는 것은 아닙니다. 명시적인 영구 볼륨 없이 컨테이너를 재생성하면 새 익명 볼륨이 연결되어 이전 데이터가 없는 것처럼 보일 수 있으므로, 위 예제처럼 named volume 또는 bind mount를 명시합니다. 자세한 내용은 PostgreSQL 데이터 영구 저장 방법에서 다룹니다.

동작 확인

# 컨테이너 실행
docker compose up -d

# API 테스트
curl http://localhost:8105/api/videos

정상이라면 빈 배열 []이 반환됩니다. 테이블은 startup 이벤트에서 자동 생성됩니다.

마무리

FastAPI + PostgreSQL + Docker Compose 조합은 NAS 자동화 시스템의 기본 스택입니다.
여기서 설명한 구조 위에 YouTube API 수집기, 성장률 계산 로직, 트렌드 대시보드가 올라갑니다.