Tutorial Lengkap FastAPI untuk ML: Build Production ML APIs

FastAPI adalah framework web Python modern dengan performa tinggi untuk membangun APIs. Dengan dokumentasi OpenAPI otomatis, type hints, dan dukungan async, FastAPI adalah pilihan ideal untuk deploy model machine learning sebagai production-ready APIs.

Mengapa FastAPI untuk ML?

• High performance: Setara dengan NodeJS dan Go
• Type safety: Pydantic validation
• Auto documentation: Swagger UI dan ReDoc
• Async support: Handle concurrent requests
• Easy testing: Built-in test client

• ML model serving APIs
• Real-time inference endpoints
• Batch prediction services
• Feature engineering APIs
• Model management systems

Instalasi

pip install fastapi uvicorn

Dengan ML dependencies
pip install fastapi uvicorn scikit-learn joblib numpy pandas

Untuk async database
pip install fastapi[all] sqlalchemy asyncpg

Verify installation
python -c "import fastapi; print(fastapi.version)"

Quick Start

1. Hello World API

# main.py
from fastapi import FastAPI

app = FastAPI(
    title="ML API",
    description="Machine Learning API dengan FastAPI",
    version="1.0.0"
)

@app.get("/")
def readroot():

    return {"message": "Selamat datang di ML API"}

@app.get("/health")
def healthcheck():
    return {"status": "healthy"}

# Run server
uvicorn main:app --reload --host 0.0.0.0 --port 8000

Akses docs di http://localhost:8000/docs

2. Simple ML Prediction Endpoint

from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()

Load model saat startup
model = joblib.load("model.joblib")

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: float
    probability: list[float] | None = None

@app.post("/predict", responsemodel=PredictionResponse)

def predict(request: PredictionRequest):
    features = np.array(request.features).reshape(1, -1)
    prediction = model.predict(features)[0]

    # Get probability jika classifier
    probability = None
    if hasattr(model, "predictproba"):
        probability = model.predictproba(features)[0].tolist()


    return PredictionResponse(
        prediction=float(prediction),
        probability=probability
    )

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from enum import Enum

class ModelType(str, Enum):
    classification = "classification"
    regression = "regression"

class FeatureInput(BaseModel):
    age: int = Field(..., ge=0, le=120, description="Umur dalam tahun")
    income: float = Field(..., gt=0, description="Pendapatan tahunan")
    education: str = Field(..., description="Tingkat pendidikan")

    class Config:
        schemaextra = {
            "example": {
                "age": 35,
                "income": 75000.0,
                "education": "sarjana"
            }
        }

class BatchPredictionRequest(BaseModel):
    instances: List[FeatureInput]
    modelversion: Optional[str] = "latest"


class PredictionResult(BaseModel):
    prediction: float
    confidence: float
    modelversion: str

class BatchPredictionResponse(BaseModel):
    predictions: List[PredictionResult]
    processingtimems: float

2. Validation

from pydantic import BaseModel, validator, rootvalidator

class MLRequest(BaseModel):
    features: List[float]

    @validator('features')
    def checkfeaturecount(cls, v):

        if len(v) != 10:
            raise ValueError('Harus menyediakan tepat 10 features')
        return v

    @validator('features', eachitem=True)
    def checkfeaturerange(cls, v):
        if not -1 <= v <= 1:
            raise ValueError('Features harus dinormalisasi antara -1 dan 1')
        return v

class DateRangeRequest(BaseModel):
    startdate: str

    enddate: str

    @rootvalidator

    def checkdates(cls, values):
        start = values.get('startdate')

        end = values.get('enddate')
        if start and end and start > end:
            raise ValueError('startdate harus sebelum enddate')
        return values

Model Loading Patterns

1. Startup Event

from fastapi import FastAPI
from contextlib import asynccontextmanager
import joblib

mlmodels = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Load models
    mlmodels["classifier"] = joblib.load("classifier.joblib")
    mlmodels["regressor"] = joblib.load("regressor.joblib")

    print("Models loaded successfully")
    yield
    # Shutdown: Cleanup
    mlmodels.clear()
    print("Models unloaded")

app = FastAPI(lifespan=lifespan)

@app.post("/predict/classification")
def classify(features: List[float]):
    model = mlmodels["classifier"]

    prediction = model.predict([features])[0]
    return {"prediction": int(prediction)}

from fastapi import FastAPI, Depends
from functools import lrucache
import joblib

class ModelManager:
    def init(self):
        self.models = {}

    def loadmodel(self, name: str, path: str):

        self.models[name] = joblib.load(path)

    def getmodel(self, name: str):
        return self.models.get(name)

@lrucache()

def getmodelmanager():

    manager = ModelManager()
    manager.loadmodel("classifier", "models/classifier.joblib")
    return manager

app = FastAPI()

@app.post("/predict")
def predict(
    features: List[float],
    manager: ModelManager = Depends(getmodelmanager)
):
    model = manager.getmodel("classifier")

    return {"prediction": model.predict([features])[0]}

from fastapi import FastAPI, HTTPException
from typing import Dict
import joblib
import os

app = FastAPI()

class ModelRegistry:
    def init(self, modelsdir: str = "models"):
        self.modelsdir = modelsdir
        self.models: Dict[str, any] = {}
        self.loadallmodels()


    def loadallmodels(self):
        for version in os.listdir(self.modelsdir):

            path = os.path.join(self.modelsdir, version, "model.joblib")
            if os.path.exists(path):
                self.models[version] = joblib.load(path)

    def getmodel(self, version: str = "latest"):

        if version == "latest":
            version = max(self.models.keys())
        if version not in self.models:
            raise KeyError(f"Model version {version} tidak ditemukan")
        return self.models[version], version

registry = ModelRegistry()

@app.post("/predict/{version}")
def predict(version: str, features: List[float]):
    try:
        model, actualversion = registry.getmodel(version)

        prediction = model.predict([features])[0]
        return {
            "prediction": prediction,
            "modelversion": actualversion

        }
    except KeyError as e:
        raise HTTPException(statuscode=404, detail=str(e))

Batch Predictions

1. Sync Batch Endpoint

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
import numpy as np
import time

class BatchRequest(BaseModel):
    instances: List[List[float]]

class BatchResponse(BaseModel):
    predictions: List[float]
    count: int
    processingtimems: float

@app.post("/predict/batch", responsemodel=BatchResponse)

def batchpredict(request: BatchRequest):
    start = time.time()

    features = np.array(request.instances)
    predictions = model.predict(features).tolist()

    processingtime = (time.time() - start) * 1000


    return BatchResponse(
        predictions=predictions,
        count=len(predictions),
        processingtimems=processingtime
    )

2. Async Batch Processing

from fastapi import FastAPI, BackgroundTasks
from uuid import uuid4
import asyncio

In-memory job storage (gunakan Redis di production)
jobs = {}

class JobStatus(BaseModel):
    jobid: str

    status: str
    result: Optional[List[float]] = None

async def processbatch(jobid: str, instances: List[List[float]]):

    jobs[jobid] = {"status": "processing"}

    # Simulate long processing
    await asyncio.sleep(5)

    features = np.array(instances)
    predictions = model.predict(features).tolist()

    jobs[jobid] = {

        "status": "completed",
        "result": predictions
    }

@app.post("/predict/batch/async")
async def asyncbatchpredict(

    request: BatchRequest,
    backgroundtasks: BackgroundTasks
):
    jobid = str(uuid4())

    jobs[jobid] = {"status": "queued"}

    backgroundtasks.addtask(
        processbatch, jobid, request.instances
    )

    return {"jobid": jobid, "status": "queued"}

@app.get("/jobs/{jobid}", responsemodel=JobStatus)
def getjobstatus(jobid: str):

    if jobid not in jobs:
        raise HTTPException(statuscode=404, detail="Job tidak ditemukan")


    job = jobs[jobid]
    return JobStatus(
        jobid=jobid,
        status=job["status"],
        result=job.get("result")
    )

File Upload untuk Predictions

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
import pandas as pd
import io

@app.post("/predict/csv")
async def predictfromcsv(file: UploadFile = File(...)):
    # Baca CSV
    contents = await file.read()
    df = pd.readcsv(io.BytesIO(contents))


    # Buat predictions
    features = df.values
    predictions = model.predict(features)

    # Tambahkan predictions ke dataframe
    df['prediction'] = predictions

    # Return sebagai CSV
    output = io.StringIO()
    df.tocsv(output, index=False)
    output.seek(0)

    return StreamingResponse(
        iter([output.getvalue()]),
        mediatype="text/csv",

        headers={"Content-Disposition": "attachment; filename=predictions.csv"}
    )

@app.post("/predict/image")
async def predictfromimage(file: UploadFile = File(...)):

    from PIL import Image
    import io

    # Baca image
    contents = await file.read()
    image = Image.open(io.BytesIO(contents))

    # Preprocess
    image = image.resize((224, 224))
    features = np.array(image) / 255.0
    features = features.reshape(1, -1)

    # Predict
    prediction = model.predict(features)[0]

    return {"prediction": int(prediction), "filename": file.filename}

from fastapi import FastAPI, Security, HTTPException
from fastapi.security import APIKeyHeader
import os

APIKEY = os.getenv("APIKEY", "secret-key")

apikeyheader = APIKeyHeader(name="X-API-Key")


def verifyapikey(apikey: str = Security(apikeyheader)):
    if apikey != APIKEY:
        raise HTTPException(statuscode=403, detail="API key tidak valid")

    return apikey

@app.post("/predict")
def predict(
    features: List[float],
    apikey: str = Security(verifyapikey)

):
    prediction = model.predict([features])[0]
    return {"prediction": prediction}

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRETKEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESSTOKENEXPIREMINUTES = 30


oauth2scheme = OAuth2PasswordBearer(tokenUrl="token")

def createaccesstoken(data: dict):
    toencode = data.copy()

    expire = datetime.utcnow() + timedelta(minutes=ACCESSTOKENEXPIREMINUTES)
    toencode.update({"exp": expire})

    return jwt.encode(toencode, SECRETKEY, algorithm=ALGORITHM)


async def getcurrentuser(token: str = Depends(oauth2scheme)):
    try:
        payload = jwt.decode(token, SECRETKEY, algorithms=[ALGORITHM])

        username = payload.get("sub")
        if username is None:
            raise HTTPException(statuscode=401, detail="Token tidak valid")
        return username
    except JWTError:
        raise HTTPException(statuscode=401, detail="Token tidak valid")


@app.post("/token")
async def login(formdata: OAuth2PasswordRequestForm = Depends()):
    # Verify credentials (simplified)
    if formdata.username != "admin" or formdata.password != "password":
        raise HTTPException(statuscode=401, detail="Credentials tidak valid")


    accesstoken = createaccesstoken(data={"sub": formdata.username})

    return {"accesstoken": accesstoken, "tokentype": "bearer"}

@app.post("/predict")
async def predict(
    features: List[float],
    currentuser: str = Depends(getcurrentuser)

):
    prediction = model.predict([features])[0]
    return {"prediction": prediction, "user": currentuser}

Error Handling

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError

app = FastAPI()

class PredictionError(Exception):
    def init(self, message: str, modelname: str):

        self.message = message
        self.modelname = modelname


@app.exceptionhandler(PredictionError)
async def predictionerrorhandler(request: Request, exc: PredictionError):
    return JSONResponse(
        statuscode=500,

        content={
            "error": "predictionfailed",
            "message": exc.message,
            "model": exc.modelname

        }
    )

@app.exceptionhandler(ValidationError)
async def validationerrorhandler(request: Request, exc: ValidationError):
    return JSONResponse(
        statuscode=422,

        content={
            "error": "validationfailed",
            "details": exc.errors()
        }
    )

@app.post("/predict")
def predict(features: List[float]):
    try:
        prediction = model.predict([features])[0]
        return {"prediction": prediction}
    except Exception as e:
        raise PredictionError(
            message=str(e),
            modelname="classifierv1"
        )

Monitoring dan Logging

from fastapi import FastAPI, Request
import logging
import time
from prometheusclient import Counter, Histogram, generatelatest
from starlette.responses import Response

Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

Prometheus metrics
REQUESTCOUNT = Counter(

    'mlapirequeststotal',
    'Total API requests',
    ['method', 'endpoint', 'status']
)

PREDICTIONLATENCY = Histogram(

    'mlpredictionlatencyseconds',
    'Prediction latency dalam detik'
)

app = FastAPI()

@app.middleware("http")
async def logrequests(request: Request, callnext):
    starttime = time.time()


    response = await callnext(request)

    processtime = time.time() - starttime

    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.statuscode} "

        f"- Time: {processtime:.3f}s"
    )

    REQUESTCOUNT.labels(

        method=request.method,
        endpoint=request.url.path,
        status=response.statuscode
    ).inc()

    return response

@app.post("/predict")
def predict(features: List[float]):
    with PREDICTIONLATENCY.time():

        prediction = model.predict([features])[0]
    return {"prediction": prediction}

@app.get("/metrics")
def metrics():
    return Response(
        generatelatest(),
        mediatype="text/plain"

    )

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

Copy application
COPY . .

Expose port
EXPOSE 8000

Run dengan uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
services:
  api:
    build: .
    ports:
"8000:8000"
    volumes:

./models:/app/models
    environment:

MODELPATH=/app/models/model.joblib
APIKEY=${APIKEY}
    healthcheck:

      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

Testing

# testapi.py
from fastapi.testclient import TestClient
from main import app
import pytest

client = TestClient(app)

def testhealthcheck():

    response = client.get("/health")
    assert response.statuscode == 200
    assert response.json()["status"] == "healthy"

def testpredict():

    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0, 4.0, 5.0]}
    )
    assert response.statuscode == 200
    assert "prediction" in response.json()

def testpredictinvalidinput():

    response = client.post(
        "/predict",
        json={"features": "invalid"}
    )
    assert response.statuscode == 422

def testbatchpredict():
    response = client.post(
        "/predict/batch",
        json={
            "instances": [
                [1.0, 2.0, 3.0],
                [4.0, 5.0, 6.0]
            ]
        }
    )
    assert response.statuscode == 200

    assert len(response.json()["predictions"]) == 2

Run dengan: pytest testapi.py -v

Kesimpulan

FastAPI adalah framework ideal untuk ML APIs dengan:

Key takeaways:

• Gunakan Pydantic untuk request/response validation
• Load models saat startup dengan lifespan events
• Implement proper error handling
• Tambahkan monitoring dengan Prometheus
• Test secara menyeluruh sebelum deployment