Coverage for src / main.py: 77%

1import logging 

2from contextlib import asynccontextmanager 

3 

4from fastapi import APIRouter, FastAPI, HTTPException 

5from fastapi.responses import JSONResponse 

6 

7from src.models.predict_model import load_artifacts, predict_churn 

8from src.schemas.api_schema import CustomerRequest 

9 

10# Instancia o logger configurado na Etapa 3 

11logger = logging.getLogger(__name__) 

12 

13 

14@asynccontextmanager 

15async def lifespan(app: FastAPI): 

16 """ 

17 O Lifespan Events do FastAPI nos permite executar código ANTES de aceitar a primeira requisição. 

18 Isso é crítico em MLOps (Cold Start). Vamos carregar as pesadas matrizes do PyTorch (.pth) 

19 e o Scikit-Learn (.joblib) diretamente na RAM do servidor aqui. 

20 """ 

21 logger.info( 

22 "Iniciando FastAPI... Puxando pesos do PyTorch para a RAM (Cold Start Optimization)" 

23 ) 

24 try: 

25 load_artifacts() 

26 logger.info("Artefatos de Inteligência Artificial carregados com sucesso!") 

27 except Exception as e: 

28 logger.error(f"Falha Crítica ao carregar os modelos: {e}") 

29 # A API pode subir mesmo falhando, mas o /health alertará sobre o erro. 

30 

31 yield # O servidor passa a aceitar requisições 

32 

33 logger.info("Desligando API. Liberando memória RAM.") 

34 

35 

36# Instancia a aplicação FastAPI injetando o Lifespan 

37app = FastAPI( 

38 title="Telco Churn API", 

39 description="API de Inferência de Churn baseada em Redes Neurais Densa (PyTorch)", 

40 version="1.0.0", 

41 lifespan=lifespan, 

42) 

43 

44 

45# Cria o Router da Versão 1 (v1) 

46v1_router = APIRouter(prefix="/v1") 

47 

48 

49@app.get("/health", tags=["Monitoring"]) 

50def health_check(): 

51 """ 

52 Rota de monitoramento de saúde do microsserviço. 

53 Utilizada por Load Balancers e Kubernetes para atestar se a API está viva. 

54 """ 

55 return {"status": "healthy", "service": "telco-churn-api"} 

56 

57 

58@v1_router.post("/predict", tags=["Inference"]) 

59def predict(customer: CustomerRequest): 

60 """ 

61 Rota de Predição em Tempo Real (Versão 1). 

62 Recebe o JSON do cliente (validado pelo Pydantic), injeta no modelo (Pandera + PyTorch), 

63 e retorna o booleano de churn considerando o Custo Financeiro (Threshold 30%). 

64 """ 

65 try: 

66 # Pydantic converte a entrada em dicionário python nativo 

67 raw_dict = customer.model_dump() 

68 

69 # Envia para o motor de inferência que desenhamos na Etapa 3.1 

70 resultado = predict_churn(raw_dict) 

71 return JSONResponse(content=resultado) 

72 

73 except ValueError as ve: 

74 # Erros mapeados (ex: Schema Pandera falhou) viram Bad Request ou Unprocessable Entity 

75 logger.warning(f"Requisição rejeitada por validação do Motor ML: {ve}") 

76 raise HTTPException(status_code=422, detail=str(ve)) 

77 except Exception as e: 

78 # Erros não mapeados (ex: PyTorch quebrou a matriz) viram Internal Server Error 500 

79 logger.error(f"Internal Server Error no Forward Pass: {e}") 

80 raise HTTPException(status_code=500, detail="Erro interno no Motor de Predição") 

81 

82 

83# Registra o Router v1 no App Principal 

84app.include_router(v1_router)