FastAPI – Building Your First API

from fastapi import FastAPI, HTTPException, UploadFile, File
from pydantic import BaseModel, Field
from typing import Optional
import asyncio, httpx, uuid

app = FastAPI(title="AI Video Analysis API", version="1.0")

# ── 데이터 모델 (Pydantic) ──────────────────────────────────
class AnalysisRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=500,
                        description="AI에게 보낼 질문")
    language: str = Field(default="ko", pattern="^(ko|en)$")

class AnalysisResult(BaseModel):
    job_id: str
    status: str          # "processing" | "done" | "error"
    answer: Optional[str] = None

# 간단한 인메모리 저장소 (실제론 Redis/DB 사용)
jobs: dict[str, AnalysisResult] = {}

# ── 엔드포인트 ──────────────────────────────────────────────
@app.post("/analyze", response_model=AnalysisResult, status_code=202)
async def start_analysis(request: AnalysisRequest,
                         video: UploadFile = File(...)):
    """비디오를 업로드하고 AI 분석을 시작합니다."""
    job_id = str(uuid.uuid4())
    jobs[job_id] = AnalysisResult(job_id=job_id, status="processing")

    # 3개 워커에 동시 요청 (챕터 1 다이어그램의 패턴)
    asyncio.create_task(_run_workers(job_id, request.prompt))
    return jobs[job_id]

@app.get("/analyze/{job_id}", response_model=AnalysisResult)
async def get_result(job_id: str):
    """분석 결과를 조회합니다."""
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    return jobs[job_id]

async def _run_workers(job_id: str, prompt: str):
    """3개 워커에 동시 요청을 보내는 내부 함수."""
    worker_urls = [
        "http://worker-a:8001/infer",  # Qwen 27B
        "http://worker-b:8002/infer",  # Qwen 9B
        "http://worker-c:8003/infer",  # Qwen 4B
    ]
    async with httpx.AsyncClient(timeout=60) as client:
        # asyncio.gather = 3개 요청 동시 발사
        tasks = [client.post(url, json={"prompt": prompt}) for url in worker_urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    answers = [r.json()["answer"] for r in results if not isinstance(r, Exception)]
    jobs[job_id] = AnalysisResult(
        job_id=job_id, status="done", answer=" | ".join(answers)
    )

# 실행: uvicorn main:app --reload --port 8000
# 문서:  http://localhost:8000/docs