File size: 2,266 Bytes
ab80adf | 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 | 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) |