Desenvolvimento de Backend de ML com FastAPI

Descubra como criar APIs de ML de alto desempenho com FastAPI.

Configuração Básica

1. Instalação

pip install fastapi uvicorn pydantic numpy pandas scikit-learn

2. Estrutura do Projeto

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(
    title="API ML",
    description="API pour prédictions ML",
    version="1.0.0"
)

# Modèles de données
class PredictionRequest(BaseModel):
    features: List[float]
    model_type: Optional[str] = "default"

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str

Endpoints da API

1. Rotas Básicas

@app.get("/")
async def root():
    return {"message": "API ML active"}

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

@app.get("/models")
async def list_models():
    return {
        "available_models": [
            "classification",
            "regression",
            "clustering"
        ]
    }

2. Endpoints de Predição

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # Carregamento do modelo
        model = load_model(request.model_type)
        
        # Predição
        prediction = model.predict([request.features])[0]
        confidence = model.predict_proba([request.features])[0]
        
        return PredictionResponse(
            prediction=float(prediction),
            confidence=float(max(confidence)),
            model_version="1.0.0"
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Gerenciamento de Modelos

1. Carregamento e Gerenciamento

from typing import Dict
import joblib
import os

class ModelManager:
    def __init__(self):
        self.models: Dict[str, object] = {}
        self.model_path = "models/"
        
    def load_model(self, model_type: str) -> object:
        if model_type not in self.models:
            path = os.path.join(self.model_path, f"{model_type}.pkl")
            if not os.path.exists(path):
                raise HTTPException(
                    status_code=404,
                    detail=f"Modèle {model_type} non trouvé"
                )
            self.models[model_type] = joblib.load(path)
        return self.models[model_type]

model_manager = ModelManager()

2. Validação de Dados

from pydantic import validator
import numpy as np

class FeatureValidation(BaseModel):
    features: List[float]
    
    @validator('features')
    def validate_features(cls, v):
        if not v:
            raise ValueError("La liste des features ne peut pas être vide")
        if not all(isinstance(x, (int, float)) for x in v):
            raise ValueError("Toutes les features doivent être numériques")
        return v

Middleware e Segurança

1. Middleware Personalizado

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2. Autenticação

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token invalide"
        )

Testes e Documentação

1. Testes Unitários

# test_api.py
from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

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

2. Documentação da API

# Adição de documentação personalizada
@app.get("/docs")
async def custom_docs():
    return {
        "endpoints": {
            "/predict": {
                "description": "Endpoint de prédiction",
                "parameters": {
                    "features": "Liste de features numériques",
                    "model_type": "Type de modèle à utiliser"
                }
            }
        }
    }

Boas Práticas

1. Desempenho

   • Cache dos modelos
   • Validação eficiente
   • Gerenciamento de memória
   • Monitoramento

2. Segurança

   • Autenticação
   • Validação das entradas
   • Rate limiting
   • Logging

3. Manutenção

   • Testes automatizados
   • Documentação
   • Versionamento
   • Monitoramento

Conclusão

Pontos-chave a reter:

• API de alto desempenho
• Segurança reforçada
• Documentação completa
• Testes automatizados

Recomendações:

• Seguir os padrões REST
• Implementar a segurança
• Manter a documentação
• Automatizar os testes
• Aprofundar-se em FastAPI