File size: 5,161 Bytes
bfde63f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | import os
import sys
from typing import Dict, Any, List, Optional
from fastapi import FastAPI, HTTPException, Security, Depends, status
from fastapi.security import APIKeyHeader
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
# Ensure project root is in sys.path
PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__))
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
from ml.inference.unified_inference import inference_service
# Configuration: Live API Key
LIVE_API_KEY = os.getenv("MODEL_API_KEY", "c360_live_key_9f8a2b7c4e1d")
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def verify_api_key(api_key: Optional[str] = Depends(api_key_header)):
if not api_key or api_key != LIVE_API_KEY:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unauthorized: Invalid or missing 'X-API-Key' header. Pass X-API-Key in request headers."
)
return api_key
# Initialize FastAPI App for Live Model
app = FastAPI(
title="Customer360 AI -- Live Model REST API",
description="Live Machine Learning Model Inference Service secured with API Key authentication.",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# Enable CORS for cross-origin requests from teammates
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Input Pydantic Schema
class CustomerInputSchema(BaseModel):
customer_id: str = Field("C1001", description="Unique customer identifier")
name: Optional[str] = Field("Sample Customer", description="Customer full name")
recency_days: int = Field(30, description="Days since last purchase")
frequency_purchases: int = Field(5, description="Total purchases in 12m")
monetary_total_spend: float = Field(15000.0, description="Lifetime spend in INR")
avg_order_value: Optional[float] = Field(3000.0, description="Average order value")
website_visits_30d: int = Field(10, description="Website visits in 30d")
product_views_30d: int = Field(20, description="Product views in 30d")
cart_additions_30d: int = Field(5, description="Cart additions in 30d")
cart_abandonments_30d: int = Field(2, description="Abandoned carts in 30d")
email_opens_30d: int = Field(5, description="Email opens in 30d")
email_clicks_30d: int = Field(2, description="Email clicks in 30d")
support_tickets_30d: int = Field(1, description="Support tickets in 30d")
engagement_change_pct: float = Field(0.0, description="Engagement change %")
return_rate: Optional[float] = Field(0.05, description="Return rate %")
discount_dependency_ratio: Optional[float] = Field(0.25, description="Discount dependency")
spend_30d: Optional[float] = Field(3000.0, description="30d spend")
spend_90d: Optional[float] = Field(9000.0, description="90d spend")
purchase_velocity: Optional[float] = Field(20.0, description="Days between purchases")
preferred_channel: Optional[str] = Field("WhatsApp", description="WhatsApp, Email, Push, SMS")
preferred_category: Optional[str] = Field("Electronics", description="Product category")
@app.get("/")
def read_root():
return {
"service": "Customer360 AI Live Model API",
"status": "online",
"authentication": "API Key Required in 'X-API-Key' header",
"endpoints": {
"predict": "/api/v1/predict (POST)",
"batch_predict": "/api/v1/batch-predict (POST)",
"health": "/health (GET)",
"docs": "/docs (GET)"
}
}
@app.get("/health")
def check_health():
return {
"status": "healthy",
"model_loaded": inference_service.loaded,
"api_key_configured": True
}
@app.post("/api/v1/predict", dependencies=[Depends(verify_api_key)])
def predict_single_customer(payload: CustomerInputSchema):
"""
Live ML Inference Endpoint.
Requires 'X-API-Key' in HTTP request headers.
"""
customer_dict = payload.model_dump()
result = inference_service.predict_customer_decision(customer_dict)
return result
@app.post("/api/v1/batch-predict", dependencies=[Depends(verify_api_key)])
def predict_batch_customers(payload: List[CustomerInputSchema]):
"""
Batch ML Inference Endpoint for multiple customers.
Requires 'X-API-Key' in HTTP request headers.
"""
results = []
for item in payload:
cdict = item.model_dump()
res = inference_service.predict_customer_decision(cdict)
results.append(res)
return {"total_processed": len(results), "predictions": results}
if __name__ == "__main__":
import uvicorn
print("=" * 65)
print(" CUSTOMER360 AI -- LIVE MODEL REST API SERVER WITH API KEY")
print("=" * 65)
print(f" LIVE MODEL API KEY : {LIVE_API_KEY}")
print(" HEADER NAME : X-API-Key")
print(" SERVER URL : http://localhost:8000")
print(" DOCUMENTATION : http://localhost:8000/docs")
print("=" * 65)
uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True)
|