develop API using fastapi and pydantic. Tested locally

api.yaml

@@ -0,0 +1,5 @@
+logged_model: "runs:/c073714a85234f378aae52da9771b8e1/woelr"
+credit_score_scaling:
+  pdo: 20
+  odds: 1
+  scorecard_points: 500

credit_score_mlops/api/inference.py

@@ -0,0 +1,89 @@
+"""
+a module for data preprocessing.
+"""
+
+import os
+
+import mlflow.sklearn
+import pandas as pd
+from dotenv import find_dotenv, load_dotenv
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+from credit_score_mlops.config import (
+    CREDIT_SCORE_SCALING_PARAMS,
+    GLOBAL_API,
+    MLFLOW_PARAMS,
+)
+from credit_score_mlops.credit_score import CreditScoreScaling
+
+load_dotenv(find_dotenv())
+
+app = FastAPI()
+
+
+class LoanApplicantData(BaseModel):
+    person_age: int
+    person_income: int
+    person_home_ownership: str
+    person_emp_length: float
+    loan_intent: str
+    loan_grade: str
+    loan_amnt: int
+    loan_int_rate: float
+    loan_percent_income: float
+    cb_person_default_on_file: str
+    cb_person_cred_hist_length: int
+
+
+class CreditScore(BaseModel):
+    credit_score: int
+
+
+def prepare_credit_scorer():
+    # Credit Scorer
+    logged_model = GLOBAL_API.logged_model
+    remote_uri = MLFLOW_PARAMS.remote_uri
+    pdo = CREDIT_SCORE_SCALING_PARAMS.pdo
+    odds = CREDIT_SCORE_SCALING_PARAMS.odds
+    scorecard_points = CREDIT_SCORE_SCALING_PARAMS.scorecard_points
+
+    # 2. Access remote MLFlow Server on DagsHub
+    mlflow.set_tracking_uri(remote_uri)  # set dagshub as the remote URI
+
+    os.environ["MLFLOW_TRACKING_USERNAME"] = os.getenv(
+        "DAGSHUB_USER_NAME"
+    )  # set up credentials for accessing remote dagshub uri
+    os.environ["MLFLOW_TRACKING_PASSWORD"] = os.getenv(
+        "DAGSHUB_PASSWORD"
+    )  # set up credentials for accessing remote dagshub uri
+
+    # 3. Get Model for Remote URI
+    loaded_model = mlflow.sklearn.load_model(logged_model)
+
+    # 4. Integrate with the CreditScoreScaling class
+    credit_scorer = CreditScoreScaling(
+        pipeline=loaded_model.pipeline, pdo=pdo, odds=odds, scorecard_points=scorecard_points
+    )
+    return credit_scorer
+
+
+credit_scorer = prepare_credit_scorer()
+
+
+@app.post("/calculate_credit_score")
+async def calculate_credit_score(loan_applicant_data: LoanApplicantData) -> CreditScore:
+    """Calculate credit score based on the loan applicant data.
+
+    Args:
+        loan_applicant_data (LoanApplicantData): Load applicant data.
+
+    Returns:
+        CreditScore: Credits score results returned in dictionary format.
+    """
+    input_df = pd.DataFrame(
+        [loan_applicant_data.model_dump().values()],
+        columns=loan_applicant_data.model_dump().keys(),
+    )
+    credit_score = credit_scorer.calculate_credit_score(input_df)["credit_score"][0]
+    return {"credit_score": round(credit_score)}

credit_score_mlops/config.py

@@ -27,6 +27,8 @@
 FIGURES_DIR = REPORTS_DIR / "figures"
 
 # Read parameters
+GLOBAL_API = read_yaml(Path("api.yaml"))
+CREDIT_SCORE_SCALING_PARAMS = GLOBAL_API.credit_score_scaling
 GLOBAL_PARAMS = read_yaml(Path("params.yaml"))
 DATA_PREPROCESSING_PARAMS = GLOBAL_PARAMS.data_preprocessing
 TRAIN_PARAMS = GLOBAL_PARAMS.train

requirements.txt

@@ -7,6 +7,7 @@ dataclasses-json==0.6.7
 dvc-s3==3.2.0
 dvclive==3.47.0
 ensure==1.0.4
+fastapi==0.112.0
 flake8==7.1.0
 flufl-lock==8.1.0
 fusepy==3.0.1