api.py

import mlflow 
import uvicorn
import pandas as pd 
from pydantic import BaseModel
from typing import Literal, List, Union
from fastapi import FastAPI, File, UploadFile
import joblib

# Log model from mlflow 
logged_model = 'runs:/.../model'

# Load model as a PyFuncModel.
loaded_model = mlflow.pyfunc.load_model(logged_model)

tags_metadata = [
    {
        "name": "Machine Learning",
        "description": "Prediction Endpoint."
    }
]

app = FastAPI(
    title="Car price prediction API",
    openapi_tags=tags_metadata
)

class PredictionFeatures(BaseModel):
    model_key: str
    mileage: int
    engine_power: int
    fuel: 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

@app.get("/", tags=["Introduction Endpoints"])
async def index():
    """
    Simply returns a welcome message!
    """
    message = "Hello world! This `/` is the most simple and default endpoint. If you want to learn more, check out documentation of the api at `/docs`"
    return message


@app.post("/predict", tags=["Machine Learning"])
async def predict(predictionFeatures: PredictionFeatures):
    # Read data 
    input_data = pd.DataFrame({
        "model_key": [predictionFeatures.model_key],
        "mileage": [predictionFeatures.mileage],
        "engine_power": [predictionFeatures.engine_power],
        "fuel": [predictionFeatures.fuel],
        "car_type": [predictionFeatures.car_type],
        "private_parking_available": [predictionFeatures.private_parking_available],
        "has_gps": [predictionFeatures.has_gps],
        "has_air_conditioning": [predictionFeatures.has_air_conditioning],
        "automatic_car": [predictionFeatures.automatic_car],
        "has_getaround_connect": [predictionFeatures.has_getaround_connect],
        "has_speed_regulator": [predictionFeatures.has_speed_regulator],
        "winter_tires": [predictionFeatures.winter_tires]
    })

    prediction = loaded_model.predict(input_data)

    # Format response
    response = {"prediction": prediction.tolist()[0]}
    return response


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)