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:
핵심 포인트: depends_on만 쓰면 PostgreSQL이 완전히 준비되기 전에 FastAPI가 먼저 뜰 수 있다.
healthcheck + condition: service_healthy를 같이 써야 순서가 보장된다.
이 내용은 이전 글 Docker Compose depends_on이 제대로 동작하지 않는 이유와 해결법에서 자세히 다뤘다.
2. SQLAlchemy 비동기 설정
FastAPI는 비동기(async) 기반이라 SQLAlchemy도 비동기 드라이버를 써야 한다.
# 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:// 접두어가 중요하다.
일반 postgresql://로 쓰면 비동기 환경에서 오류가 난다.
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: 컨테이너 재시작 후 데이터 사라짐
volume 설정 없이 PostgreSQL을 띄운 경우다. 이 내용은 이전 글 Docker 재시작 후 PostgreSQL 데이터가 사라지는 이유와 영구 저장 방법에서 자세히 다뤘다.
동작 확인
# 컨테이너 실행 docker compose up -d # API 테스트 curl http://localhost:8105/api/videos
정상이라면 빈 배열 []이 반환된다. 테이블은 startup 이벤트에서 자동 생성된다.
마무리
FastAPI + PostgreSQL + Docker Compose 조합은 NAS 자동화 시스템의 기본 스택이다. 여기서 설명한 구조 위에 YouTube API 수집기, 성장률 계산 로직, 트렌드 대시보드가 올라간다.
관련 글:
댓글
댓글 쓰기