The Prediction Service

Context

A prediction service takes one listing as JSON over HTTP, checks it against a declared schema before anything else runs, prices it with a model loaded once when the service starts, and sends the price back as JSON. The check is a pydantic model, a Python class whose fields carry types and bounds, and a request that fails it is answered with status 422 and a list of exactly what was wrong.

Every system that wants a price talks to this endpoint: the listing site's form when a seller presses "estimate", a nightly job that revalues the portfolio, an analyst's notebook. None of them imports your pipeline. They send a request shaped by the schema the service publishes, and they trust whatever number comes back, so the schema is the contract and the endpoint is the only part of the model they ever see.

Lesson 6.2.3's app answers /health and nothing else. Hand the champion a listing straight from a request and the failures are the ones Lesson 6.2.2 met, plus quieter ones: a neighborhood the encoder never saw, an area sent as the word eighty, a field the model must not see, and a fractional room count that the frame's integer column would cut to a whole one without a word. A model loaded inside the route instead of at startup pays for the load on every request, and a broken alias then hides behind a healthy /health.

Goal

• Write schema.py, a pydantic Listing with extra="forbid", numeric bounds and Literal categories, and read the four error types it raises on a bad listing.
• Predict which wrongly typed values lax mode converts and which it refuses, and show that True passes as one room.
• Load the model once in a FastAPI lifespan, measure what that costs at startup and saves per request, and read the startup failure a missing alias causes.
• Build POST /predict, which validates a Listing and returns a Quote, and trace every 422 body back to the field and rule that produced it.
• Test the service against a model saved by a fixture instead of the registry, and diagnose the two failures a TestClient used without with produces.

Prerequisites

Both are optional, and the lesson stands without them. They are here for anyone who wants pydantic and FastAPI's startup hook in their own words first.

• YouTube - "Why Python Needs Pydantic for Real Applications" by ArjanCodes. Fourteen minutes on what pydantic is, validation of fields and whole models, custom serialization, and how FastAPI uses it.
• FastAPI documentation - "Lifespan Events". The use case is loading a machine learning model once before the app takes requests, with an @asynccontextmanager function whose code before yield runs at startup and after it at shutdown.

Table of Contents

• A Listing Schema With pydantic
• What Lax Mode Converts
• Loading the Model Once in a Lifespan
• POST /predict and Its 422s
• Testing the Service Without the Registry

Theory

Everything runs in ~/dev/listings-service at the state Lesson 6.2.3 left it, with (bootcamp) deactivated as Module 6's setup page says: the registry in mlflow.db holds listings-price versions 1 and 2, champion points at version 1, and app.py serves /health. The lesson declares pydantic in pyproject.toml and uv.lock and changes four project files, src/listings/schema.py (new), src/listings/app.py, tests/conftest.py and tests/test_app.py, and every probe script lives in Lesson 6.1.3's probe folder, ~/dev/listings-probes, run from the project folder as uv run python ../listings-probes/<name>.py, so the project holds only what it ships. The service runs in Terminal 1 and every curl goes in Terminal 2, as in Lesson 6.2.3.

A Listing Schema With pydantic

What. A pydantic model is a class that inherits from BaseModel and declares one typed attribute per field. Building one from a dictionary checks every field against its type and its Field(...) constraints, collects every failure, and raises one ValidationError that lists them all. Listing declares the nine features the pipeline was fitted on, and Quote declares the one thing the service returns. Why. The pipeline accepts any frame with the right column names and fails, or worse prices, whatever it is handed. A schema moves the check to the edge of the service, where a bad listing is refused with a reason a client can act on, before it reaches to_frame or the model. When. In front of every model that receives inputs from another program, written from the columns the model was trained on.

pydantic is already installed, because FastAPI cannot run without it: a FastAPI route declares its request body as a pydantic model. FastAPI accepts any pydantic from 2.9.0 on, though, and schema.py is the project's own pydantic code, whose error types and messages this lesson quotes, so the project chooses the version itself, the case where Lesson 6.1.1 pins a library directly, at the version uv.lock already holds:

uv add pydantic==2.14.1

Resolved 98 packages in 75ms
   Building listings @ file:///home/goku/dev/listings-service
      Built listings @ file:///home/goku/dev/listings-service
Prepared 1 package in 8ms
Uninstalled 1 package in 0.44ms
Installed 1 package in 0.61ms
 ~ listings==0.1.0 (from file:///home/goku/dev/listings-service)

The same 98 packages resolve, and the one reinstalled is the project itself. pyproject.toml gains "pydantic==2.14.1" among its dependencies, and uv.lock records it as one of the project's own.

Save src/listings/schema.py:

"""What one listing must look like before the model sees it."""

from typing import Literal

from pydantic import BaseModel, ConfigDict, Field


class Listing(BaseModel):
    model_config = ConfigDict(extra="forbid")

    area_sqm: float | None = Field(default=None, gt=0, le=1000)
    year_built: int | None = Field(default=None, ge=1800, le=2026)
    distance_km: float = Field(ge=0, le=100)
    rooms: int = Field(ge=1, le=50)
    bathrooms: int = Field(ge=1, le=20)
    neighborhood: Literal["Downtown", "Riverside", "Hillcrest", "Eastgate",
                          "Northend", "Lakeview", "Southgate"]
    property_type: Literal["apartment", "townhouse", "house"]
    energy_rating: Literal["A", "B", "C", "D", "E", "F", "G"] | None = None
    condition: Literal["poor", "fair", "good", "excellent"]


class Quote(BaseModel):
    price: int

The file stays free of comments because the class is its own documentation, and the lines read as follows. model_config = ConfigDict(extra="forbid") makes any key the class does not declare an error, where pydantic's default, "ignore", drops it silently. float | None is a type that admits a float or None, and = Field(default=None, ...) makes the field optional, so a listing that leaves out the area is valid and arrives as None, the blank Lesson 3.2.3's imputer fills. A field with no default, distance_km here, is required. gt, ge and le are greater than, greater or equal and less or equal. Literal[...], from the standard library's typing, admits exactly the values listed and nothing else: the seven neighborhoods, three property types, seven ratings and four conditions that listings.csv holds. The bounds are wider than the data, whose areas run from 24 to 278.5 square meters, because they refuse what cannot be a listing, a negative area or zero rooms, and not what the model has merely never seen.

A probe, saved as ~/dev/listings-probes/try_schema.py:

from pydantic import ValidationError

from listings.schema import Listing

listing = Listing(
    area_sqm=85.0, year_built=1998, distance_km=3.0, rooms=3, bathrooms=1,
    neighborhood="Riverside", property_type="apartment",
    energy_rating="C", condition="good",
)
print(listing)
print(listing.model_dump())

try:
    Listing(
        area_sqm="eighty", distance_km=3.0, rooms=0, bathrooms=1,
        neighborhood="Uptown", property_type="apartment",
        condition="good", listing_id=30001,
    )
except ValidationError as err:
    print(err)

uv run python ../listings-probes/try_schema.py

area_sqm=85.0 year_built=1998 distance_km=3.0 rooms=3 bathrooms=1 neighborhood='Riverside' property_type='apartment' energy_rating='C' condition='good'
{'area_sqm': 85.0, 'year_built': 1998, 'distance_km': 3.0, 'rooms': 3, 'bathrooms': 1, 'neighborhood': 'Riverside', 'property_type': 'apartment', 'energy_rating': 'C', 'condition': 'good'}
4 validation errors for Listing
area_sqm
  Input should be a valid number, unable to parse string as a number [type=float_parsing, input_value='eighty', input_type=str]
    For further information visit https://errors.pydantic.dev/2.14/v/float_parsing
rooms
  Input should be greater than or equal to 1 [type=greater_than_equal, input_value=0, input_type=int]
    For further information visit https://errors.pydantic.dev/2.14/v/greater_than_equal
neighborhood
  Input should be 'Downtown', 'Riverside', 'Hillcrest', 'Eastgate', 'Northend', 'Lakeview' or 'Southgate' [type=literal_error, input_value='Uptown', input_type=str]
    For further information visit https://errors.pydantic.dev/2.14/v/literal_error
listing_id
  Extra inputs are not permitted [type=extra_forbidden, input_value=30001, input_type=int]
    For further information visit https://errors.pydantic.dev/2.14/v/extra_forbidden

The valid listing is the complete Riverside apartment Lesson 3.2.3's Assignment priced, and model_dump() turns it back into a plain dictionary in the class's field order. The bad one fails four ways at once, and the error names each field, its rule as a type and the value that broke it: a word where a number belongs, zero rooms under the bound of 1, a neighborhood outside the seven, and listing_id, the CSV's own key, which the model was never fitted on. Every failure comes back in one error, so a client fixes all four in one round instead of discovering them one request at a time.

DON'T:

• DON'T leave extra at its default. With extra="ignore" the same class took a listing carrying listing_id and a price of 1, returned the nine features and said nothing, so a client sending the target itself would never learn the service ignored it.
• DON'T type rooms as a float to be lenient. to_frame's astype cast a 3.5 to the int64 3 without a word, and the champion priced it at 384235, the three-room price.
• DON'T copy the training data's minimum and maximum into the bounds. A bound refuses what cannot exist, and the 24 to 278.5 square meters of listings.csv is a fact about one sample.

What Lax Mode Converts

What. Pydantic validates in lax mode by default: a value of the wrong Python type is converted when the conversion loses nothing, and refused when it would. A string holding a whole number becomes an int, a string holding a number becomes a float, 3.0 becomes 3, and True becomes 1, while 3.5 is refused for an int field because the fraction would be lost. strict=True, passed to model_validate, the method that builds a model from a dictionary, turns these conversions off, and keeps one: an integer still passes for a float field, 85 for area_sqm arriving as 85.0. Why. Real clients send numbers as text, from HTML forms, CSV exports and spreadsheets, and refusing "3" for a room count would turn honest requests into errors. Knowing exactly which conversions happen is the difference between a schema you can reason about and one you hope about. When. Every time you read a 200 and want to know what the client actually sent, and every time you choose between lax and strict for a field.

~/dev/listings-probes/lax.py:

from pydantic import ValidationError

from listings.schema import Listing

LISTING = {
    "area_sqm": 85.0, "year_built": 1998, "distance_km": 3.0, "rooms": 3,
    "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment",
    "energy_rating": "C", "condition": "good",
}

for change in ({"rooms": "3"}, {"area_sqm": "85"}, {"rooms": 3.0},
               {"rooms": True}, {"rooms": 3.5}):
    try:
        listing = Listing(**(LISTING | change))
        print(change, "->", listing.rooms, listing.area_sqm)
    except ValidationError as err:
        print(change, "->", err.errors()[0]["type"])

try:
    Listing.model_validate(LISTING | {"rooms": "3"}, strict=True)
except ValidationError as err:
    print("strict ->", err.errors()[0]["type"])

uv run python ../listings-probes/lax.py

{'rooms': '3'} -> 3 85.0
{'area_sqm': '85'} -> 3 85.0
{'rooms': 3.0} -> 3 85.0
{'rooms': True} -> 1 85.0
{'rooms': 3.5} -> int_from_float
strict -> int_type

LISTING | change is the reference listing with one key replaced, and err.errors() is the list of failures as dictionaries, each with the type the printed error showed. Three conversions are the ones a form needs: "3" and "85" become numbers and 3.0 a whole room count. True passing as one room is lax mode's least obvious rule, inherited from Python, where True == 1, and it means a client that sends a boolean by mistake gets a valid listing with one room. 3.5 is refused with int_from_float, because no whole room count equals it, and the same "3" that lax mode accepted fails under strict=True with int_type.

DON'T:

• DON'T read a 200 as proof the client sent the declared type. Lax mode converted "3", 3.0 and True, and the service sees only the converted values.
• DON'T switch the whole schema to strict to stop one surprise. Strict refuses "3" too, the value a form most often sends.

Loading the Model Once in a Lifespan

What. A lifespan is a function FastAPI runs around the whole life of the app: its code up to yield runs once at startup, before the server accepts a single request, and its code after yield runs once at shutdown. Here it loads the model named by the MODEL_URI environment variable, by default the champion alias models:/listings-price@champion Lesson 6.2.2 set, and stores it on app.state, an object FastAPI keeps for exactly this, where every request can reach it. Why. Loading a model reads the registry and the artifact from disk and rebuilds the pipeline, which takes far longer than a prediction. Done once, the cost lands on startup, and a model that cannot load stops the service before it pretends to be up. When. For anything a service needs on every request and can build once: a model, a lookup table, a connection pool.

src/listings/app.py becomes the whole service, the lifespan and the prediction route together. The route's lines belong to the next subtitle.

"""The price service: one listing in, one price out."""

import os
from contextlib import asynccontextmanager

import mlflow
from fastapi import FastAPI, Request

from listings.data import to_frame
from listings.schema import Listing, Quote

DEFAULT_MODEL_URI = "models:/listings-price@champion"


@asynccontextmanager
async def lifespan(app):
    uri = os.environ.get("MODEL_URI", DEFAULT_MODEL_URI)
    app.state.model = mlflow.pyfunc.load_model(uri)
    yield


app = FastAPI(title="listings-service", lifespan=lifespan)


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


@app.post("/predict")
def predict(listing: Listing, request: Request) -> Quote:
    frame = to_frame([listing.model_dump()])
    price = request.app.state.model.predict(frame)[0]
    return Quote(price=round(price))

os.environ.get("MODEL_URI", DEFAULT_MODEL_URI) reads the environment variable, Module 0's named value the shell hands every process it starts, and falls back to the champion when it is unset, so the same code serves another version or a folder on disk without an edit. async def with a yield in it makes an async generator function, a function FastAPI's event loop can pause at the yield and resume after it. @asynccontextmanager, from the standard library, turns it into an async context manager, the asynchronous form of the object a with statement opens and closes, which is what FastAPI(lifespan=...) expects. The lifespan has nothing to release, so nothing follows its yield. The model is mlflow.pyfunc.load_model's, the loader Lesson 6.2.2 enforced the signature with.

In Terminal 1:

uv run uvicorn listings.app:app

INFO:     Started server process [122206]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

The lifespan runs between Waiting for application startup. and Application startup complete., and uvicorn binds port 8000 only after it. Lesson 6.2.3's app answered its first /health about half a second after the command, and this one takes about five, the time to import MLflow and load the model, and a curl sent inside those seconds fails with curl: (7) Failed to connect, exit code 7, because nothing listens yet. Wait for Uvicorn running on before the first request. The five seconds buy every request after them: a variant of this route that called load_model inside predict took 2.20 seconds on its first request and about 0.16 on every one after, where this service answers in 0.015 to 0.034 seconds, round trip from curl included.

Ctrl+C in Terminal 1 stops it, and the lifespan's exit runs in the middle of the shutdown:

^CINFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [122206]

Now point it at an alias the registry does not hold:

MODEL_URI=models:/listings-price@nobody uv run uvicorn listings.app:app; echo $?

The top and bottom of what it prints, with the ten MLflow frames between line 18 of app.py and the exception cut:

INFO:     Started server process [123520]
INFO:     Waiting for application startup.
ERROR:    Traceback (most recent call last):
  File "/home/goku/dev/listings-service/.venv/lib/python3.12/site-packages/starlette/routing.py", line 656, in lifespan
    async with self.lifespan_context(app) as maybe_state:
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/goku/.local/share/uv/python/cpython-3.12.8-linux-x86_64-gnu/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/goku/dev/listings-service/src/listings/app.py", line 18, in lifespan
    app.state.model = mlflow.pyfunc.load_model(uri)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
mlflow.exceptions.MlflowException: Registered model alias nobody not found.

ERROR:    Application startup failed. Exiting.
3

MODEL_URI=... uv run ... sets the variable for that one command only. Starlette, Lesson 6.2.3's toolkit under FastAPI, enters the lifespan with async with, line 18 of app.py raises, and uvicorn refuses to start, exits with code 3 and never binds the port. That is the behavior to want. The variant that loaded per request, given the same alias, started cleanly, answered /health with 200 and failed every /predict with a 500, Internal Server Error, a broken service that looked alive to anything checking its health.

DON'T:

• DON'T load the model inside the route. Every request paid about 0.16 seconds for it, and a missing alias became a 500 per request behind a /health that answered 200.
• DON'T load it at import time, at the top of app.py. A variant that did, run against an empty registry, stopped pytest at collection with 1 error during collection, before any fixture could point MODEL_URI at another model.
• DON'T send the first request the moment the command is typed. The port opens only after Application startup complete., about five seconds in.

POST /predict and Its 422s

What. @app.post("/predict") registers a route for POST requests, whose body carries the listing. A parameter annotated with a pydantic model, listing: Listing, tells FastAPI to read the JSON body and validate it before the function runs, and a parameter annotated Request receives the incoming request itself, whose request.app.state holds the model. The return annotation -> Quote is the response model: FastAPI validates what the function returns against it and publishes both schemas in the OpenAPI document behind /docs. Why. Validation that FastAPI runs before the function means the function body only ever sees a valid Listing, so nothing in it checks anything. A request that fails gets status 422, Unprocessable Entity, the HTTP code for a well-formed request whose content is wrong, with pydantic's errors as JSON, and the pipeline never runs. When. For every route that takes structured input, and every time a client asks why its request was refused.

The function body is three lines. listing.model_dump() gives the validated dictionary, to_frame, Lesson 6.2.2's, builds a one-row frame with the training columns and dtypes, predict returns an array whose first value is the price, and round makes it a whole number for Quote. Start the service again in Terminal 1, wait for Uvicorn running on, and in Terminal 2:

LISTING='{"area_sqm": 85.0, "year_built": 1998, "distance_km": 3.0, "rooms": 3, "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment", "energy_rating": "C", "condition": "good"}'
curl -i -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d "$LISTING"

HTTP/1.1 200 OK
date: Tue, 03 Nov 2026 12:22:08 GMT
server: uvicorn
content-length: 16
content-type: application/json

{"price":384235}

-H adds a request header, here the one that says the body is JSON, and -d sends the body. 384235 is the price Lesson 3.2.3's Assignment gave the same complete listing from its saved pipeline, so the service and the training script agree. The same listing with the area, the year and the rating left out:

curl -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d '{"distance_km": 3.0, "rooms": 3, "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment", "condition": "good"}'

{"price":434348}

The three optional fields arrived as None, to_frame made them blanks of the training dtypes, and the imputer filled them, Lesson 3.2.3's 434348 to the unit. The bad listing of the first subtitle, now over HTTP:

curl -i -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d '{"area_sqm": "eighty", "distance_km": 3.0, "rooms": 0, "bathrooms": 1, "neighborhood": "Uptown", "property_type": "apartment", "condition": "good", "listing_id": 30001}'

HTTP/1.1 422 Unprocessable Entity
date: Tue, 03 Nov 2026 12:22:09 GMT
server: uvicorn
content-length: 685
content-type: application/json

{"detail":[{"type":"float_parsing","loc":["body","area_sqm"],"msg":"Input should be a valid number, unable to parse string as a number","input":"eighty"},{"type":"greater_than_equal","loc":["body","rooms"],"msg":"Input should be greater than or equal to 1","input":0,"ctx":{"ge":1}},{"type":"literal_error","loc":["body","neighborhood"],"msg":"Input should be 'Downtown', 'Riverside', 'Hillcrest', 'Eastgate', 'Northend', 'Lakeview' or 'Southgate'","input":"Uptown","ctx":{"expected":"'Downtown', 'Riverside', 'Hillcrest', 'Eastgate', 'Northend', 'Lakeview' or 'Southgate'"}},{"type":"extra_forbidden","loc":["body","listing_id"],"msg":"Extra inputs are not permitted","input":30001}]}

detail holds one entry per failure, the same four the probe raised, each with its type, its loc, the location as a path that starts at body, a msg, the input that broke it and, where a rule has a parameter, a ctx holding it. A listing that leaves out distance_km comes back with one entry:

curl -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d '{"area_sqm": 85.0, "year_built": 1998, "rooms": 3, "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment", "energy_rating": "C", "condition": "good"}'

{"detail":[{"type":"missing","loc":["body","distance_km"],"msg":"Field required","input":{"area_sqm":85.0,"year_built":1998,"rooms":3,"bathrooms":1,"neighborhood":"Riverside","property_type":"apartment","energy_rating":"C","condition":"good"}}]}

And the complete listing sent without the Content-Type header, which curl -d then labels as form data:

curl -X POST http://127.0.0.1:8000/predict -d "$LISTING"

{"detail":[{"type":"model_attributes_type","loc":["body"],"msg":"Input should be a valid dictionary or object to extract fields from","input":"{\"area_sqm\": 85.0, \"year_built\": 1998, \"distance_km\": 3.0, \"rooms\": 3, \"bathrooms\": 1, \"neighborhood\": \"Riverside\", \"property_type\": \"apartment\", \"energy_rating\": \"C\", \"condition\": \"good\"}"}]}

Without the header FastAPI does not parse the body as JSON, so Listing receives one string, and loc is ["body"] alone: the whole body is the wrong shape, and no field was even looked at. Terminal 1 logs each request with its status, the 422s as "POST /predict HTTP/1.1" 422 Unprocessable Entity. The schema a client reads is in the OpenAPI document, the JSON behind /docs:

curl -s http://127.0.0.1:8000/openapi.json | uv run python -c "import json, sys; schemas = json.load(sys.stdin)['components']['schemas']; print(schemas['Listing']['properties']['rooms']); print(schemas['Listing']['required']); print(schemas['Listing']['additionalProperties']); print(schemas['Quote'])"

{'type': 'integer', 'maximum': 50.0, 'minimum': 1.0, 'title': 'Rooms'}
['distance_km', 'rooms', 'bathrooms', 'neighborhood', 'property_type', 'condition']
False
{'properties': {'price': {'type': 'integer', 'title': 'Price'}}, 'type': 'object', 'required': ['price'], 'title': 'Quote'}

The bounds, the six required fields, extra="forbid" as additionalProperties: False and the response's one integer are all published from the two classes, so a client team reads the contract from the running service instead of from your code. Stop the service with Ctrl+C.

DON'T:

• DON'T leave out -H "Content-Type: application/json". Without it the whole body arrived as one string and came back 422 model_attributes_type at ["body"].
• DON'T return the bare NumPy value from the route. Quote(price=round(price)) is what the response model publishes, and a client reads {"price":384235} and nothing else.
• DON'T answer a bad listing with a 500. A 422 names the field, the rule and the input, and a 500 says only that the service broke.

Testing the Service Without the Registry

What. The app's tests now run against a model a session fixture fits and saves into a temporary folder with mlflow.sklearn.save_model, which writes the same model folder log_model writes but straight to a path, with no tracking run and no registry. A function fixture points MODEL_URI at that folder through monkeypatch.setenv, the environment-variable form of the monkeypatch fixture Lesson 6.1.2's Assignment used, which sets the variable for one test and restores it afterwards, and opens TestClient(app) in a with block, which is what runs the lifespan. Why. The registry is local state: its champion moves when someone promotes, and a fresh clone or a CI machine has no mlflow.db at all. Tests that load their own model, trained the same way as version 1 from the same split, give the same prices on every machine, and the with block exercises the lifespan exactly as uvicorn does. When. For every test of an app that loads something at startup.

tests/conftest.py gains two imports and the model_dir fixture:

import mlflow
import pytest
from mlflow.models import infer_signature

from listings.data import load, split
from listings.features import build_pipeline


@pytest.fixture(scope="session")
def listings():
    return load()


@pytest.fixture(scope="session")
def split_data(listings):
    return split(listings)


@pytest.fixture(scope="session")
def trained(split_data):
    X_train, _, y_train, _ = split_data
    return build_pipeline().fit(X_train, y_train)


@pytest.fixture(scope="session")
def model_dir(trained, split_data, tmp_path_factory):
    X_train = split_data[0]
    path = tmp_path_factory.mktemp("mlflow") / "model"
    mlflow.sklearn.save_model(
        trained,
        path,
        signature=infer_signature(X_train, trained.predict(X_train)),
        skops_trusted_types=["numpy.dtype"],
    )
    return path

tmp_path_factory is the session-wide form of Lesson 6.1.2's tmp_path: mktemp("mlflow") makes a fresh temporary folder that lives for the whole run, so the model is saved once for every test. The signature and skops_trusted_types are train.py's own, so this folder loads exactly as a registered version does. tests/test_app.py is rewritten:

import pytest
from fastapi.testclient import TestClient

from listings.app import app

LISTING = {
    "area_sqm": 85.0, "year_built": 1998, "distance_km": 3.0, "rooms": 3,
    "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment",
    "energy_rating": "C", "condition": "good",
}


@pytest.fixture
def client(model_dir, monkeypatch):
    monkeypatch.setenv("MODEL_URI", str(model_dir))
    with TestClient(app) as client:
        yield client


def test_health(client):
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}


def test_predict(client):
    response = client.post("/predict", json=LISTING)
    assert response.status_code == 200
    assert response.json() == {"price": 384235}


def test_blank_fields_are_priced(client):
    blank = {k: v for k, v in LISTING.items()
             if k not in ("area_sqm", "year_built", "energy_rating")}
    response = client.post("/predict", json=blank)
    assert response.json() == {"price": 434348}


@pytest.mark.parametrize("change", [
    {"rooms": 0},
    {"neighborhood": "Uptown"},
    {"area_sqm": "eighty"},
    {"listing_id": 30001},
], ids=["no-rooms", "unknown-neighborhood", "text-area", "extra-field"])
def test_bad_listing_is_422(client, change):
    response = client.post("/predict", json=LISTING | change)
    assert response.status_code == 422


def test_missing_field_is_422(client):
    incomplete = {k: v for k, v in LISTING.items() if k != "distance_km"}
    assert client.post("/predict", json=incomplete).status_code == 422

The module-level client of Lesson 6.2.3 becomes a fixture. A fixture that yields hands its value to the test at the yield and runs the rest of its body after the test ends, so each test gets a client inside an open with block, and the block closes, running the lifespan's shutdown, when the test is done. client.post(..., json=LISTING) sends the dictionary as a JSON body with the JSON header set. The 384235 and 434348 asserted here are the champion's own prices, because version 1 and trained are the same build_pipeline() fitted on the same 2,400 rows, and a linear regression fitted twice on the same rows lands on the same coefficients. The four bad listings run as one parametrized test, Lesson 6.1.2's parametrize with readable ids. The whole suite:

uv run pytest

The report's collection and per-file lines, and its last line:

collected 26 items

tests/test_app.py ........                                               [ 30%]
tests/test_data.py ....                                                  [ 46%]
tests/test_features.py .....                                             [ 65%]
tests/test_model.py .........                                            [100%]

======================== 26 passed, 2 warnings in 3.01s ========================

Lesson 6.2.3's 20 tests lost test_unknown_route_is_404 and gained seven, so 26 pass. The two warnings in the summary between those lines are Lesson 6.2.3's Starlette notice about httpx2, and a new one, MLflow's hint from infer_signature in model_dir that integer columns cannot hold missing values. The integer columns are rooms and bathrooms, and Listing makes both required, so no request reaches the model with either one blank.

Now break the fixture the way a hurried test is written, by replacing the with block's two lines with return TestClient(app), and run the app's tests:

uv run pytest tests/test_app.py

Three lines of the report and its last:

tests/test_app.py .FF.....                                               [100%]
FAILED tests/test_app.py::test_predict - AttributeError: 'State' object has n...
FAILED tests/test_app.py::test_blank_fields_are_priced - AttributeError: 'Sta...
=================== 2 failed, 6 passed, 2 warnings in 3.01s ====================

The traceback of each failure ends at price = request.app.state.model.predict(frame)[0] with AttributeError: 'State' object has no attribute 'model'. A TestClient used without with sends requests but never runs the lifespan, so app.state has no model. Six tests still pass: /health never touches the model, and the five 422 tests are refused by validation before predict runs, so they pass with no model at all. Put the with block back and the 26 pass again, then commit the lesson's work, the four project files with pyproject.toml and uv.lock:

git add pyproject.toml uv.lock src/listings/schema.py src/listings/app.py tests/conftest.py tests/test_app.py
git commit -m "feat: the prediction service with a validated listing"

[main 7c80595] feat: the prediction service with a validated listing
 6 files changed, 110 insertions(+), 7 deletions(-)
 create mode 100644 src/listings/schema.py

The hash is your commit's own and differs from this one.

DON'T:

• DON'T create a TestClient without with for an app that has a lifespan. Two tests failed with 'State' object has no attribute 'model', and the six that passed never needed the model.
• DON'T take a passing 422 test as evidence that the service can predict. Validation refuses the request before the route's body runs.
• DON'T point the tests at models:/listings-price@champion. Every promote would change what the suite tests, and a machine with no mlflow.db could not run it.

Lab Exercises

Lab Exercise 1: Every Training Listing Through the Schema

A schema that refuses listings the model was trained on would refuse real clients too. In ~/dev/listings-probes/check_rows.py, write a function failures(rows) that builds a Listing from each dictionary in rows, catches every ValidationError, and returns a Counter of (field, error type) pairs as most_common(), the field being the first entry of each error's loc. Run it three times: on every column of listings.csv as records, load().to_dict("records"), on the nine FEATURES columns only, and on the nine with every blank removed from each dictionary, using pd.notna. Print each result, and the number of rows in the last one. Say in a comment what each run's errors mean and why the third passes.

Lab Exercise 2: Another Version, and No Registry at All

Serve version 2, Lesson 6.2.2's model without the missing indicators, by starting the service with MODEL_URI=models:/listings-price/2, Lesson 6.2.2's address by version, and price the complete reference listing and the one with the area, the year and the rating blank. Then show that the app's tests never read the registry. MLflow reads its tracking database from the MLFLOW_TRACKING_URI variable when it is set, and sqlite:///mlflow.db otherwise. Run tests/test_app.py with MLFLOW_TRACKING_URI=sqlite:////tmp/empty.db, a database that does not exist yet, check whether the file exists afterwards, then start the service under the same variable, and remove the file at the end. Say in a comment what the two versions' prices show and what each half of the second part proves.

from collections import Counter

import pandas as pd
from pydantic import ValidationError

from listings.data import FEATURES, load
from listings.schema import Listing


def failures(rows):
    errors = Counter()
    for row in rows:
        try:
            Listing(**row)
        except ValidationError as err:
            errors.update((e["loc"][0], e["type"]) for e in err.errors())
    return errors.most_common()


listings = load()
print(failures(listings.to_dict("records")))
rows = listings[FEATURES].to_dict("records")
print(failures(rows))
present = [{k: v for k, v in row.items() if pd.notna(v)} for row in rows]
print(len(present), failures(present))
# Every column: the four that are not features are refused in all 3,000
# rows, price among them. Nine features: pandas writes a blank as NaN,
# a float, which is not None, so the 267 blank ratings, 140 years and
# 125 areas fail. Blanks dropped: the optional fields default to None.

uv run python ../listings-probes/check_rows.py

[(('listing_id', 'extra_forbidden'), 3000), (('listing_date', 'extra_forbidden'), 3000), (('price', 'extra_forbidden'), 3000), (('sold_in_30_days', 'extra_forbidden'), 3000), (('energy_rating', 'literal_error'), 267), (('year_built', 'finite_number'), 140), (('area_sqm', 'less_than_equal'), 125)]
[(('energy_rating', 'literal_error'), 267), (('year_built', 'finite_number'), 140), (('area_sqm', 'less_than_equal'), 125)]
3000 []

MODEL_URI=models:/listings-price/2 uv run uvicorn listings.app:app

curl -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d '{"area_sqm": 85.0, "year_built": 1998, "distance_km": 3.0, "rooms": 3, "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment", "energy_rating": "C", "condition": "good"}'
curl -X POST http://127.0.0.1:8000/predict -H "Content-Type: application/json" -d '{"distance_km": 3.0, "rooms": 3, "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment", "condition": "good"}'

{"price":385826}{"price":392147}

MLFLOW_TRACKING_URI=sqlite:////tmp/empty.db uv run pytest tests/test_app.py -q
ls /tmp/empty.db

........                                                                 [100%]
8 passed, 2 warnings in 2.53s
ls: cannot access '/tmp/empty.db': No such file or directory

MLFLOW_TRACKING_URI=sqlite:////tmp/empty.db uv run uvicorn listings.app:app
rm /tmp/empty.db

INFO:     Started server process [129568]
INFO:     Waiting for application startup.
2026/11/03 13:26:59 INFO mlflow.store.db.utils: Creating initial MLflow database tables...
2026/11/03 13:26:59 INFO mlflow.store.db.utils: Updating database tables
ERROR:    Traceback (most recent call last):
mlflow.exceptions.MlflowException: Registered Model with name=listings-price not found

ERROR:    Application startup failed. Exiting.

Theory Assessment

Q1: Three Requests, Three Answers

Three requests reach /predict, each the reference listing changed once: the first leaves out area_sqm, the second leaves out distance_km, and the third adds "price": 412000. Say which get a price and which a 422, with the error type, what in Listing decides each, and what the third request would get if model_config were deleted.

Hint: Read which fields have a default and what extra does when it is not set.

Q2: The Load Moved Into the Route

A teammate moves mlflow.pyfunc.load_model from the lifespan into predict, so that "a promoted version is served without a restart". Say what each request then costs, what a misspelled alias does in each design, and what the lifespan design serves after promote.py moves champion while the service runs.

Hint: The lifespan runs once. Ask when the registry is read in each design.

Q3: Six Passed Without the Model

With the client fixture written as return TestClient(app), the app's tests read 2 failed, 6 passed. Say which two fail and with what, why the other six pass, and what this says about the five 422 tests as evidence for the service.

Hint: Ask which tests reach the line request.app.state.model.predict(frame), and when FastAPI validates the body.

Q4: "The Schema Refuses Every Wrong Type"

A teammate says the schema refuses any wrongly typed field and points at "area_sqm": "eighty" coming back 422. A client then sends the reference listing four times with rooms as "3", 3.0, true and 3.5. Say what each request gets back, with the price where there is one, what the teammate's sentence gets wrong, and what 3.5 would be priced at if rooms were declared float.

Hint: Lax mode converts what it can convert without loss, and to_frame casts rooms to int64. A one-room version of the reference listing over HTTP comes back at 342612.

Assignment

Assignment: A Batch Route Beside the Service

The nightly revaluation job prices thousands of listings, and one HTTP request per listing spends most of its time on everything except the model. Write ~/dev/listings-probes/batch_service.py, a trial service kept outside the project until it has earned a place, that imports lifespan from listings.app so it loads the model the same way, and serves two routes: POST /predict/batch, which takes a JSON list of listings and returns a list of quotes priced with one predict call on one frame, and POST /predict, which prices one listing through the batch function.

A list body needs two things the lesson did not use. list[Listing] as a parameter's type makes FastAPI read a JSON array and validate each element as a Listing, and the response annotation list[Quote] publishes a list of quotes. Annotated[list[Listing], Field(min_length=1, max_length=100)], with Annotated from typing, attaches length bounds to the list itself, so an empty list and a list of 101 listings are refused before any element is validated. When an element fails, its error's loc carries the element's index between body and the field name.

In the same file, with a TestClient used correctly and the champion loaded by default, send the reference listing, its blank version and a five-room listing in poor condition as one batch and one at a time, and print the batch's quotes and whether they equal the single ones. Send a batch whose second listing has zero rooms and print the status with each error's type and loc. Print the status and the error type for an empty list and for 101 listings. Time one batch of 100 reference listings against 100 single requests. Say in a comment what the loc path tells a client and why the batch is faster.

Hints:

• predict returns one price per row, so the batch route builds the frame from [listing.model_dump() for listing in listings] and returns one Quote per price.
• time.perf_counter() before and after each part gives elapsed seconds, and [LISTING] * 100 makes the list.

import time
from typing import Annotated

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from pydantic import Field

from listings.app import lifespan
from listings.data import to_frame
from listings.schema import Listing, Quote

app = FastAPI(title="listings-batch", lifespan=lifespan)
Batch = Annotated[list[Listing], Field(min_length=1, max_length=100)]


@app.post("/predict/batch")
def predict_batch(listings: Batch, request: Request) -> list[Quote]:
    frame = to_frame([listing.model_dump() for listing in listings])
    prices = request.app.state.model.predict(frame)
    return [Quote(price=round(price)) for price in prices]


@app.post("/predict")
def predict(listing: Listing, request: Request) -> Quote:
    return predict_batch([listing], request)[0]


LISTING = {
    "area_sqm": 85.0, "year_built": 1998, "distance_km": 3.0, "rooms": 3,
    "bathrooms": 1, "neighborhood": "Riverside", "property_type": "apartment",
    "energy_rating": "C", "condition": "good",
}
blank = {k: v for k, v in LISTING.items()
         if k not in ("area_sqm", "year_built", "energy_rating")}
batch = [LISTING, blank, LISTING | {"rooms": 5, "condition": "poor"}]

with TestClient(app) as client:
    together = client.post("/predict/batch", json=batch).json()
    alone = [client.post("/predict", json=one).json() for one in batch]
    print(together, together == alone)

    bad = client.post("/predict/batch", json=[LISTING, LISTING | {"rooms": 0}])
    print(bad.status_code, [(e["type"], e["loc"]) for e in bad.json()["detail"]])
    for size in (0, 101):
        reply = client.post("/predict/batch", json=[LISTING] * size)
        print(size, reply.status_code, reply.json()["detail"][0]["type"])

    start = time.perf_counter()
    client.post("/predict/batch", json=[LISTING] * 100)
    one_call = time.perf_counter() - start
    start = time.perf_counter()
    for _ in range(100):
        client.post("/predict", json=LISTING)
    print(f"{one_call * 1000:.0f} ms batched, {(time.perf_counter() - start) * 1000:.0f} ms one by one")
# loc ["body", 1, "rooms"] names the second listing's rooms, so a client
# fixes one row and resends. The batch pays the per-call overhead of
# the route and the model's input checks once, not once per listing.

uv run python ../listings-probes/batch_service.py

/home/goku/dev/listings-service/.venv/lib/python3.12/site-packages/fastapi/testclient.py:1: StarletteDeprecationWarning: Using `httpx` with `starlette.testclient` is deprecated; install `httpx2` instead.
  from starlette.testclient import TestClient as TestClient  # noqa
[{'price': 384235}, {'price': 434348}, {'price': 358338}] True
422 [('greater_than_equal', ['body', 1, 'rooms'])]
0 422 too_short
101 422 too_long
21 ms batched, 1666 ms one by one

Additional Resources

Optional, and nothing later in the module relies on them. Worth a look for the full set of pydantic's conversions or for reshaping the 422 a client sees.

• Pydantic documentation - "Conversion Table". Every input type pydantic accepts for every field type, marked for lax and strict mode and for Python and JSON input, str to int and bool to int among the lax-only rows.
• FastAPI documentation - "Handling Errors", Override request validation exceptions. Replacing the default handler of RequestValidationError to change the status code or the body of a validation failure.

Summary

The service is three files of very little code, and almost none of it is about the model. The pipeline from Module 3 still does all the pricing, untouched since Lesson 6.1.1 restated it. What changed is where the questions get asked: whether a listing is a listing, whether the model exists, and whether the tests mean what they say are now answered before a price is computed, at the schema, at startup and in a fixture, instead of after something has already gone wrong inside predict.

Each of those boundaries has an edge you only see by pushing on it. The schema refuses eighty and quietly accepts true as one room. The lifespan turns a missing alias into a service that will not start, which looks worse than one that starts and fails per request, and is the better of the two. A test client that never ran the lifespan still passed six of eight tests, because the tests that passed never needed the model. In every case the comfortable reading of a green result was wrong, and the honest one came from asking what that result could not have seen.

That is the shape of the whole module from here. Containers, a database and monitoring all add places where something can be healthy while something else is broken, and the habit this lesson practiced, sending the request that should fail and reading exactly how it fails, is what makes those places visible.

Cheat Sheet

Part A - Reference Table

Part B - Compact Code Block

@asynccontextmanager
async def lifespan(app):
    uri = os.environ.get("MODEL_URI", DEFAULT_MODEL_URI)
    app.state.model = mlflow.pyfunc.load_model(uri)
    yield  # shutdown code goes after


@app.post("/predict")
def predict(listing: Listing, request: Request) -> Quote:
    # to_frame gives the training dtypes
    frame = to_frame([listing.model_dump()])
    price = request.app.state.model.predict(frame)[0]
    return Quote(price=round(price))

@pytest.fixture
def client(model_dir, monkeypatch):
    monkeypatch.setenv("MODEL_URI", str(model_dir))
    with TestClient(app) as client:  # runs the lifespan
        yield client