api.py

from fastapi import FastAPI
import pandas as pd
import joblib
from pydantic import BaseModel
from typing import List
import os
import boto3

app = FastAPI()

MODEL_LOCAL_PATH = "/tmp/rfr_tuned.joblib"

def download_model_from_s3():
    bucket = os.getenv("S3_BUCKET")
    model_key = os.getenv("S3_MODEL_KEY")
    region = os.getenv("AWS_DEFAULT_REGION")

    s3_client = boto3.client("s3", region_name=region)
    s3_client.download_file(bucket, model_key, MODEL_LOCAL_PATH)

download_model_from_s3()
model = joblib.load(MODEL_LOCAL_PATH)

@app.get("/")
async def root():
    return {"message": "Api is running",
            "docs": "/docs"}

class Health(BaseModel):
    status: str
    model_loaded: bool


class CarFeatures(BaseModel):
    model_key: str
    mileage: int
    engine_power: int
    fuel: str
    paint_color: str
    car_type: str
    private_parking_available: bool
    has_gps: bool
    has_air_conditioning: bool
    automatic_car: bool
    has_getaround_connect: bool
    has_speed_regulator: bool
    winter_tires: bool


class PredictRequest(BaseModel):
    input: List[CarFeatures]

@app.post("/predict")
async def predict(payload: PredictRequest):
    records = [item.model_dump() for item in payload.input]
    df = pd.DataFrame(records)
    predictions = model.predict(df)

    return {"prediction": predictions.tolist()}