API

churn_prediction.api.app

FastAPI application entry point for the Churn Prediction Batch Inference API.

This module initializes the FastAPI application, configures global middleware, and manages the application lifecycle—specifically the loading and unloading of ML artifacts (OrdinalEncoder and XGBoost Booster) into the application state.

create_app()

Create and configure the FastAPI application.

Initializes the application metadata, lifecycle management, CORS configuration, and API routers.

Returns:

Type	Description
FastAPI	Fully configured FastAPI application instance ready to serve prediction, explainability, download, and health endpoints.

Source code in src/churn_prediction/api/app.py

def create_app() -> FastAPI:
    """Create and configure the FastAPI application.

    Initializes the application metadata, lifecycle management, CORS
    configuration, and API routers.

    Returns
    -------
    fastapi.FastAPI
        Fully configured FastAPI application instance ready to serve
        prediction, explainability, download, and health endpoints.
    """
    logger = get_logger()
    logger.info("Creating FastAPI application instance...")

    # Fetch version from pyproject.toml via installed package metadata
    try:
        app_version = metadata.version("churn-prediction")
    except metadata.PackageNotFoundError:
        app_version = "0.1.0-dev"
        logger.warning("Package version not found; defaulting to %s", app_version)

    app = FastAPI(
        title="Churn Prediction Batch Inference API",
        description=(
            "This API provides high-performance batch inference for customer churn modeling. "
            "It utilizes a pre-trained XGBoost model and Scikit-Learn preprocessors to "
            "predict the likelihood of customer attrition based on behavioral data."
        ),
        version=app_version,
        lifespan=lifespan,
        docs_url="/docs",
        redoc_url="/redoc",
    )

    # Configure CORS middleware to allow requests from any origin
    logger.debug("Configuring CORSMiddleware with allow_origins=['*']")
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=False,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Register routers for health checks and prediction endpoints
    logger.info("Registering API routers: health, predict, explain")
    app.include_router(health.router)
    app.include_router(predict.router)
    app.include_router(explain.router)

    logger.info("FastAPI application created successfully.")
    return app

lifespan(app) async

Manage startup and shutdown of the FastAPI application.

During startup, loads the fitted preprocessing encoder and trained XGBoost model from the configured artifact directory and stores them in app.state for dependency injection.

During shutdown, clears the references from application state so that model artifacts can be released from memory.

Parameters:

Name	Type	Description	Default
app	FastAPI	FastAPI application instance whose lifecycle is being managed.	required

Yields:

Type	Description
None	Control is yielded to the running FastAPI application after all required ML artifacts have been loaded.

Raises:

Type	Description
FileNotFoundError	If the required model, preprocessing artifact, or demonstration dataset cannot be found.
Exception	If an ML artifact cannot be loaded successfully.

Source code in src/churn_prediction/api/app.py

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    """Manage startup and shutdown of the FastAPI application.

    During startup, loads the fitted preprocessing encoder and trained
    XGBoost model from the configured artifact directory and stores them in
    ``app.state`` for dependency injection.

    During shutdown, clears the references from application state so that
    model artifacts can be released from memory.

    Parameters
    ----------
    app : fastapi.FastAPI
        FastAPI application instance whose lifecycle is being managed.

    Yields
    ------
    None
        Control is yielded to the running FastAPI application after all
        required ML artifacts have been loaded.

    Raises
    ------
    FileNotFoundError
        If the required model, preprocessing artifact, or demonstration
        dataset cannot be found.
    Exception
        If an ML artifact cannot be loaded successfully.
    """
    logger = get_logger()
    settings = get_settings()

    # Define paths to required ML artifacts
    encoder_path = settings.MODEL_DIR / "preprocessor.joblib"
    booster_path = settings.MODEL_DIR / "model.ubj"
    card_path = settings.MODEL_DIR / "card.json"
    required_artifacts = [encoder_path, booster_path, card_path]

    missing_artifacts = [str(artifact) for artifact in required_artifacts if not artifact.is_file()]
    if missing_artifacts:
        logger.critical("Missing artifacts: %s", ", ".join(missing_artifacts))
        raise FileNotFoundError("Required artifacts are missing.")

    # Load the pre-fitted encoder and XGBoost model
    try:
        logger.info("Loading preprocessor artifact (joblib)...")
        encoder: OrdinalEncoder = joblib.load(encoder_path)

        logger.info("Loading XGBoost booster model...")
        booster = xgb.Booster()
        booster.load_model(str(booster_path))

        app.state.encoder = encoder
        app.state.booster = booster
        logger.info("Successfully loaded all ML artifacts into memory.")
    except Exception as err:
        logger.error("Critical error loading ML artifacts: %s", err)
        raise

    yield

    logger.info("Shutting down application. Clearing app state...")
    app.state.encoder = None
    app.state.booster = None
    logger.debug("Application state cleared.")

churn_prediction.api.dependencies

Model and Encoder Dependency Providers.

This module contains utility functions used as FastAPI dependencies to retrieve the preprocessor (OrdinalEncoder) and the predictive model (XGBoost Booster) from the application state. It ensures that these artifacts are properly loaded before being used in request handlers, raising HTTP 503 errors if they are missing.

get_encoder(request)

Retrieve the fitted encoder from FastAPI application state.

Parameters:

Name	Type	Description	Default
request	Request	Incoming FastAPI request providing access to the application state.	required

Returns:

Type	Description
OrdinalEncoder	Fitted ordinal encoder loaded during application startup.

Raises:

Type	Description
HTTPException	HTTP 503 error if the encoder has not been loaded into application state.

Source code in src/churn_prediction/api/dependencies.py

def get_encoder(request: Request) -> OrdinalEncoder:
    """Retrieve the fitted encoder from FastAPI application state.

    Parameters
    ----------
    request : fastapi.Request
        Incoming FastAPI request providing access to the application state.

    Returns
    -------
    sklearn.preprocessing.OrdinalEncoder
        Fitted ordinal encoder loaded during application startup.

    Raises
    ------
    fastapi.HTTPException
        HTTP 503 error if the encoder has not been loaded into application
        state.
    """
    encoder: OrdinalEncoder | None = getattr(request.app.state, "encoder", None)
    if encoder is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Encoder artifact is not loaded in memory.",
        )
    return encoder

get_model(request)

Retrieve the trained XGBoost model from application state.

Parameters:

Name	Type	Description	Default
request	Request	Incoming FastAPI request providing access to the application state.	required

Returns:

Type	Description
Booster	Trained XGBoost booster loaded during application startup.

Raises:

Type	Description
HTTPException	HTTP 503 error if the model has not been loaded into application state.

Source code in src/churn_prediction/api/dependencies.py

def get_model(request: Request) -> xgb.Booster:
    """Retrieve the trained XGBoost model from application state.

    Parameters
    ----------
    request : fastapi.Request
        Incoming FastAPI request providing access to the application state.

    Returns
    -------
    xgboost.Booster
        Trained XGBoost booster loaded during application startup.

    Raises
    ------
    fastapi.HTTPException
        HTTP 503 error if the model has not been loaded into application
        state.
    """
    model: xgb.Booster | None = getattr(request.app.state, "booster", None)
    if model is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="XGBoost model artifact is not loaded in memory.",
        )
    return model

churn_prediction.api.routes.predict

Inference API Module.

This module defines the FastAPI router and endpoints for generating batch predictions using a pre-trained XGBoost model and a scikit-learn OrdinalEncoder. It handles CSV file uploads, converts the incoming CSV payload into an optimized Parquet file on disk asynchronously, triggers the prediction pipeline, and provides a mechanism to download prediction artifacts.

PredictionResponse

Bases: BaseModel

Response model for the prediction endpoint.

download_predictions(batch_id) async

Return the generated prediction CSV for a batch.

Locates the prediction artifact associated with batch_id and returns it as a downloadable HTTP file response.

Parameters:

Name	Type	Description	Default
batch_id	str	Unique prediction batch identifier used to locate the output artifact.	required

Returns:

Type	Description
FileResponse	HTTP response containing the generated prediction CSV.

Raises:

Type	Description
HTTPException	HTTP 400 if batch_id violates the expected identifier format or represents an unsafe path. HTTP 404 if the corresponding prediction artifact does not exist.

Source code in src/churn_prediction/api/routes/predict.py

@router.get(
    "/predictions/download/{batch_id}",
    response_class=FileResponse,
    status_code=status.HTTP_200_OK,
    summary="Download Prediction Results",
    description=(
        "Fetch the generated predictions CSV file using the filename provided as a query parameter. "
        "Ensures that the requested file exists within the secure predictions directory "
        "and prevents directory traversal attacks."
    ),
    responses={
        status.HTTP_200_OK: {"description": "CSV file successfully retrieved"},
        status.HTTP_400_BAD_REQUEST: {"description": "Invalid file path or security violation"},
        status.HTTP_404_NOT_FOUND: {"description": "The requested prediction file was not found"},
    },
)
async def download_predictions(
    batch_id: Annotated[
        str,
        PathParam(
            ...,
            description="Unique execution batch identifier",
            pattern=BATCH_ID_PATTERN,
            min_length=1,
            max_length=128,
        ),
    ],
) -> FileResponse:
    """Return the generated prediction CSV for a batch.

    Locates the prediction artifact associated with ``batch_id`` and returns
    it as a downloadable HTTP file response.

    Parameters
    ----------
    batch_id : str
        Unique prediction batch identifier used to locate the output artifact.

    Returns
    -------
    fastapi.responses.FileResponse
        HTTP response containing the generated prediction CSV.

    Raises
    ------
    fastapi.HTTPException
        HTTP 400 if ``batch_id`` violates the expected identifier format or
        represents an unsafe path.
        HTTP 404 if the corresponding prediction artifact does not exist.
    """
    logger = get_logger()
    settings = get_settings()

    logger.debug("Download artifact requested for batch_id: %s", batch_id)

    prediction_dir = Path(settings.PREDICTION_DATA_DIR).expanduser().resolve()
    target_filename = f"prediction_{batch_id}.csv"
    file_path = (prediction_dir / target_filename).resolve()

    if not file_path.is_relative_to(prediction_dir):
        logger.warning("Attempted directory traversal attack: %s", file_path)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file path")

    if not file_path.is_file():
        logger.error("Requested prediction download artifact not found: %s", file_path)
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Prediction artifact for batch '{batch_id}' not found.",
        )

    logger.info("Serving prediction file: %s", file_path)
    return FileResponse(
        path=file_path,
        media_type="text/csv",
        filename=target_filename,
    )

predict_batch(file, encoder, model) async

Process an uploaded dataset and initiate batch churn prediction.

Accepts a customer CSV upload, persists the input data, executes the prediction pipeline using the pre-loaded encoder and XGBoost model, and returns metadata identifying the generated prediction batch.

Parameters:

Name	Type	Description	Default
file	UploadFile	Uploaded CSV file containing customer records to score.	required
encoder	OrdinalEncoder	Pre-fitted encoder used to transform categorical input features.	required
model	Booster	Trained XGBoost booster used to generate churn predictions.	required

Returns:

Type	Description
PredictionResponse	Response containing the batch identifier, original filename, processing status, number of records, and prediction download URL.

Raises:

Type	Description
HTTPException	If the uploaded file is invalid, cannot be persisted, or the prediction pipeline fails.
ValueError	If CSV-to-Parquet conversion or dataset preprocessing fails.
ParserError	If the uploaded CSV cannot be parsed.

Source code in src/churn_prediction/api/routes/predict.py

@router.post(
    "/predict/batch",
    response_model=PredictionResponse,
    status_code=status.HTTP_200_OK,
    summary="Execute Batch Prediction Job",
    description=(
        "Upload a CSV file containing feature records to generate churn predictions. "
        "The system validates the file format, persists it to storage, and executes "
        "the prediction pipeline using a pre-loaded XGBoost model and OrdinalEncoder. "
        "Returns a `batch_id` which can be used to download the results."
    ),
    responses={
        status.HTTP_200_OK: {"description": "Predictions successfully generated"},
        status.HTTP_400_BAD_REQUEST: {"description": "Invalid file format provided"},
        status.HTTP_422_UNPROCESSABLE_CONTENT: {
            "description": "Pipeline failed during feature transformation or inference"
        },
        status.HTTP_500_INTERNAL_SERVER_ERROR: {
            "description": "Internal server error while persisting file"
        },
    },
)
async def predict_batch(
    file: Annotated[UploadFile, File(description="CSV File containing tabular feature records")],
    encoder: Annotated[OrdinalEncoder, Depends(get_encoder)],
    model: Annotated[xgb.Booster, Depends(get_model)],
) -> PredictionResponse:
    """Process an uploaded dataset and initiate batch churn prediction.

    Accepts a customer CSV upload, persists the input data, executes the
    prediction pipeline using the pre-loaded encoder and XGBoost model, and
    returns metadata identifying the generated prediction batch.

    Parameters
    ----------
    file : fastapi.UploadFile
        Uploaded CSV file containing customer records to score.
    encoder : sklearn.preprocessing.OrdinalEncoder
        Pre-fitted encoder used to transform categorical input features.
    model : xgboost.Booster
        Trained XGBoost booster used to generate churn predictions.

    Returns
    -------
    PredictionResponse
        Response containing the batch identifier, original filename,
        processing status, number of records, and prediction download URL.

    Raises
    ------
    fastapi.HTTPException
        If the uploaded file is invalid, cannot be persisted, or the
        prediction pipeline fails.
    ValueError
        If CSV-to-Parquet conversion or dataset preprocessing fails.
    pandas.errors.ParserError
        If the uploaded CSV cannot be parsed.
    """
    logger = get_logger()
    settings = get_settings()

    # Generate a unique batch identifier for this prediction job
    batch_id = f"{datetime.now(UTC):%Y%m%dT%H%M%S}_{uuid4().hex[:4]}"
    original_filename = Path(file.filename).name if file.filename else "upload.csv"
    logger.info(
        "Received batch prediction request. Batch ID: %s, Filename: %s",
        batch_id,
        original_filename,
    )

    # Validate the uploaded file format
    if not original_filename.lower().endswith(".csv"):
        logger.warning("Invalid file upload attempt: %s", file.filename)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid file format. Please upload a valid CSV file (.csv).",
        )

    # Persist the uploaded CSV file to the designated raw data directory
    raw_file_dir = settings.RAW_DATA_DIR.expanduser().resolve()
    clean_stem = Path(original_filename).stem
    raw_file_path = (raw_file_dir / f"{clean_stem}_{batch_id}.parquet").resolve()

    logger.debug("Persisting uploaded file to: %s", raw_file_path)
    try:
        await run_in_threadpool(
            _to_parquet,
            file_obj=file.file,
            output_path=raw_file_path,
        )
        logger.debug("Successfully persisted file to disk.")
    except (pd.errors.EmptyDataError, pd.errors.ParserError, ValueError) as parse_err:
        logger.warning("Corrupted or malformed CSV upload: %s", parse_err)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Failed to parse CSV payload: {parse_err}",
        ) from parse_err
    except Exception as err:
        logger.error("Failed to write uploaded file to disk: %s", err)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to persist uploaded payload: {err}",
        ) from err
    finally:
        await file.close()

    # Run the prediction pipeline in a separate thread to avoid blocking the event loop
    try:
        logger.info("Triggering prediction pipeline for batch %s", batch_id)
        await run_in_threadpool(
            run_prediction_pipeline,
            batch_id=batch_id,
            raw_file_path=raw_file_path,
            loaded_encoder=encoder,
            loaded_booster=model,
        )
        logger.info("Prediction pipeline completed successfully for batch %s", batch_id)
    except Exception as pipeline_err:
        logger.exception("Batch prediction execution failed: %s", pipeline_err)
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"Batch execution pipeline failed: {pipeline_err}",
        ) from pipeline_err

    return PredictionResponse(
        batch_id=batch_id,
        filename=original_filename,
        status="SUCCESS",
        download_url=f"/api/v1/predictions/download/{batch_id}",
    )

churn_prediction.api.routes.explain

Explainability API Module.

This module provides endpoints to generate model interpretability reports using SHAP (SHapley Additive exPlanations) values for the churn prediction model. It integrates with the XGBoost booster and processed datasets to visualize feature contributions.

ExplainabilityResponse

Bases: BaseModel

Response model for the explainability endpoint.

generate_explanation(batch_id, xgb_booster) async

Generate SHAP explainability artifacts for a prediction batch.

Loads the processed input features associated with the specified batch, computes SHAP explanations using the supplied XGBoost booster, and persists dashboard-ready explainability artifacts.

Parameters:

Name	Type	Description	Default
batch_id	str	Unique identifier of the prediction batch to explain.	required
xgb_booster	Booster	Trained XGBoost model used to calculate SHAP contributions.	required

Returns:

Type	Description
ExplainabilityResponse	Response containing the batch identifier and location of the generated explainability artifacts.

Raises:

Type	Description
HTTPException	HTTP 404 if the batch input artifact does not exist. HTTP 500 if data loading or SHAP artifact generation fails.

Source code in src/churn_prediction/api/routes/explain.py

@router.post(
    "/explain/{batch_id}",
    response_model=ExplainabilityResponse,
    status_code=status.HTTP_200_OK,
    summary="Generate SHAP Explainability Plots",
    description=(
        "Generates SHAP explainability plots for the churn prediction model using "
        "the processed dataset corresponding to the current session. The plots are "
        "saved to disk, and the endpoint returns the path to the generated plots."
    ),
    responses={
        status.HTTP_200_OK: {
            "description": "Successfully generated SHAP explainability plots.",
            "model": ExplainabilityResponse,
        },
        status.HTTP_404_NOT_FOUND: {
            "description": "Processed data file for the current session not found.",
        },
        status.HTTP_500_INTERNAL_SERVER_ERROR: {
            "description": "An unexpected error occurred during plot generation.",
        },
    },
)
async def generate_explanation(
    batch_id: Annotated[
        str,
        PathParam(
            ...,
            description="Unique identifier for the target processing batch",
            pattern=BATCH_ID_PATTERN,
            min_length=1,
            max_length=128,
        ),
    ],
    xgb_booster: Annotated[xgb.Booster, Depends(get_model)],
) -> ExplainabilityResponse:
    """Generate SHAP explainability artifacts for a prediction batch.

    Loads the processed input features associated with the specified batch,
    computes SHAP explanations using the supplied XGBoost booster, and
    persists dashboard-ready explainability artifacts.

    Parameters
    ----------
    batch_id : str
        Unique identifier of the prediction batch to explain.
    xgb_booster : xgboost.Booster
        Trained XGBoost model used to calculate SHAP contributions.

    Returns
    -------
    ExplainabilityResponse
        Response containing the batch identifier and location of the
        generated explainability artifacts.

    Raises
    ------
    fastapi.HTTPException
        HTTP 404 if the batch input artifact does not exist.
        HTTP 500 if data loading or SHAP artifact generation fails.
    """
    # Initialize logger and settings
    logger = get_logger()
    settings = get_settings()

    base_processed_dir = Path(settings.PROCESSED_DATA_DIR).expanduser().resolve()
    base_artifacts_dir = Path(settings.SHAP_REPORT_DIR).expanduser().resolve()

    processed_filepath = (base_processed_dir / f"processed_{batch_id}.parquet").resolve()
    logger.debug("Resolved processed_filepath: %s", processed_filepath)

    artifacts_dir = (base_artifacts_dir / batch_id).resolve()
    logger.debug("Resolved artifacts_dir: %s", artifacts_dir)

    try:
        # Load the processed data
        if not processed_filepath.exists():
            logger.error("Processed data file not found: %s", processed_filepath)
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Processed data for batch {batch_id} not found at {processed_filepath}",
            )

        # Load the processed data into a DataFrame
        logger.debug("Loading processed data from %s", processed_filepath)
        df = pd.read_parquet(processed_filepath)
        logger.info("Loaded dataframe with shape: %s", df.shape)

        # Generate SHAP explainability plots
        logger.info("Generating SHAP plots for batch: %s", batch_id)
        generate_shap_artifacts(batch_id, df, xgb_booster, artifacts_dir)

        logger.info("Successfully generated SHAP plots at: %s", artifacts_dir)

        return ExplainabilityResponse(
            batch_id=batch_id,
            status="success",
            artifacts_path=str(artifacts_dir),
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(
            "Unexpected error occurred while generating SHAP plots for batch %s: %s",
            batch_id,
            str(e),
        )
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"An unexpected error occurred while generating SHAP plots for batch {batch_id}: {e!s}",
        ) from e

churn_prediction.api.routes.health

Health Check API Module.

This module provides a FastAPI router with endpoints to monitor the operational status of the API. It verifies the availability of critical application state dependencies, specifically the encoder and the machine learning model, to ensure the service is ready to handle requests.

HealthResponse

Bases: BaseModel

Response model for the health check endpoint.

check_health(request) async

Report API and machine-learning dependency health.

Verifies that the preprocessing encoder and XGBoost model are available in application state and reports the current process memory usage and health-check timestamp.

Parameters:

Name	Type	Description	Default
request	Request	Incoming FastAPI request used to inspect application state.	required

Returns:

Type	Description
HealthResponse	Health information containing API status, encoder availability, model availability, memory utilization, and timestamp.

Source code in src/churn_prediction/api/routes/health.py

@router.get(
    "/health",
    response_model=HealthResponse,
    status_code=status.HTTP_200_OK,
    summary="System Health Check",
    description=(
        "Performs a diagnostic check of the API's operational state. "
        "This endpoint verifies that critical ML dependencies (encoder and booster model) "
        "are successfully loaded into memory and reports current system resource utilization. "
        "It is primarily used by orchestrators (e.g., Kubernetes) for liveness and readiness probes."
    ),
    responses={
        status.HTTP_200_OK: {
            "description": "Health check completed successfully. Status may be 'HEALTHY' or 'DEGRADED'.",
            "model": HealthResponse,
        },
    },
)
async def check_health(request: Request) -> HealthResponse:
    """Report API and machine-learning dependency health.

    Verifies that the preprocessing encoder and XGBoost model are available
    in application state and reports the current process memory usage and
    health-check timestamp.

    Parameters
    ----------
    request : fastapi.Request
        Incoming FastAPI request used to inspect application state.

    Returns
    -------
    HealthResponse
        Health information containing API status, encoder availability,
        model availability, memory utilization, and timestamp.
    """
    # Check if the encoder and model are loaded in the application state
    encoder_loaded = getattr(request.app.state, "encoder", None) is not None
    model_loaded = getattr(request.app.state, "booster", None) is not None
    is_healthy = encoder_loaded and model_loaded

    return HealthResponse(
        status="HEALTHY" if is_healthy else "DEGRADED",
        encoder_loaded=encoder_loaded,
        model_loaded=model_loaded,
        memory_usage_percent=psutil.virtual_memory().percent,
        timestamp=datetime.now(UTC).isoformat(),
    )