Table of Contents

FastAPI for ML APIs

Rumman Ansari September 16, 2026 70 views Subject Details

Model Deployment

FastAPI for ML APIs

Build a typed and production-oriented Machine Learning API using FastAPI, Pydantic validation, lifespan-based model loading, individual and batch inference, structured responses, health and readiness checks, automated documentation, testing, security controls, monitoring, and deployment best practices.

Introduction

FastAPI is a Python framework for building HTTP APIs using standard Python type hints. It integrates closely with Pydantic for request and response validation and automatically produces OpenAPI-based API documentation.

In a Machine Learning deployment, FastAPI can provide the interface between a client application and a saved model. The client submits structured input, FastAPI validates the request, an inference service applies the saved preprocessing and model pipeline, and the API returns a typed prediction response.

This lesson builds a customer-churn prediction API. The service loads an approved scikit-learn pipeline during application startup and provides health, readiness, metadata, individual prediction, and batch prediction endpoints.

Project goal: Create a maintainable FastAPI service that validates production input, reuses an approved Machine Learning pipeline, returns traceable predictions, documents its contract automatically, and handles failures safely.

Learning Objectives

By the End of This Lesson

Explain how FastAPI supports Machine Learning inference
Understand path operations and HTTP methods
Define typed request and response models
Validate numerical ranges and categorical values
Load a saved model during application startup
Use a lifespan function for model initialization
Separate route logic from inference logic
Create health and readiness endpoints
Create model-information endpoints
Create individual and batch prediction endpoints
Return probabilities, thresholds, labels, versions, and request identifiers
Define custom validation and service exceptions
Create consistent error-response models
Use automatic interactive API documentation
Test endpoints with FastAPI TestClient
Use environment-based configuration
Recognize blocking and asynchronous workloads
Apply authentication and authorization controls
Monitor latency, errors, traffic, and model behavior
Prepare the service for production deployment

Prerequisites

Recommended Knowledge

Python functions, classes, modules, and type hints
JSON objects and arrays
HTTP methods and status codes
pandas DataFrames
scikit-learn pipelines
Saving and loading models with Joblib
Input validation
Virtual environments
Command-line fundamentals
Basic automated testing

Required Packages

FastAPI
Uvicorn
Pydantic
pandas
NumPy
scikit-learn
Joblib
pytest
HTTPX

python -m pip install "fastapi[standard]" pandas numpy scikit-learn joblib pytest httpx

What Is FastAPI?

FastAPI maps HTTP paths and methods to Python functions called path operations. Python type annotations and Pydantic models define how incoming and outgoing data should be validated and documented.

FastAPI Concept	Meaning
Application	The FastAPI application object
Path Operation	A Python function associated with an HTTP path and method
Pydantic Model	A typed schema used for data validation and serialization
Dependency	Reusable logic provided to path operations through dependency injection
Lifespan	Startup and shutdown logic for application resources
Response Model	A declared schema for API output
OpenAPI	A standardized description of the API contract
ASGI	The interface between asynchronous Python applications and compatible servers

Flask vs FastAPI

Area	Flask	FastAPI
Application Interface	WSGI	ASGI
Validation	Usually added through application code or extensions	Typed validation through Pydantic integration
API Schema	Usually configured separately	Generated from declared paths and models
Interactive Documentation	Usually added separately	Generated from the OpenAPI contract
Asynchronous Functions	Not the framework's original design focus	Supported through ASGI and async path operations
Learning Use	Flexible and explicit web framework concepts	Typed API contracts and validation with less manual schema code

Selection principle: Choose a framework according to team skills, ecosystem, validation requirements, integrations, operational standards, and project constraints rather than popularity alone.

ML API Architecture

Prediction Flow

Client → Authentication → Pydantic Validation → Inference Service → Saved Pipeline → Response Model → Monitoring

Layer	Responsibility
Transport Layer	Receives HTTP requests and returns HTTP responses
Schema Layer	Defines typed request, response, and error contracts
Validation Layer	Rejects missing, malformed, unsupported, or unsafe values
Inference Layer	Applies the saved preprocessing and estimator
Decision Layer	Applies the approved threshold and class mapping
Security Layer	Controls identity, permissions, secrets, and traffic
Observability Layer	Tracks health, latency, errors, versions, and model behavior

Recommended Project Structure

fastapi-ml-api/
├── README.md
├── requirements.txt
├── .gitignore
├── artifacts/
│   ├── customer_churn_pipeline.joblib
│   └── model_metadata.json
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── configuration.py
│   ├── dependencies.py
│   ├── exceptions.py
│   ├── inference.py
│   ├── schemas.py
│   └── routers/
│       ├── __init__.py
│       ├── health.py
│       └── predictions.py
└── tests/
    ├── conftest.py
    ├── test_health.py
    ├── test_predictions.py
    └── test_validation.py

Project principle: Separate schemas, inference, configuration, routes, dependencies, exceptions, tests, and artifacts so each concern can be reviewed and tested independently.

Application Configuration

import os
from dataclasses import dataclass
from pathlib import Path


@dataclass(
    frozen=True
)
class Settings:

    base_directory: Path = Path(
        __file__
    ).resolve().parent.parent

    model_path: Path = Path(
        os.getenv(
            "MODEL_PATH",
            str(
                base_directory
                / "artifacts"
                / "customer_churn_pipeline.joblib"
            )
        )
    )

    metadata_path: Path = Path(
        os.getenv(
            "METADATA_PATH",
            str(
                base_directory
                / "artifacts"
                / "model_metadata.json"
            )
        )
    )

    maximum_batch_size: int = int(
        os.getenv(
            "MAXIMUM_BATCH_SIZE",
            "100"
        )
    )

    application_name: str = (
        "Customer Churn Prediction API"
    )

    application_version: str = "1.0.0"


settings = Settings()

Configuration rule: Use environment configuration for deployment-specific values. Never embed secrets directly in source code.

Model Metadata

{
    "model_name": "customer-churn-classifier",
    "model_version": "1.0.0",
    "decision_threshold": 0.6,
    "positive_class": "churned",
    "negative_class": "retained",
    "numerical_features": [
        "tenure_months",
        "monthly_charges",
        "total_charges"
    ],
    "categorical_features": [
        "contract_type",
        "payment_method",
        "tech_support"
    ],
    "intended_use": "Retention-review prioritization",
    "prohibited_use": "Automatic denial or reduction of customer service"
}

Define the Prediction Request

from typing import Literal

from pydantic import BaseModel
from pydantic import ConfigDict
from pydantic import Field
from pydantic import field_validator


class ChurnPredictionRequest(
    BaseModel
):

    model_config = ConfigDict(
        extra="forbid",
        str_strip_whitespace=True
    )

    tenure_months: int = Field(
        ge=0,
        le=1200,
        description=(
            "Number of months in the "
            "customer relationship"
        )
    )

    monthly_charges: float = Field(
        ge=0,
        description=(
            "Current recurring customer charge"
        )
    )

    total_charges: float = Field(
        ge=0,
        description=(
            "Historical charges available "
            "at prediction time"
        )
    )

    contract_type: Literal[
        "Monthly",
        "One year",
        "Two year"
    ]

    payment_method: Literal[
        "Electronic",
        "Bank transfer",
        "Credit card",
        "Mailed check"
    ]

    tech_support: Literal[
        "Yes",
        "No"
    ]

    @field_validator(
        "monthly_charges",
        "total_charges"
    )
    @classmethod
    def validate_finite_number(
        cls,
        value
    ):
        import math

        if not math.isfinite(
            value
        ):
            raise ValueError(
                "The value must be finite."
            )

        return value

Strict contract: Setting extra="forbid" rejects unexpected fields rather than silently ignoring them.

Define the Prediction Response

from pydantic import BaseModel
from pydantic import Field


class ChurnPredictionResponse(
    BaseModel
):

    predicted_class: str

    positive_class_probability: float = Field(
        ge=0,
        le=1
    )

    decision_threshold: float = Field(
        ge=0,
        le=1
    )

    retention_review_recommended: bool

    model_name: str

    model_version: str

    request_id: str

Response-model benefit: A declared response model documents the output contract and validates the structure returned by the path operation.

Define Batch Schemas

from pydantic import BaseModel
from pydantic import ConfigDict
from pydantic import Field


class BatchPredictionRequest(
    BaseModel
):

    model_config = ConfigDict(
        extra="forbid"
    )

    records: list[
        ChurnPredictionRequest
    ] = Field(
        min_length=1
    )


class BatchPredictionItem(
    BaseModel
):

    item_index: int

    predicted_class: str

    positive_class_probability: float = Field(
        ge=0,
        le=1
    )

    retention_review_recommended: bool


class BatchPredictionResponse(
    BaseModel
):

    predictions: list[
        BatchPredictionItem
    ]

    prediction_count: int

    decision_threshold: float

    model_name: str

    model_version: str

    request_id: str

Define Error Responses

from pydantic import BaseModel


class ErrorDetail(
    BaseModel
):

    code: str

    message: str


class ErrorResponse(
    BaseModel
):

    error: ErrorDetail

    request_id: str | None = None

Error Code	Purpose
`validation_error`	The request does not satisfy the input contract
`batch_too_large`	The request exceeds the approved batch limit
`model_not_ready`	The approved model is unavailable
`prediction_failed`	The inference operation failed
`internal_error`	An unexpected server error occurred

Create the Inference Service

import json
from pathlib import Path

import joblib
import pandas as pd


class ModelNotReadyError(
    RuntimeError
):
    pass


class PredictionError(
    RuntimeError
):
    pass


class ModelService:

    def __init__(
        self,
        model_path: Path,
        metadata_path: Path
    ):
        self.model_path = model_path

        self.metadata_path = metadata_path

        self.model = None

        self.metadata = None

    def load(
        self
    ) -> None:
        if not self.model_path.exists():

            raise FileNotFoundError(
                "The approved model artifact "
                "was not found."
            )

        if not self.metadata_path.exists():

            raise FileNotFoundError(
                "The model metadata file "
                "was not found."
            )

        with open(
            self.metadata_path,
            "r",
            encoding="utf-8"
        ) as metadata_file:

            self.metadata = json.load(
                metadata_file
            )

        self.model = joblib.load(
            self.model_path
        )

    def unload(
        self
    ) -> None:
        self.model = None

        self.metadata = None

    @property
    def ready(
        self
    ) -> bool:
        return (
            self.model is not None
            and self.metadata is not None
        )

    @property
    def required_features(
        self
    ) -> list[str]:
        if self.metadata is None:

            raise ModelNotReadyError(
                "Model metadata is unavailable."
            )

        return [
            *self.metadata[
                "numerical_features"
            ],
            *self.metadata[
                "categorical_features"
            ]
        ]

Add Individual Prediction Logic

def predict_one(
    self,
    record: dict
) -> dict:
    if not self.ready:

        raise ModelNotReadyError(
            "The model is not ready."
        )

    try:
        input_frame = pd.DataFrame(
            [
                record
            ]
        )[
            self.required_features
        ]

        probability = float(
            self.model.predict_proba(
                input_frame
            )[0, 1]
        )

    except Exception as error:

        raise PredictionError(
            "The model could not complete "
            "the prediction."
        ) from error

    threshold = float(
        self.metadata[
            "decision_threshold"
        ]
    )

    review_recommended = (
        probability
        >=
        threshold
    )

    predicted_class = (
        self.metadata[
            "positive_class"
        ]
        if review_recommended
        else self.metadata[
            "negative_class"
        ]
    )

    return {
        "predicted_class": predicted_class,
        "positive_class_probability": (
            probability
        ),
        "decision_threshold": threshold,
        "retention_review_recommended": (
            review_recommended
        ),
        "model_name": self.metadata[
            "model_name"
        ],
        "model_version": self.metadata[
            "model_version"
        ]
    }

Add Batch Prediction Logic

def predict_many(
    self,
    records: list[dict]
) -> list[dict]:
    if not self.ready:

        raise ModelNotReadyError(
            "The model is not ready."
        )

    try:
        input_frame = pd.DataFrame(
            records
        )[
            self.required_features
        ]

        probabilities = (
            self.model.predict_proba(
                input_frame
            )[:, 1]
        )

    except Exception as error:

        raise PredictionError(
            "The model could not complete "
            "the batch prediction."
        ) from error

    threshold = float(
        self.metadata[
            "decision_threshold"
        ]
    )

    predictions = []

    for item_index, probability in enumerate(
        probabilities
    ):
        probability = float(
            probability
        )

        review_recommended = (
            probability
            >=
            threshold
        )

        predicted_class = (
            self.metadata[
                "positive_class"
            ]
            if review_recommended
            else self.metadata[
                "negative_class"
            ]
        )

        predictions.append(
            {
                "item_index": item_index,
                "predicted_class": (
                    predicted_class
                ),
                "positive_class_probability": (
                    probability
                ),
                "retention_review_recommended": (
                    review_recommended
                )
            }
        )

    return predictions

Load the Model with Lifespan

from contextlib import asynccontextmanager

from fastapi import FastAPI


model_service = ModelService(
    model_path=settings.model_path,
    metadata_path=settings.metadata_path
)


@asynccontextmanager
async def lifespan(
    app: FastAPI
):
    model_service.load()

    app.state.model_service = (
        model_service
    )

    yield

    model_service.unload()

Lifespan principle: Load shared resources once during application startup and release them during shutdown rather than loading the model for every request.

Create the FastAPI Application

from fastapi import FastAPI


app = FastAPI(
    title=settings.application_name,
    version=settings.application_version,
    description=(
        "A typed Machine Learning API for "
        "customer-churn inference."
    ),
    lifespan=lifespan
)

Health Endpoint

from fastapi import APIRouter


health_router = APIRouter(
    tags=[
        "Health"
    ]
)


@health_router.get(
    "/health"
)
def health_check():

    return {
        "status": "healthy",
        "service": (
            "customer-churn-api"
        )
    }

The health endpoint confirms that the application process is responding. It does not confirm that the model has loaded successfully.

Readiness Endpoint

from fastapi import HTTPException
from fastapi import Request
from starlette import status


@health_router.get(
    "/ready"
)
def readiness_check(
    request: Request
):
    service = request.app.state.model_service

    if not service.ready:

        raise HTTPException(
            status_code=(
                status.HTTP_503_SERVICE_UNAVAILABLE
            ),
            detail={
                "code": "model_not_ready",
                "message": (
                    "The model is not ready "
                    "to process predictions."
                )
            }
        )

    return {
        "status": "ready",
        "model_loaded": True,
        "model_name": service.metadata[
            "model_name"
        ],
        "model_version": service.metadata[
            "model_version"
        ]
    }

Model Information Endpoint

@health_router.get(
    "/model-info"
)
def model_information(
    request: Request
):
    service = request.app.state.model_service

    if not service.ready:

        raise HTTPException(
            status_code=503,
            detail={
                "code": "model_not_ready",
                "message": (
                    "Model metadata is unavailable."
                )
            }
        )

    metadata = service.metadata

    return {
        "model_name": metadata[
            "model_name"
        ],
        "model_version": metadata[
            "model_version"
        ],
        "positive_class": metadata[
            "positive_class"
        ],
        "negative_class": metadata[
            "negative_class"
        ],
        "decision_threshold": metadata[
            "decision_threshold"
        ],
        "intended_use": metadata[
            "intended_use"
        ],
        "prohibited_use": metadata[
            "prohibited_use"
        ]
    }

Exposure rule: Return only approved non-sensitive metadata. Do not expose internal paths, credentials, restricted metrics, or confidential configuration.

Create a Model-Service Dependency

from typing import Annotated

from fastapi import Depends
from fastapi import Request


def get_model_service(
    request: Request
) -> ModelService:
    service = request.app.state.model_service

    if not service.ready:

        raise ModelNotReadyError(
            "The model is not ready."
        )

    return service


ModelServiceDependency = Annotated[
    ModelService,
    Depends(
        get_model_service
    )
]

Dependency benefit: Dependency injection centralizes resource access and makes path operations easier to test and maintain.

Individual Prediction Endpoint

import uuid

from fastapi import APIRouter


prediction_router = APIRouter(
    prefix="/predictions",
    tags=[
        "Predictions"
    ]
)


@prediction_router.post(
    "",
    response_model=(
        ChurnPredictionResponse
    )
)
def create_prediction(
    prediction_request: (
        ChurnPredictionRequest
    ),
    service: ModelServiceDependency
):
    request_id = str(
        uuid.uuid4()
    )

    prediction = service.predict_one(
        prediction_request.model_dump()
    )

    return ChurnPredictionResponse(
        **prediction,
        request_id=request_id
    )

Typed endpoint: The request model validates the input before the path operation executes, and the response model validates the returned structure.

Batch Prediction Endpoint

@prediction_router.post(
    "/batch",
    response_model=(
        BatchPredictionResponse
    )
)
def create_batch_prediction(
    batch_request: BatchPredictionRequest,
    service: ModelServiceDependency
):
    if (
        len(
            batch_request.records
        )
        >
        settings.maximum_batch_size
    ):
        raise HTTPException(
            status_code=400,
            detail={
                "code": "batch_too_large",
                "message": (
                    "The request exceeds the "
                    "maximum batch size."
                )
            }
        )

    request_id = str(
        uuid.uuid4()
    )

    records = [
        record.model_dump()
        for record in batch_request.records
    ]

    predictions = service.predict_many(
        records
    )

    return BatchPredictionResponse(
        predictions=predictions,
        prediction_count=len(
            predictions
        ),
        decision_threshold=float(
            service.metadata[
                "decision_threshold"
            ]
        ),
        model_name=service.metadata[
            "model_name"
        ],
        model_version=service.metadata[
            "model_version"
        ],
        request_id=request_id
    )

Batch limit: Use explicit record limits to protect service reliability. Very large workloads may require an asynchronous or scheduled batch inference architecture.

Register the Routers

app.include_router(
    health_router
)


app.include_router(
    prediction_router
)

Add Exception Handlers

import uuid

from fastapi import Request
from fastapi.responses import JSONResponse


@app.exception_handler(
    ModelNotReadyError
)
async def handle_model_not_ready(
    request: Request,
    error: ModelNotReadyError
):
    request_id = str(
        uuid.uuid4()
    )

    return JSONResponse(
        status_code=503,
        content={
            "error": {
                "code": "model_not_ready",
                "message": str(
                    error
                )
            },
            "request_id": request_id
        }
    )


@app.exception_handler(
    PredictionError
)
async def handle_prediction_error(
    request: Request,
    error: PredictionError
):
    request_id = str(
        uuid.uuid4()
    )

    request.app.logger.error(
        "Prediction failure",
        extra={
            "request_id": request_id
        }
    )

    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "code": "prediction_failed",
                "message": (
                    "The service could not "
                    "complete the prediction."
                )
            },
            "request_id": request_id
        }
    )

Unsafe Response Return raw stack traces, server paths, environment variables, or model-loading details to the client.

Safer Response Return a stable error code and client-safe message while preserving restricted diagnostic information in protected logs.

Run the Application

fastapi dev app/main.py

A common alternative during local development is to start the ASGI application with Uvicorn.

uvicorn app.main:app --reload --host 127.0.0.1 --port 8000

Development rule: Automatic reload is intended for development. Do not use reload or development debugging behavior in a production deployment.

Interactive API Documentation

Once the application is running, FastAPI generates interactive API documentation from the declared path operations and Pydantic schemas.

Path	Purpose
`/docs`	Interactive Swagger UI documentation
`/redoc`	Alternative ReDoc API documentation
`/openapi.json`	Generated OpenAPI schema

Documentation caution: API documentation can reveal endpoint and schema details. Control access appropriately when the API is private or sensitive.

Test with curl

curl http://127.0.0.1:8000/health

curl http://127.0.0.1:8000/ready

curl \
  -X POST \
  -H "Content-Type: application/json" \
  -d '{
    "tenure_months": 18,
    "monthly_charges": 64.5,
    "total_charges": 1161.0,
    "contract_type": "Monthly",
    "payment_method": "Electronic",
    "tech_support": "No"
  }' \
  http://127.0.0.1:8000/predictions

Synthetic-input notice: These values are invented for API testing and do not represent a real customer.

Example Prediction Response

{
    "predicted_class": "churned",
    "positive_class_probability": 0.71,
    "decision_threshold": 0.6,
    "retention_review_recommended": true,
    "model_name": "customer-churn-classifier",
    "model_version": "1.0.0",
    "request_id": "generated-request-identifier"
}

Illustrative response: The probability is an example of the response structure. Actual values must be generated by the deployed approved model.

HTTP Status Codes

Status	Meaning
`200`	The request completed successfully
`400`	The request violates an application rule such as batch size
`401`	The client is not authenticated
`403`	The authenticated client is not authorized
`404`	The requested resource does not exist
`413`	The request body exceeds the permitted size
`422`	The submitted data does not satisfy the declared request schema
`429`	The client exceeded an enforced request limit
`500`	An unexpected internal failure occurred
`503`	The model or a required service is not ready

Synchronous vs Asynchronous Endpoints

Endpoint Style	Appropriate Use
`def`	Ordinary blocking functions such as many CPU-bound model calls
`async def`	Awaitable operations such as compatible asynchronous network or database calls

Incorrect Assumption Declaring a CPU-heavy prediction route with async def automatically makes model inference faster.

Correct Understanding Asynchronous syntax helps when work can yield control while waiting. CPU-heavy inference may require process workers, specialized model servers, queues, accelerators, or another scaling strategy.

Create Test Fixtures

import pytest
from fastapi.testclient import TestClient

from app.main import app


@pytest.fixture()
def client():

    with TestClient(
        app
    ) as test_client:

        yield test_client

Testing benefit: Using TestClient as a context manager executes the application's lifespan logic during the test.

Test Health and Readiness

def test_health_endpoint(
    client
):
    response = client.get(
        "/health"
    )

    assert response.status_code == 200

    payload = response.json()

    assert payload[
        "status"
    ] == "healthy"


def test_readiness_endpoint(
    client
):
    response = client.get(
        "/ready"
    )

    assert response.status_code == 200

    payload = response.json()

    assert payload[
        "status"
    ] == "ready"

    assert payload[
        "model_loaded"
    ] is True

Test a Valid Prediction

def test_valid_prediction(
    client
):
    response = client.post(
        "/predictions",
        json={
            "tenure_months": 18,
            "monthly_charges": 64.5,
            "total_charges": 1161.0,
            "contract_type": "Monthly",
            "payment_method": "Electronic",
            "tech_support": "No"
        }
    )

    assert response.status_code == 200

    payload = response.json()

    assert payload[
        "predicted_class"
    ] in {
        "churned",
        "retained"
    }

    assert (
        0.0
        <=
        payload[
            "positive_class_probability"
        ]
        <=
        1.0
    )

    assert isinstance(
        payload[
            "retention_review_recommended"
        ],
        bool
    )

    assert payload[
        "model_version"
    ]

    assert payload[
        "request_id"
    ]

Test Automatic Validation

def test_missing_fields_are_rejected(
    client
):
    response = client.post(
        "/predictions",
        json={
            "tenure_months": 18
        }
    )

    assert response.status_code == 422


def test_negative_tenure_is_rejected(
    client
):
    response = client.post(
        "/predictions",
        json={
            "tenure_months": -1,
            "monthly_charges": 64.5,
            "total_charges": 1161.0,
            "contract_type": "Monthly",
            "payment_method": "Electronic",
            "tech_support": "No"
        }
    )

    assert response.status_code == 422


def test_unknown_category_is_rejected(
    client
):
    response = client.post(
        "/predictions",
        json={
            "tenure_months": 18,
            "monthly_charges": 64.5,
            "total_charges": 1161.0,
            "contract_type": "Unknown",
            "payment_method": "Electronic",
            "tech_support": "No"
        }
    )

    assert response.status_code == 422


def test_unexpected_field_is_rejected(
    client
):
    response = client.post(
        "/predictions",
        json={
            "tenure_months": 18,
            "monthly_charges": 64.5,
            "total_charges": 1161.0,
            "contract_type": "Monthly",
            "payment_method": "Electronic",
            "tech_support": "No",
            "unexpected_field": "value"
        }
    )

    assert response.status_code == 422

Test Batch Prediction

def test_batch_prediction(
    client
):
    record = {
        "tenure_months": 18,
        "monthly_charges": 64.5,
        "total_charges": 1161.0,
        "contract_type": "Monthly",
        "payment_method": "Electronic",
        "tech_support": "No"
    }

    response = client.post(
        "/predictions/batch",
        json={
            "records": [
                record,
                record
            ]
        }
    )

    assert response.status_code == 200

    payload = response.json()

    assert payload[
        "prediction_count"
    ] == 2

    assert len(
        payload[
            "predictions"
        ]
    ) == 2

python -m pytest -v

Authentication and Authorization

A production prediction endpoint should not assume that possession of its URL grants permission to use it. Authentication verifies the caller, while authorization verifies that the caller may perform the requested operation.

from typing import Annotated

from fastapi import Header
from fastapi import HTTPException


def require_api_key(
    x_api_key: Annotated[
        str | None,
        Header()
    ] = None
):
    expected_key = settings_api_key

    if (
        x_api_key is None
        or x_api_key != expected_key
    ):
        raise HTTPException(
            status_code=401,
            detail={
                "code": "unauthorized",
                "message": (
                    "Valid authentication "
                    "is required."
                )
            }
        )

Educational example: A static API key illustrates a dependency but is not a complete enterprise identity solution. Use the approved authentication, authorization, rotation, secret-management, and audit controls of the deployment environment.

Apply the Security Dependency

from fastapi import Depends


@prediction_router.post(
    "",
    response_model=(
        ChurnPredictionResponse
    ),
    dependencies=[
        Depends(
            require_api_key
        )
    ]
)
def create_prediction(
    prediction_request: (
        ChurnPredictionRequest
    ),
    service: ModelServiceDependency
):
    request_id = str(
        uuid.uuid4()
    )

    prediction = service.predict_one(
        prediction_request.model_dump()
    )

    return ChurnPredictionResponse(
        **prediction,
        request_id=request_id
    )

Production Security Requirements

Authenticated application or user identity
Explicit authorization for prediction access
Encrypted transport
Approved secret storage
Credential rotation
Request-size limits
Rate limits and abuse protection
Restricted network access
Dependency and container scanning
Auditable access records

Request Logging Middleware

import logging
import time
import uuid

from fastapi import Request


logger = logging.getLogger(
    "ml_api"
)


@app.middleware(
    "http"
)
async def request_logging_middleware(
    request: Request,
    call_next
):
    request_id = request.headers.get(
        "X-Request-ID",
        str(
            uuid.uuid4()
        )
    )

    started_at = time.perf_counter()

    response = await call_next(
        request
    )

    duration_ms = (
        time.perf_counter()
        - started_at
    ) * 1000

    response.headers[
        "X-Request-ID"
    ] = request_id

    logger.info(
        "request_completed",
        extra={
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "status_code": (
                response.status_code
            ),
            "duration_ms": round(
                duration_ms,
                3
            )
        }
    )

    return response

Privacy rule: Log approved operational metadata without recording complete customer payloads, access tokens, credentials, or unnecessary personal information.

Monitoring Requirements

Monitoring Area	Example Measure
Availability	Health and readiness success
Latency	Request, validation, preprocessing, and inference duration
Error Rate	Validation, authentication, prediction, and server errors
Traffic	Requests and records processed
Batch Size	Distribution of records per batch request
Input Quality	Missing, invalid, unsupported, or out-of-range values
Category Drift	New or changing categorical input values
Prediction Drift	Changes in probability and predicted-class distributions
Model Version	Requests and outcomes associated with each version
Delayed Quality	Performance after verified outcomes become available
Business Outcome	Effect on the approved retention-review workflow

Production Server

FastAPI applications run through an ASGI server. Uvicorn is commonly used for local execution and deployment, either directly or through a platform that manages the serving process.

uvicorn app.main:app --host 0.0.0.0 --port 8000

Environment	Execution Approach
Development	Local server with reload and approved test data
Automated Testing	TestClient with application lifespan
Production	Configured ASGI server or approved managed platform

Production principle: Configure startup, workers, timeouts, health checks, scaling, and termination behavior according to the model workload and hosting environment.

Worker Considerations

Separate worker processes can increase concurrency, but each process may load its own copy of the model. The appropriate worker count depends on model size, memory, CPU usage, request latency, concurrency, and infrastructure limits.

Concern	Review Question
Memory	How much memory does each loaded model instance require?
CPU	Is prediction compute-bound?
GPU	Can several workers safely share the accelerator?
Latency	How does worker count affect response time?
Throughput	How many requests must be processed?
Startup	How long does each worker require to load the artifact?

Capacity rule: Choose worker and scaling configuration from measured production-like load tests rather than a fixed formula.

Reverse Proxy and Gateway

Typical Serving Path

Client → API Gateway or Reverse Proxy → ASGI Server → FastAPI → ML Pipeline

Possible Gateway Responsibilities

TLS termination
Authentication integration
Authorization policy enforcement
Request routing
Rate limiting
Request-size enforcement
Access logging
Load distribution

Proxy caution: Trust forwarded headers only when the application is behind a correctly configured trusted proxy.

Container Readiness

Prepare the API for Docker

Declare all dependencies
Use an environment-configurable model path
Load the model before readiness succeeds
Write logs to standard output or an approved destination
Keep secrets outside the image
Use immutable model and application versions
Provide health and readiness endpoints
Use a production server command
Avoid local machine-specific paths
Do not store persistent application data inside an ephemeral container

Responsible ML API Design

Responsible-Use Requirements

Return the model version with each prediction
Preserve the validated decision threshold
Communicate that probability is not certainty
Document intended and prohibited uses
Do not return unsupported causal explanations
Protect personal and confidential input
Evaluate relevant operating groups
Provide human review for consequential decisions
Support correction of inaccurate source data
Maintain fallback and rollback procedures
Monitor model and operational outcomes

Unsafe API Behavior Treat the prediction as a final decision and trigger a harmful or consequential action without policy checks.

Responsible API Behavior Return the approved model output to an authorized workflow that applies business rules, eligibility checks, and required human oversight.

Common FastAPI ML Mistakes

Mistake Load the model inside every prediction request.

Correction Load and verify the model during application startup using lifespan logic.

Mistake Accept an unrestricted dictionary as prediction input.

Correction Define typed Pydantic request models with ranges, categories, and unexpected-field handling.

Mistake Rebuild preprocessing inside the endpoint.

Correction Load and use the complete fitted preprocessing and model pipeline.

Mistake Declare every endpoint asynchronous without reviewing the workload.

Correction Use synchronous or asynchronous functions according to the operations performed by the endpoint.

Mistake Return internal exceptions directly.

Correction Return stable client-safe errors and preserve restricted diagnostics in protected logs.

Mistake Allow unlimited batch size.

Correction Enforce batch and request-size limits or use an asynchronous batch architecture.

Mistake Expose interactive documentation and prediction endpoints without access controls.

Correction Configure authentication, authorization, network access, rate limits, and documentation exposure according to the environment.

Mistake Select worker counts without measuring memory and latency.

Correction Use production-like load tests to determine appropriate resource and scaling configuration.

Recommended FastAPI Workflow

Freeze the Approved Pipeline

Save the validated preprocessing, model, threshold, schema, metadata, and version.

Define Typed Contracts

Create Pydantic models for individual requests, batch requests, predictions, metadata, and errors.

Create the Inference Service

Separate artifact loading, feature ordering, inference, and threshold logic from path operations.

Use Lifespan Initialization

Load shared model resources before accepting prediction traffic.

Create Operational Endpoints

Add health, readiness, model-information, individual prediction, and batch prediction endpoints.

Add Controlled Errors

Map validation, readiness, prediction, authorization, and internal failures to stable response structures.

Test the API

Test lifespan, schemas, valid predictions, invalid values, batch limits, metadata, and failure behavior.

Secure the Endpoints

Add authentication, authorization, encrypted transport, secret management, network controls, and rate limits.

Deploy Through ASGI

Configure the ASGI server, gateway, workers, resources, timeouts, and termination behavior.

Monitor and Maintain

Track availability, latency, errors, inputs, predictions, versions, delayed outcomes, and business impact.

Hands-On Practice

Create the Project Structure

Separate application configuration, schemas, inference, routers, dependencies, exceptions, artifacts, and tests.

Define Pydantic Schemas

Create typed individual, batch, response, metadata, and error models.

Load the Model with Lifespan

Initialize the trusted model artifact during startup and clear shared resources during shutdown.

Create Health and Readiness Endpoints

Distinguish process availability from model readiness.

Create Prediction Endpoints

Implement individual and limited batch prediction using the reusable model service.

Test Automatic Documentation

Inspect request and response schemas through the generated documentation interface.

Create Automated Tests

Test valid input, missing fields, ranges, categories, additional fields, batch limits, readiness, and prediction output.

Run a Production-Like Test

Start the application without reload, send representative requests, and record latency, memory, errors, and model version.

Knowledge Check

What role does FastAPI play in Machine Learning deployment?

Explain how it connects validated HTTP requests to a reusable inference service.

What is a Pydantic model?

Explain how it defines, validates, documents, and serializes structured API data.

Why use application lifespan?

Explain how shared resources can be initialized before requests and released during shutdown.

Why should unexpected request fields be controlled?

Explain contract clarity, client errors, inconsistent input, and safe inference.

What is the purpose of a response model?

Explain output validation, serialization, documentation, and contract consistency.

Why are health and readiness different?

Explain process operation versus the ability to accept safe prediction traffic.

Why does async syntax not automatically speed up inference?

Explain awaitable waiting compared with CPU-bound model computation.

Why limit batch size?

Explain memory use, latency, fairness, service reliability, and abuse control.

Why return the model version?

Explain traceability, monitoring, investigation, deployment comparison, and rollback.

What should be monitored after deployment?

Explain availability, latency, errors, traffic, input quality, prediction drift, model versions, delayed performance, and business outcomes.

FastAPI Readiness Checklist

Confirm Before Deployment

The approved complete pipeline is available
The model artifact comes from a trusted source
The artifact and metadata versions are recorded
The model loads during lifespan startup
Readiness remains false when model loading fails
Typed request schemas are defined
Typed response schemas are defined
Unexpected input fields are controlled
Numerical ranges are validated
Categorical values are validated
Non-finite numbers are rejected
Individual prediction is tested
Batch size is limited
Health and readiness endpoints are available
Model metadata exposure is controlled
Client-safe error responses are implemented
Automatic documentation exposure is reviewed
Automated tests pass
Authentication and authorization are configured
Secrets are stored outside source code
Request IDs are supported
Logs exclude sensitive payloads
Production server settings are documented
Worker and resource settings are load-tested
Monitoring and alert ownership are defined
Fallback and rollback procedures are available

Lesson Summary

FastAPI Area	Main Outcome
Pydantic Request Model	Defines and validates prediction input
Response Model	Defines and validates prediction output
Lifespan	Loads and releases shared model resources
Inference Service	Separates model behavior from HTTP route logic
Dependency Injection	Provides reusable application resources to endpoints
Prediction Endpoint	Returns typed and versioned model output
Batch Endpoint	Processes a controlled collection of records
Health Check	Reports application-process availability
Readiness Check	Reports whether the model can accept prediction traffic
OpenAPI Documentation	Documents paths and contracts from declared schemas
Exception Handling	Provides stable client-safe errors
Automated Testing	Verifies schemas, lifespan, endpoints, and model behavior
Security	Protects identity, permissions, secrets, traffic, and data
Monitoring	Tracks service, data, prediction, model, and business behavior

Key Takeaway

FastAPI for Machine Learning APIs provides a typed interface between client applications and saved model pipelines. Pydantic schemas validate requests and responses, lifespan logic loads the approved model once during startup, dependency injection provides reusable application resources, and the generated OpenAPI schema documents the API contract. A reliable deployment must also enforce authentication, authorization, batch and request limits, safe error handling, protected logging, automated testing, model version traceability, production ASGI configuration, health and readiness checks, monitoring, fallback behavior, and rollback capability. FastAPI simplifies API construction, but responsible deployment still depends on complete engineering, security, and governance controls.

Get Machine Learning updates by email

Short, practical lessons and course updates from Rumman Ansari. No spam.