src/app/app.py

from fastapi import FastAPI
import uvicorn
from datetime import datetime
from typing import Annotated
import os
import sys
import datetime
import pandas as pd

sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from src.utils import load_file, make_predcition, date_extracts





# Create an instance of FastAPI
app = FastAPI(debug=True)

# get absolute path
DIRPATH = os.path.dirname(os.path.realpath(__file__))

# set path for ml files
ml_contents_path = os.path.join(DIRPATH, '..', 'assets', 'ml_components', 'toolkit_folder')

# get contents
ml_contents = load_file(ml_contents_path)

Encoder = ml_contents["OneHotEncoder"]
model = ml_contents["model"]
features_ = ml_contents['feature_names']



# define endpoints

@app.get('/')
def root():
    return 'Welcome to the Gorecery Sales Forecasting API'

@app.get('/health')
def check_health():
    return {'status': 'ok'}

@app.post('/predict')
async def predict_sales( store_id: int, category_id: int, onpromotion: int,
                  city: str, store_type: int, cluster: int, date_: Annotated[datetime.date, "The date of sales"] = datetime.date.today()):

    # create a dictionary of inputs
    input = {
    'store_id':[store_id], 
    'category_id':[category_id], 
    'onpromotion' :[onpromotion],
    'type' : [store_type], 
    'cluster': [cluster],
    'city' : [city],
    'date_': [date_]
    }   

    # convert to dataframe and extract datetime features
    input_data = pd.DataFrame(input)
    date_extracts(input_data)


    # make prediction
    sales = make_predcition(Encoder, model, input)
    sales_value = float(sales[0])
    return {'sales': sales_value}



if __name__ == "__main__":
    uvicorn.run('app:app', reload=True)