실습 2 — FastAPI 로 예측 API 만들기

from pathlib import Path

# 서버로 띄울 수 있게 파일로 만든다. 실습 3, 4가 이 파일을 쓴다
Path("out_delay_api.py").write_text('''"""배송 지연 예측 API. 실행: uvicorn out_delay_api:app --port 8013"""
from typing import Literal

import joblib
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel, Field

BUNDLE = joblib.load("out_delay_model.joblib")       # 서버가 뜰 때 한 번만 불러 둔다
PIPE = BUNDLE["파이프라인"]

app = FastAPI(title="배송 지연 예측 API", version=BUNDLE["판"])


class Order(BaseModel):
    """예측할 주문 하나. 칸마다 타입과 범위, 고를 수 있는 값을 정한다"""
    주문시각: int = Field(ge=0, le=23, description="주문한 시각(0~23)")
    거리km: float = Field(gt=0, le=1000)
    무게kg: float = Field(ge=0, le=100)
    지역: Literal["서울", "경기", "인천", "지방"]
    상품분류: Literal["의류", "생활용품", "식품", "도서", "가전"]
    날씨: Literal["맑음", "흐림", "비", "눈", "모름"] = "모름"
    택배사: Literal["한결택배", "빠른별택배", "새벽길택배"]


class Batch(BaseModel):
    주문들: list[Order] = Field(min_length=1, max_length=500)


def predict_rows(orders):
    """검사를 통과한 주문들을 학습 때와 같은 열의 표로 바꿔 확률을 계산한다"""
    table = pd.DataFrame([o.model_dump() for o in orders])[BUNDLE["숫자열"] + BUNDLE["범주열"]]
    probs = PIPE.predict_proba(table)[:, 1]
    t = BUNDLE["기준값"]
    return [{"지연확률": round(float(p), 4), "지연예상": bool(p >= t)} for p in probs]


@app.get("/health")
def health():
    return {"상태": "ok", "판": BUNDLE["판"]}


@app.get("/model")
def model_info():
    return {k: BUNDLE[k] for k in ("판", "기준값", "숫자열", "범주열", "시험성능", "sklearn")}


@app.post("/predict")
def predict(order: Order):
    return {**predict_rows([order])[0], "기준값": BUNDLE["기준값"], "판": BUNDLE["판"]}


@app.post("/predict/batch")
def predict_batch(batch: Batch):
    return {"건수": len(batch.주문들), "판": BUNDLE["판"], "결과": predict_rows(batch.주문들)}
''', encoding="utf-8")
print("API 파일:", Path("out_delay_api.py").stat().st_size, "바이트")

import sys
sys.path.insert(0, ".")                       # 작업 폴더(data)의 out_delay_api.py 를 찾게 한다
from fastapi.testclient import TestClient
from out_delay_api import app

client = TestClient(app)                      # 서버를 띄우지 않고 앱에 요청을 보낸다
r = client.get("/health")
print(r.status_code, r.json())
print(client.get("/model").json())

order = {"주문시각": 22, "거리km": 197.6, "무게kg": 4.6, "지역": "지방", "상품분류": "도서", "날씨": "눈", "택배사": "빠른별택배"}
r = client.post("/predict", json=order)       # json= 에 사전을 주면 JSON 본문과 Content-Type 을 알아서 붙인다
print(r.status_code, r.headers["content-type"])
print(r.json())

near = dict(order, 주문시각=9, 거리km=7.3, 지역="서울", 날씨="맑음", 택배사="한결택배")
print(client.post("/predict", json=near).json())

spec = client.get("/openapi.json").json()     # /docs 화면이 읽는 API 명세(OpenAPI)
for path, methods in spec["paths"].items():
    for m in methods:
        print(m.upper(), path)
fields = spec["components"]["schemas"]["Order"]["properties"]
print("Order 칸:", {k: v.get("enum", v.get("type")) for k, v in fields.items()})