RugvAI Labs

Building APIs with FastAPI: From Endpoint to Deployment

FastAPI is the default choice for typed Python APIs: automatic OpenAPI docs, Pydantic-typed everything, async-native. In this lesson you'll ship the skeleton a real service is built on — a typed endpoint, validation, and an interactive /docs page generated for free.

Learning Objectives

After this lesson, you will be able to:

Explain what an API is, how REST works, and why HTTP status codes matter
Build a FastAPI application with path parameters, query parameters, and request bodies
Use Pydantic models for request validation, serialization, and nested data structures
Serve a trained ML model behind a /predict endpoint with batch prediction support
Add error handling, CORS middleware, and basic authentication to a FastAPI app
Write integration tests for API endpoints using httpx and TestClient
Distinguish ASGI from WSGI and understand why FastAPI is async-first

Why

Why This Matters

ML models are useless if nobody can reach them -- training a model is only half the job. The other half is serving it behind an API so that frontends, mobile apps, data pipelines, and other services can send data and get predictions back. FastAPI is the dominant framework for Python ML serving because it is fast, type-safe, and generates automatic documentation
APIs are how modern software communicates -- every time you use Google Maps, check the weather, or ask ChatGPT a question, your app is calling an API. Understanding APIs is not optional for any software or ML engineer
FastAPI uses the same Pydantic models you already know -- if you have written Pydantic validation code, you already know 70% of FastAPI. The framework eliminates the boilerplate that made Flask painful for production use

Prerequisites

Before You Start

Review

#Why APIs? REST, HTTP, and JSON

An API (Application Programming Interface) is a contract between two pieces of software. One side sends a request, the other sends a response. The most common pattern on the web is REST (Representational State Transfer), which uses HTTP methods to operate on resources.

#HTTP Methods

Method	Purpose	Example	Analogy
`GET`	Read data	Get a user's profile	Looking up a book in a library
`POST`	Create data	Submit a new prediction request	Filling out a form
`PUT`	Replace data entirely	Update all user fields	Rewriting an entire chapter
`PATCH`	Update data partially	Change just the user's email

#HTTP Status Codes

Status codes tell the client what happened:

200 OK              -- Request succeeded, here is the data
201 Created         -- A new resource was created
204 No Content      -- Success, but nothing to return (e.g., after DELETE)
400 Bad Request     -- Client sent invalid data
401 Unauthorized    -- Client is not authenticated
403 Forbidden       -- Client is authenticated but not allowed
404 Not Found       -- The resource does not exist
422 Unprocessable   -- Data is syntactically correct but semantically invalid
500 Internal Error  -- Something broke on the server

#JSON -- The Universal Data Format

APIs exchange data as JSON (JavaScript Object Notation). It maps directly to Python dictionaries:

pythonrunnable cell

# JSON
{
    "name": "Alice",
    "age": 28,
    "scores": [95, 87, 92],
    "address": {
        "city": "San Francisco",
        "state": "CA"
    }
}

# Python equivalent
data = {
    "name": "Alice",
    "age": 28,
    "scores": [95, 87, 92],
    "address": {
        "city": "San Francisco",
        "state": "CA"
    }
}

#FastAPI Basics -- Your First API

#Installation

bash

pip install fastapi uvicorn

fastapi is the framework. uvicorn is the ASGI server that runs it.

#Hello World

pythonreference · read-only

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    """Root endpoint -- returns a welcome message."""
    return {"message": "Hello, World!"}

@app.get("/health")
def health_check():
    """Health check endpoint for monitoring."""
    return {"status": "healthy"}

Run it:

bash

uvicorn main:app --reload

main = the Python file (main.py)
app = the FastAPI instance
--reload = auto-restart on code changes (development only)

Visit http://localhost:8000 and you see {"message": "Hello, World!"}. Visit http://localhost:8000/docs and you get automatic interactive documentation powered by Swagger UI -- this is one of FastAPI's killer features.

#Path Parameters

Path parameters are variables embedded in the URL:

pythonreference · read-only

@app.get("/users/{user_id}")
def get_user(user_id: int):
    """Get a user by their ID."""
    return {"user_id": user_id, "name": f"User #{user_id}"}

@app.get("/models/{model_name}/version/{version}")
def get_model_version(model_name: str, version: int):
    """Get a specific version of an ML model."""
    return {
        "model": model_name,
        "version": version,
        "status": "active"
    }

FastAPI automatically validates types. If you request /users/abc, you get a 422 error because abc is not an integer. No manual validation needed.

#Query Parameters

Query parameters come after the ? in a URL:

pythonreference · read-only

from typing import Optional

@app.get("/predictions")
def list_predictions(
    model_name: str,
    limit: int = 10,
    offset: int = 0,
    status: Optional[str] = None,
):
    """List predictions with pagination and optional filtering."""
    return {
        "model": model_name,
        "limit": limit,
        "offset": offset,
        "status_filter": status,
    }

Request: GET /predictions?model_name=classifier&limit=5&status=completed

FastAPI distinguishes path params from query params automatically: if the parameter is in the path, it is a path param. Otherwise, it is a query param.

#Request Body with Pydantic

For POST and PUT requests, you send data in the request body. FastAPI uses Pydantic models to define and validate the shape of that data:

pythonreference · read-only

from pydantic import BaseModel

class PredictionRequest(BaseModel):
    """Schema for a prediction request."""
    features: list[float]
    model_name: str = "default"
    return_probabilities: bool = False

@app.post("/predict")
def predict(request: PredictionRequest):
    """Make a prediction based on input features."""
    # request.features is already validated as list[float]
    # request.model_name defaults to "default" if not provided
    prediction = sum(request.features) / len(request.features)  # placeholder
    return {
        "prediction": prediction,
        "model_used": request.model_name,
        "input_dim": len(request.features),
    }

Send a POST request with JSON body:

json

{
    "features": [1.5, 2.3, 0.7, 4.1],
    "model_name": "linear_v2",
    "return_probabilities": true
}

FastAPI automatically parses the JSON, validates it against the Pydantic model, and returns a 422 error with detailed messages if validation fails.

What Do You Think?

You send a POST request to /predict with body {"features": "not a list", "model_name": "test"}. What happens?

The answer is FastAPI returns a 422 Unprocessable Entity. Pydantic validates every field against its type annotation before your function runs. If features is not a list[float], you get a clear error message like "value is not a valid list". Your endpoint code never sees invalid data.

#Pydantic Models -- Validation, Serialization, and Nested Models

Pydantic is not just for type checking. It handles validation, serialization, default values, and complex nested structures.

#Field Validators

pythonreference · read-only

from pydantic import BaseModel, Field, field_validator

class TrainingConfig(BaseModel):
    """Configuration for model training."""
    model_name: str = Field(..., min_length=1, max_length=100)
    learning_rate: float = Field(default=0.001, gt=0, lt=1)
    epochs: int = Field(default=10, ge=1, le=1000)
    batch_size: int = Field(default=32, description="Must be a power of 2")
    dropout: float = Field(default=0.1, ge=0, le=1)

    @field_validator("batch_size")
    @classmethod
    def batch_size_must_be_power_of_two(cls, v):
        if v & (v - 1) != 0:
            raise ValueError(f"batch_size must be a power of 2, got {v}")
        return v

    @field_validator("model_name")
    @classmethod
    def model_name_must_be_lowercase(cls, v):
        if v != v.lower():
            raise ValueError("model_name must be lowercase")
        return v

Field(...) means "this field is required" (no default). Field(default=0.001, gt=0) means "defaults to 0.001 and must be greater than 0."

#Nested Models

pythonreference · read-only

class DatasetInfo(BaseModel):
    name: str
    num_samples: int = Field(ge=1)
    num_features: int = Field(ge=1)
    target_column: str

class ModelMetrics(BaseModel):
    accuracy: float = Field(ge=0, le=1)
    precision: float = Field(ge=0, le=1)
    recall: float = Field(ge=0, le=1)
    f1_score: float = Field(ge=0, le=1)

class TrainingResult(BaseModel):
    """Complete training result with nested models."""
    config: TrainingConfig
    dataset: DatasetInfo
    metrics: ModelMetrics
    training_time_seconds: float
    model_path: str

# FastAPI automatically handles nested validation
@app.post("/train", response_model=TrainingResult)
def train_model(config: TrainingConfig, dataset: DatasetInfo):
    """Train a model and return the results."""
    # ... training logic ...
    return TrainingResult(
        config=config,
        dataset=dataset,
        metrics=ModelMetrics(
            accuracy=0.92, precision=0.89,
            recall=0.91, f1_score=0.90,
        ),
        training_time_seconds=142.5,
        model_path="/models/classifier_v3.pkl",
    )

#Response Models

The response_model parameter in the decorator controls what gets sent back to the client. It filters out any extra fields and validates the response:

pythonreference · read-only

class UserPublic(BaseModel):
    """Public user info -- excludes sensitive fields."""
    id: int
    name: str
    email: str

class UserInternal(BaseModel):
    """Internal user info -- includes everything."""
    id: int
    name: str
    email: str
    hashed_password: str
    api_key: str

@app.get("/users/{user_id}", response_model=UserPublic)
def get_user(user_id: int):
    # Even though we return all fields internally,
    # FastAPI strips hashed_password and api_key from the response
    user = get_user_from_db(user_id)  # returns UserInternal
    return user

#ML Model Serving -- The /predict Endpoint

This is the most common use case for FastAPI in the ML world: serving a trained model behind an API.

#Loading a Trained Model

pythonreference · read-only

import pickle
from pathlib import Path
from fastapi import FastAPI
from pydantic import BaseModel, Field
import numpy as np

app = FastAPI(title="ML Prediction Service", version="1.0.0")

# Load model at startup (once, not per request)
MODEL_PATH = Path("models/classifier.pkl")

def load_model():
    """Load the trained model from disk."""
    with open(MODEL_PATH, "rb") as f:
        return pickle.load(f)

# Use a lifespan event to load the model once at startup
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Load ML model on startup, clean up on shutdown."""
    app.state.model = load_model()
    print(f"Model loaded from {MODEL_PATH}")
    yield
    print("Shutting down, cleaning up...")

app = FastAPI(title="ML Prediction Service", lifespan=lifespan)

#Single Prediction Endpoint

pythonreference · read-only

class PredictRequest(BaseModel):
    features: list[float] = Field(..., min_length=1)
    
class PredictResponse(BaseModel):
    prediction: int
    confidence: float = Field(ge=0, le=1)
    model_version: str

@app.post("/predict", response_model=PredictResponse)
def predict_single(request: PredictRequest):
    """Make a single prediction."""
    model = app.state.model
    features = np.array(request.features).reshape(1, -1)
    
    prediction = model.predict(features)[0]
    probabilities = model.predict_proba(features)[0]
    confidence = float(max(probabilities))
    
    return PredictResponse(
        prediction=int(prediction),
        confidence=confidence,
        model_version="1.0.0",
    )

#Batch Prediction Endpoint

pythonreference · read-only

class BatchPredictRequest(BaseModel):
    samples: list[list[float]] = Field(..., min_length=1, max_length=1000)

class BatchPredictResponse(BaseModel):
    predictions: list[int]
    confidences: list[float]
    count: int
    model_version: str

@app.post("/predict/batch", response_model=BatchPredictResponse)
def predict_batch(request: BatchPredictRequest):
    """Make predictions on a batch of samples (up to 1000)."""
    model = app.state.model
    features = np.array(request.samples)
    
    predictions = model.predict(features)
    probabilities = model.predict_proba(features)
    confidences = [float(max(row)) for row in probabilities]
    
    return BatchPredictResponse(
        predictions=[int(p) for p in predictions],
        confidences=confidences,
        count=len(predictions),
        model_version="1.0.0",
    )

#Model Metadata Endpoint

pythonreference · read-only

@app.get("/model/info")
def model_info():
    """Return metadata about the currently loaded model."""
    model = app.state.model
    return {
        "model_type": type(model).__name__,
        "n_features": model.n_features_in_,
        "classes": model.classes_.tolist(),
        "model_version": "1.0.0",
    }

What Do You Think?

You load a large ML model inside the /predict endpoint function (not at startup). What happens when 100 users hit the endpoint simultaneously?

The answer is the model loads 100 times. Each request executes the entire function body, including the model loading. This is why you load models at startup using lifespan events or global variables -- the model lives in memory once and is shared across all requests. Loading a 500MB model per request would be catastrophic.

#Error Handling, Middleware, CORS, and Authentication

#Custom Error Handling

pythonreference · read-only

from fastapi import FastAPI, HTTPException, status

@app.post("/predict")
def predict(request: PredictRequest):
    """Predict with proper error handling."""
    model = app.state.model
    
    # Check feature dimensions
    expected_features = model.n_features_in_
    if len(request.features) != expected_features:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Expected {expected_features} features, got {len(request.features)}",
        )
    
    try:
        features = np.array(request.features).reshape(1, -1)
        prediction = model.predict(features)[0]
        return {"prediction": int(prediction)}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Prediction failed: {str(e)}",
        )

#CORS Middleware

When a frontend on localhost:3000 tries to call your API on localhost:8000, the browser blocks it by default (Cross-Origin Resource Sharing policy). You need CORS middleware:

pythonreference · read-only

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

#Request Logging Middleware

pythonreference · read-only

import time
import logging

logger = logging.getLogger("api")

@app.middleware("http")
async def log_requests(request, call_next):
    """Log every request with timing information."""
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    logger.info(
        f"{request.method} {request.url.path} "
        f"-> {response.status_code} ({duration:.3f}s)"
    )
    return response

#Basic API Key Authentication

pythonreference · read-only

from fastapi import Depends, Security
from fastapi.security import APIKeyHeader
import os

API_KEY = os.environ.get("API_KEY", "dev-key-change-me")
api_key_header = APIKeyHeader(name="X-API-Key")

def verify_api_key(api_key: str = Security(api_key_header)):
    """Verify the API key from the request header."""
    if api_key != API_KEY:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key",
        )
    return api_key

@app.post("/predict", dependencies=[Depends(verify_api_key)])
def predict_secure(request: PredictRequest):
    """Secured prediction endpoint -- requires valid API key."""
    # ... prediction logic ...
    return {"prediction": 1}

Watch Out

Watch out

Fix: Either use a sync def endpoint (FastAPI runs it in a thread pool automatically) or use asyncio-compatible libraries (httpx, aiofiles) in async def endpoints. Never put blocking code inside an async def — it blocks the entire event loop, freezing ALL concurrent requests.

Running blocking operations (like loading a model, reading a large file, or calling a slow external API) inside an async FastAPI endpoint

FastAPI runs on a single-threaded async event loop (uvicorn). When you write 'async def predict()', FastAPI expects the function to yield control with 'await' so other requests can be served. If you call time.sleep(5) or pickle.load() inside an async def, the entire server stops responding for 5 seconds. A regular 'def' function is automatically run in a thread pool, so blocking is safe. Rule of thumb: use 'def' unless you are doing real async I/O (HTTP calls with httpx, database queries with async drivers).

Dependency Injection with `Depends`

FastAPI's Depends system lets you declare reusable logic (authentication, database sessions, rate limiting) that runs before your endpoint and injects its result as a parameter. This is called dependency injection.

pythonreference · read-only

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session

# --- Database session dependency ---
def get_db():
    """Provide a database session for the duration of one request."""
    db = SessionLocal()     # create a new session
    try:
        yield db            # inject the session into the endpoint
        db.commit()         # commit if no exception
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()          # always close, even if an exception occurred

@app.get("/experiments/{exp_id}")
def get_experiment(
    exp_id: int,
    db: Session = Depends(get_db),   # FastAPI calls get_db() and injects the result
):
    """Fetch an experiment by ID. db is automatically provided."""
    exp = db.query(Experiment).filter(Experiment.id == exp_id).first()
    if not exp:
        raise HTTPException(status_code=404, detail=f"Experiment {exp_id} not found")
    return exp

# --- Composable dependencies: auth + db ---
def get_current_user(api_key: str = Security(api_key_header)) -> str:
    if api_key != VALID_API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key  # or return the user object from DB

@app.post("/predict", dependencies=[Depends(get_current_user)])
def predict_secured(
    request: PredictRequest,
    db: Session = Depends(get_db),
):
    """Endpoint protected by both authentication AND a database session."""
    # FastAPI resolved get_current_user (auth check) and get_db (DB session)
    # before this function body runs
    ...

Depends works for:

Database sessions. One session per request, committed/rolled back automatically
Authentication. Verify JWT tokens or API keys, raise 401 if invalid
Rate limiting. Check request counts against Redis before the endpoint runs
Settings injection. Inject typed settings without global variables
Pagination parameters. Standardize limit/offset parsing across all list endpoints

Watch Out

Watch out

Fix: FastAPI caches dependency results within a single request by default. If get_db() is depended on by both the endpoint and a sub-dependency, FastAPI calls it ONCE and reuses the session. To get a fresh call each time, set use_cache=False: Depends(get_db, use_cache=False).

Declaring the same dependency multiple times in the same request chain, assuming FastAPI will call it multiple times.

This caching behaviour is usually what you want (one DB session per request, not one per dependency), but it can surprise you when you expect side-effects to run multiple times. The caching key is the dependency function itself — two different functions that do the same thing are NOT cached together.

```python

get_db is called ONCE, even though two things depend on it

@app.post("/items") def create_item( item: ItemCreate, db: Session = Depends(get_db), # FastAPI calls get_db() user = Depends(get_current_user), # get_current_user also calls Depends(get_db)

FastAPI reuses the SAME db session for both — correct behaviour

): ...

To force a fresh call (rare):

from fastapi import Depends Depends(get_db, use_cache=False) ```

Most of the time, the caching is the right behaviour. Be aware of it when debugging unexpected lifecycle issues.

#Testing with httpx and TestClient

FastAPI provides a TestClient that lets you test endpoints without starting a real server:

pythonreference · read-only

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    """Test the root endpoint returns a welcome message."""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, World!"}

def test_health_check():
    """Test the health endpoint."""
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

def test_predict_valid_input():
    """Test prediction with valid features."""
    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0, 4.0]},
    )
    assert response.status_code == 200
    data = response.json()
    assert "prediction" in data
    assert "confidence" in data

def test_predict_invalid_features():
    """Test prediction with invalid feature type."""
    response = client.post(
        "/predict",
        json={"features": "not a list"},
    )
    assert response.status_code == 422

def test_predict_empty_features():
    """Test prediction with empty feature list."""
    response = client.post(
        "/predict",
        json={"features": []},
    )
    assert response.status_code == 422

def test_predict_wrong_dimensions():
    """Test prediction with wrong number of features."""
    response = client.post(
        "/predict",
        json={"features": [1.0]},  # model expects 4 features
    )
    assert response.status_code == 400
    assert "Expected" in response.json()["detail"]

#Testing with pytest Fixtures

pythonreference · read-only

import pytest
from fastapi.testclient import TestClient

@pytest.fixture
def client():
    """Create a test client with a fresh app instance."""
    from main import app
    return TestClient(app)

@pytest.fixture
def auth_headers():
    """Provide authentication headers for secured endpoints."""
    return {"X-API-Key": "dev-key-change-me"}

def test_secured_endpoint_without_key(client):
    """Requests without API key should be rejected."""
    response = client.post("/predict", json={"features": [1.0, 2.0, 3.0, 4.0]})
    assert response.status_code == 401  # or 403

def test_secured_endpoint_with_key(client, auth_headers):
    """Requests with valid API key should succeed."""
    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0, 4.0]},
        headers=auth_headers,
    )
    assert response.status_code == 200

@pytest.mark.parametrize("features,expected_status", [
    ([1.0, 2.0, 3.0, 4.0], 200),
    ([0.0, 0.0, 0.0, 0.0], 200),
    ([], 422),
    ("invalid", 422),
])
def test_predict_various_inputs(client, auth_headers, features, expected_status):
    """Parametrized test for various prediction inputs."""
    response = client.post(
        "/predict",
        json={"features": features},
        headers=auth_headers,
    )
    assert response.status_code == expected_status

#Async Testing with httpx

For async endpoints, use httpx.AsyncClient:

pythonreference · read-only

import httpx
import pytest

@pytest.mark.asyncio
async def test_predict_async():
    """Test with async client for full async stack testing."""
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/predict",
            json={"features": [1.0, 2.0, 3.0, 4.0]},
        )
        assert response.status_code == 200

#Code Playground

pythonplayground.py · Pyodide

# FastAPI Basics — Building a Prediction API
# This playground simulates FastAPI patterns without a running server

from pydantic import BaseModel, Field, field_validator
from typing import Optional
import math

# ── Pydantic Models ─────────────────────────────────────────────────────────
class PredictRequest(BaseModel):
  """Validated prediction request."""
  features: list[float] = Field(..., min_length=1, max_length=100)
  model_name: str = "default"
  return_probabilities: bool = False
  
  @field_validator("features")
  @classmethod
  def features_must_be_finite(cls, v):
      for i, val in enumerate(v):
          if math.isnan(val) or math.isinf(val):
              raise ValueError(f"Feature at index {i} is not finite: {val}")
      return v

class PredictResponse(BaseModel):
  prediction: int
  confidence: float = Field(ge=0, le=1)
  model_name: str
  input_dim: int

class BatchRequest(BaseModel):
  samples: list[list[float]] = Field(..., min_length=1, max_length=1000)

class BatchResponse(BaseModel):
  predictions: list[int]
  count: int

# ── Simulated Model ─────────────────────────────────────────────────────────
class SimpleClassifier:
  """A toy classifier that predicts based on feature mean."""
  def predict(self, features: list[float]) -> int:
      return 1 if sum(features) / len(features) > 0.5 else 0
  
  def predict_proba(self, features: list[float]) -> list[float]:
      mean = sum(features) / len(features)
      prob_1 = 1 / (1 + math.exp(-mean))  # sigmoid
      return [1 - prob_1, prob_1]

model = SimpleClassifier()

# ── Simulated Endpoint Functions ────────────────────────────────────────────
def predict_endpoint(request_data: dict) -> dict:
  """Simulate the /predict endpoint."""
  # Validate input with Pydantic
  request = PredictRequest(**request_data)
  
  prediction = model.predict(request.features)
  probabilities = model.predict_proba(request.features)
  confidence = max(probabilities)
  
  response = PredictResponse(
      prediction=prediction,
      confidence=round(confidence, 4),
      model_name=request.model_name,
      input_dim=len(request.features),
  )
  return response.model_dump()

def batch_endpoint(request_data: dict) -> dict:
  """Simulate the /predict/batch endpoint."""
  request = BatchRequest(**request_data)
  predictions = [model.predict(sample) for sample in request.samples]
  response = BatchResponse(predictions=predictions, count=len(predictions))
  return response.model_dump()

# ── Test the endpoints ──────────────────────────────────────────────────────
print("=== Single Prediction ===")
result = predict_endpoint({
  "features": [0.8, 1.2, 0.3, 0.9],
  "model_name": "sigmoid_v1",
})
print(f"Result: {result}")

print("\n=== Batch Prediction ===")
batch_result = batch_endpoint({
  "samples": [
      [0.1, 0.2, 0.1],
      [0.9, 1.5, 2.0],
      [0.5, 0.5, 0.5],
  ]
})
print(f"Batch Result: {batch_result}")

print("\n=== Validation Tests ===")
# Test: valid request
try:
  result = predict_endpoint({"features": [1.0, 2.0, 3.0]})
  print(f"Valid request: OK -> prediction={result['prediction']}")
except Exception as e:
  print(f"Valid request: FAILED -> {e}")

# Test: empty features (should fail validation)
try:
  result = predict_endpoint({"features": []})
  print("Empty features: Should have failed!")
except Exception as e:
  print(f"Empty features: Correctly rejected -> {type(e).__name__}")

# Test: NaN in features (should fail custom validator)
try:
  result = predict_endpoint({"features": [1.0, float("nan"), 3.0]})
  print("NaN features: Should have failed!")
except Exception as e:
  print(f"NaN features: Correctly rejected -> {type(e).__name__}")

# Test: wrong type (should fail validation)
try:
  result = predict_endpoint({"features": "not a list"})
  print("Wrong type: Should have failed!")
except Exception as e:
  print(f"Wrong type: Correctly rejected -> {type(e).__name__}")

# CHALLENGES:
# 1. Add a "temperature" field to PredictRequest (float, 0.0 to 2.0, default 1.0)
# 2. Create a ModelInfo response model and a model_info_endpoint() function
# 3. Add a field_validator to BatchRequest that checks all samples have the same length
# 4. Simulate an error handling wrapper that returns status codes like FastAPI would

Tests · Build Pydantic models, validate inputs, simulate API endpoints, and handle errors!

#Key Takeaways

FastAPI validates inputs automatically with Pydantic -- define a BaseModel for your request body, and FastAPI rejects invalid data with a 422 error before your code runs. No manual if isinstance() checks needed
Load ML models at startup, not per request -- use lifespan events or global variables to load the model once. Loading per request multiplies memory usage and latency by the number of concurrent users
Use def for CPU-bound endpoints, async def for I/O-bound -- putting blocking code (model inference, file I/O) inside async def freezes the entire server. FastAPI runs sync functions in a thread pool automatically
response_model controls what the client sees -- use it to strip sensitive fields (passwords, API keys) from responses and to generate accurate API documentation
TestClient makes API testing trivial -- no running server needed. Combine with pytest fixtures and parametrize for comprehensive endpoint testing
Visit /docs for free interactive documentation -- FastAPI generates Swagger UI automatically from your type annotations and Pydantic models

Quick Check1 / 5

What happens when a POST request sends data that does not match the Pydantic model?

Your Reflection

Continue Learning

Discussion

Building APIs with FastAPI: From Endpoint to Deployment

After this lesson, you will be able to:

Before You Start

#Why APIs? REST, HTTP, and JSON

#HTTP Methods

#HTTP Status Codes

#JSON -- The Universal Data Format

#FastAPI Basics -- Your First API

#Installation

#Hello World

#Path Parameters

#Query Parameters

#Request Body with Pydantic

#Pydantic Models -- Validation, Serialization, and Nested Models

#Field Validators

#Nested Models

#Response Models

#ML Model Serving -- The /predict Endpoint

#Loading a Trained Model

#Single Prediction Endpoint

#Batch Prediction Endpoint

#Model Metadata Endpoint

#Error Handling, Middleware, CORS, and Authentication

#Custom Error Handling

#CORS Middleware

#Request Logging Middleware

#Basic API Key Authentication

Dependency Injection with `Depends`

#Testing with httpx and TestClient

#Testing with pytest Fixtures

#Async Testing with httpx

#Code Playground

ASGI vs WSGI: Why FastAPI Is Async-First

#Key Takeaways

Your Reflection

Continue Learning

Discussion

Building APIs with FastAPI: From Endpoint to Deployment

#Why APIs? REST, HTTP, and JSON

#HTTP Methods

#HTTP Status Codes

#JSON -- The Universal Data Format

#FastAPI Basics -- Your First API

#Installation

#Hello World

#Path Parameters

#Query Parameters

#Request Body with Pydantic

#Pydantic Models -- Validation, Serialization, and Nested Models

#Field Validators

#Nested Models

#Response Models

#ML Model Serving -- The /predict Endpoint

#Loading a Trained Model

#Single Prediction Endpoint

#Batch Prediction Endpoint

#Model Metadata Endpoint

#Error Handling, Middleware, CORS, and Authentication

#Custom Error Handling

#CORS Middleware

#Request Logging Middleware

#Basic API Key Authentication

Dependency Injection with Depends

#Testing with httpx and TestClient

#Testing with pytest Fixtures

#Async Testing with httpx

#Code Playground

ASGI vs WSGI: Why FastAPI Is Async-First

#Key Takeaways

Dependency Injection with `Depends`