Spaces:
Sleeping
Sleeping
File size: 1,803 Bytes
48c8b68 | 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 | from fastapi import APIRouter, Depends, HTTPException, Header, status
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.models.schemas import InputSchema, PredictionOutput
from app.models.models import PredictionLog
from app.services.ml_service import ml_service
from app.core.config import settings
router = APIRouter()
def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != settings.API_KEY:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key",
)
return x_api_key
@router.post("/predict", response_model=PredictionOutput)
def predict(input_data: InputSchema, db: Session = Depends(get_db), api_key: str = Depends(verify_api_key)):
# Convert Pydantic model to dict
data_dict = input_data.dict()
try:
# Make prediction
prediction, probability = ml_service.predict(data_dict)
# Log to Database
db_log = PredictionLog(
**data_dict,
prediction=prediction,
probability=probability
)
db.add(db_log)
db.commit()
db.refresh(db_log)
return PredictionOutput(prediction=prediction, probability=probability)
except Exception as e:
print(f"Error during prediction: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/health")
def health_check():
return {"status": "healthy"}
@router.get("/model/info")
def model_info(api_key: str = Depends(verify_api_key)):
model = ml_service.model
if not model:
return {"status": "Model not loaded"}
return {
"type": str(type(model)),
"params": model.get_params() if hasattr(model, "get_params") else "Unknown"
}
|