Project, Part 1: Build a Data Explorer API

Time to put the whole tutorial to work in one project. Over this page and the next you will build, test and publish the BitByte Data Explorer API: a small FastAPI backend that calls another API (the BitByte Practice API), checks what comes back, reshapes it into clean answers and hides the upstream's failures behind clear error messages. On this page you plan it and build it on a feature branch; on the next you test it, document it and merge it through a pull request.

This shape is everywhere in AI work. A chatbot backend calls an LLM provider; a RAG service calls a vector database and an embedding API; an agent's "tools" are small endpoints that call other systems. In every case your backend sits between a client and an upstream service you do not control, and its job is the same: never hang, never pass on garbage, and always answer with something the client can understand.

What you will learn

• How to turn a short brief into a list of endpoints, response shapes and error rules before writing code.
• How to set up a project with uv and Git, and do the work on a feature branch in small commits.
• How to keep all upstream calls in one module, with a timeout on every request, pagination, validation of what comes back and one error type for every kind of failure.
• How to expose clean FastAPI endpoints with typed parameters, response models and correct status codes (404, 422, 502, 504).
• How to check the finished API with curl and break it on purpose to see the error paths work.

The brief

The BitByte team wants a small "data explorer" service for its website and, later, for an AI study assistant. The assistant must not talk to the Practice API directly: it should get short, predictable answers from one place. The brief, as the product owner wrote it:

"List our courses, filter them by level and rating, show one course, give a per-level summary, and show today's weather for a city with a one-line advice. If the data source is slow or down, say so clearly instead of crashing."

Turn that into a contract first. This table is what you build, and it is also what the README on the next page will document:

And the rules that apply to every endpoint, the "non-functional requirements":

• Every upstream call has a timeout. A slow upstream must not make our server hang. Too slow → 504 Gateway Timeout.
• Upstream failures become 502 Bad Gateway with a short detail message: unreachable, 5xx, a body that is not JSON, or data in an unexpected shape. The client never sees a Python traceback or a 500.
• Upstream data is validated with Pydantic before we use it. Data you receive from another system is input, just like a user's request.
• Configuration comes from environment variables (PRACTICE_API_URL, UPSTREAM_TIMEOUT), so the same code runs on a laptop and on a server.
• The work happens on a feature branch in small commits, ready for review.

The design

Three files with one job each. Routes do not know how HTTP calls are made, and the upstream client does not know about FastAPI:

data-explorer/
├── pyproject.toml      dependencies and the FastAPI entrypoint (managed by uv)
├── uv.lock             exact versions of every package
├── .gitignore
└── app/
    ├── __init__.py     makes app/ a package
    ├── schemas.py      Pydantic models: what we send out, what we accept from upstream
    ├── upstream.py     every call to the Practice API: timeouts, errors, pagination, validation
    └── main.py         the FastAPI app: routes, parameters, status codes

  client (browser, curl, AI assistant)
        │  GET /weather/sylhet
        ▼
  Data Explorer API  :8000   main.py ──► upstream.py
        │  GET /v1/weather?city=sylhet   (timeout 5 s)
        ▼
  BitByte Practice API  :8800

Each kind of upstream trouble maps to one answer from our API:

Step 1: The project and the repository

Make sure the Practice API is running in its own terminal (fastapi run practice_api.py --port 8800, from Web APIs and REST: Reading Docs and Testing with curl). Then create the project folder, a Git repository and a uv project:

mkdir -p ~/projects/data-explorer
cd ~/projects/data-explorer
git init
uv init --bare --name data-explorer
uv add -q "fastapi[standard]" requests
cat pyproject.toml

Initialized empty Git repository in /home/nadia/projects/data-explorer/.git/
Initialized project `data-explorer`
[project]
name = "data-explorer"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi[standard]>=0.143.0",
    "requests>=2.34.2",
]

uv init --bare writes only pyproject.toml (no sample main.py or README), and uv add creates .venv, installs the packages and writes the exact versions to uv.lock. Your requires-python line shows the Python you have (3.14 is the course standard; 3.12 or newer works). With pip instead of uv, the same setup is:

python -m venv .venv
source .venv/bin/activate          # Windows PowerShell: .venv\Scripts\Activate.ps1
pip install "fastapi[standard]" requests
pip freeze > requirements.txt

Tell the FastAPI command where the app lives, so you can type fastapi run without a file name. This is the setting FastAPI's own docs now recommend:

cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF
mkdir app
touch app/__init__.py

Before the first commit, the .gitignore (see .gitignore: Keeping Secrets and Junk Out). The virtual environment, caches, logs and real .env files never go into Git:

# file: .gitignore
# Python
__pycache__/
*.pyc

# Virtual environment (rebuilt from uv.lock with `uv sync`)
.venv/

# Secrets and local settings
.env

# Logs from local runs
*.log

Commit the skeleton on main, then start a feature branch for the real work:

git add .
git status --short
git commit -q -m "Set up the Data Explorer project with uv"
git switch -c feature/data-explorer

A  .gitignore
A  app/__init__.py
A  pyproject.toml
A  uv.lock
Switched to a new branch 'feature/data-explorer'

Notice what git status --short listed: .venv is missing, as it should be. uv.lock is committed: it is what lets a teammate, a CI server or a Docker build install exactly the same versions (uv sync).

Step 2: Schemas first

Write down the shapes before the logic. Course is used twice: to validate what the Practice API sends and to describe what we send out. UpstreamWeather uses the Practice API's field names; Weather uses ours.

# file: app/schemas.py
"""The shapes of the data this API sends out and accepts from upstream."""
from typing import Literal

from pydantic import BaseModel, Field

Level = Literal["beginner", "intermediate", "advanced"]


class Course(BaseModel):
    id: int = Field(ge=1)
    title: str = Field(min_length=1)
    level: Level
    hours: int = Field(ge=0)
    rating: float = Field(ge=0, le=5)


class CourseList(BaseModel):
    count: int
    items: list[Course]


class LevelSummary(BaseModel):
    level: Level
    courses: int
    total_hours: int
    average_rating: float


class UpstreamWeather(BaseModel):
    """What the Practice API sends for /v1/weather (its names, not ours)."""
    city: str
    temp_c: float
    humidity: int
    condition: str


class Weather(BaseModel):
    city: str
    temperature_c: float
    humidity_pct: int = Field(ge=0, le=100)
    condition: str
    advice: str

Why a Literal for the level? It is a closed list: FastAPI will reject ?level=expert with a 422 before your code runs, the allowed values appear in /docs, and a course from upstream with an unknown level fails validation instead of slipping through. Check the model on a good and a bad record:

from pydantic import ValidationError
from app.schemas import Course

print(Course.model_validate({"id": 8, "title": "RAG Systems", "level": "advanced", "hours": 10, "rating": 4.9}))
try:
    Course.model_validate({"id": 8, "title": "RAG Systems", "level": "expert", "hours": "ten", "rating": 4.9})
except ValidationError as exc:
    print(exc.error_count(), "errors:", [e["loc"][0] for e in exc.errors()])

id=8 title='RAG Systems' level='advanced' hours=10 rating=4.9
2 errors: ['level', 'hours']

git add app/schemas.py
git commit -q -m "Add response and upstream schemas"

Step 3: The upstream client

This file is the only place that knows the Practice API exists. If it moves, changes its field names or gets replaced by a real data provider, this is the one file you change.

# file: app/upstream.py
"""Everything that talks to the BitByte Practice API lives in this one file."""
import os

import requests
from pydantic import BaseModel, ValidationError

from app.schemas import Course, UpstreamWeather

BASE_URL = os.environ.get("PRACTICE_API_URL", "http://localhost:8800")
TIMEOUT_SECONDS = float(os.environ.get("UPSTREAM_TIMEOUT", "5"))
MAX_PAGES = 50

session = requests.Session()
session.headers["User-Agent"] = "bitbyte-data-explorer/0.1"


class UpstreamError(Exception):
    """The Practice API was unreachable, too slow, or sent something we cannot use."""

    def __init__(self, message: str, status_code: int = 502):
        super().__init__(message)
        self.message = message
        self.status_code = status_code


def get_json(path: str, params: dict | None = None) -> dict | None:
    """GET a Practice API path and return the JSON body, or None for a 404."""
    try:
        response = session.get(f"{BASE_URL}{path}", params=params, timeout=TIMEOUT_SECONDS)
    except requests.Timeout:
        raise UpstreamError(f"Practice API did not answer within {TIMEOUT_SECONDS:g} s", 504)
    except requests.ConnectionError:
        raise UpstreamError("Practice API is unreachable")
    if response.status_code == 404:
        return None
    if not response.ok:
        raise UpstreamError(f"Practice API failed with status {response.status_code}")
    try:
        return response.json()
    except ValueError:
        raise UpstreamError("Practice API sent a body that is not JSON")


def parse(model: type[BaseModel], data: dict):
    """Validate upstream data before we trust it."""
    try:
        return model.model_validate(data)
    except ValidationError:
        raise UpstreamError(f"Practice API sent {model.__name__} data in an unexpected shape")


def fetch_courses(level: str | None = None) -> list[Course]:
    """All courses (optionally of one level), following next_page to the end."""
    courses, page = [], 1
    while page is not None:
        if page > MAX_PAGES:
            raise UpstreamError("Practice API returned too many pages")
        data = get_json("/v1/courses", {"level": level, "page": page, "per_page": 20})
        if data is None:
            raise UpstreamError("Practice API has no /v1/courses endpoint")
        courses += [parse(Course, item) for item in data["items"]]
        page = data["next_page"]
    return courses


def fetch_course(course_id: int) -> Course | None:
    data = get_json(f"/v1/courses/{course_id}")
    return None if data is None else parse(Course, data)


def fetch_weather(city: str) -> UpstreamWeather | None:
    data = get_json("/v1/weather", {"city": city})
    return None if data is None else parse(UpstreamWeather, data)

The decisions in this file, one by one:

• Settings from the environment, with a sensible default. On a server the upstream will live at another address; nobody should edit code to point at it.
• One requests.Session for the whole app: it reuses connections (faster) and sends our User-Agent on every call, so the upstream's operators can see who is calling (see Reliable API Calls: Timeouts, Retries, Pagination and LLM APIs).
• timeout= on every request. Without it, requests waits forever, and every stuck request holds one of your server's worker threads.
• One exception type, UpstreamError, carrying the status code our API should answer with. Callers catch one thing, not five requests exceptions.
• 404 is not an error here. "No such course" is a normal answer, so get_json returns None and the route decides what to tell the client.
• Pagination with a safety limit. The loop follows next_page until it is null; MAX_PAGES stops a buggy upstream from keeping you in the loop forever.
• parse validates everything that comes in. Notice the error message names the model but does not include the upstream's data: error messages end up in logs and in front of users.

Point the app at the Practice API through the environment variable (the same address as the default, but now you can see where it comes from), and try the client from Python before any route exists:

export PRACTICE_API_URL=http://localhost:8800
export UPSTREAM_TIMEOUT=5

from app import upstream

courses = upstream.fetch_courses()
print(len(courses), "courses; first:", courses[0])
print(upstream.fetch_course(99))
print(upstream.fetch_weather("Sylhet"))

10 courses; first: id=1 title='How AI Works' level='beginner' hours=3 rating=4.7
None
city='Sylhet' temp_c=27.4 humidity=85 condition='Rain'

All 10 courses arrived even though the Practice API sends at most 20 per page and defaults to 5: fetch_courses asked for 20 per page and followed next_page. Course 99 came back as None, not as an exception.

Now the failure paths. The Practice API has a /v1/slow endpoint that waits as long as you ask; lower the timeout to one second and ask it to wait two:

upstream.TIMEOUT_SECONDS = 1
try:
    upstream.get_json("/v1/slow", {"seconds": 2})
except upstream.UpstreamError as exc:
    print(exc.status_code, exc.message)
upstream.TIMEOUT_SECONDS = 5

try:
    upstream.parse(upstream.Course, {"id": 1, "title": "How AI Works", "level": "beginner"})
except upstream.UpstreamError as exc:
    print(exc.status_code, exc.message)

504 Practice API did not answer within 1 s
502 Practice API sent Course data in an unexpected shape

After one second the client gave up and turned the timeout into a 504 error with a clear message. A course without hours and rating was refused instead of being passed on half-empty. Commit the client:

git add app/upstream.py
git commit -q -m "Add Practice API client with timeouts, pagination and validation"

Step 4: The endpoints

# file: app/main.py
"""BitByte Data Explorer API: clean, validated endpoints over the BitByte Practice API."""
from typing import Annotated, Literal

from fastapi import FastAPI, HTTPException, Path, Query, Request
from fastapi.responses import JSONResponse

from app import upstream
from app.schemas import Course, CourseList, Level, LevelSummary, Weather

app = FastAPI(title="BitByte Data Explorer API", version="0.1.0")


@app.exception_handler(upstream.UpstreamError)
def upstream_failed(request: Request, exc: upstream.UpstreamError) -> JSONResponse:
    return JSONResponse({"detail": exc.message}, status_code=exc.status_code)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/courses")
def list_courses(
    level: Level | None = None,
    min_rating: Annotated[float, Query(ge=0, le=5)] = 0,
    sort: Literal["id", "rating", "hours"] = "id",
) -> CourseList:
    courses = [c for c in upstream.fetch_courses(level) if c.rating >= min_rating]
    courses.sort(key=lambda c: getattr(c, sort), reverse=sort != "id")
    return CourseList(count=len(courses), items=courses)


@app.get("/courses/{course_id}")
def get_course(course_id: Annotated[int, Path(ge=1)]) -> Course:
    course = upstream.fetch_course(course_id)
    if course is None:
        raise HTTPException(404, f"Course {course_id} not found")
    return course


@app.get("/summary")
def summary() -> list[LevelSummary]:
    by_level: dict[str, list[Course]] = {}
    for course in upstream.fetch_courses():
        by_level.setdefault(course.level, []).append(course)
    return [
        LevelSummary(
            level=level,
            courses=len(group),
            total_hours=sum(c.hours for c in group),
            average_rating=round(sum(c.rating for c in group) / len(group), 2),
        )
        for level, group in by_level.items()
    ]


def advice_for(temp_c: float, condition: str) -> str:
    if condition == "Rain":
        return "Carry an umbrella."
    if temp_c >= 30:
        return "Hot day: drink water and avoid the midday sun."
    return "Good weather to go out."


@app.get("/weather/{city}")
def weather(city: Annotated[str, Path(min_length=2, max_length=40)]) -> Weather:
    data = upstream.fetch_weather(city)
    if data is None:
        raise HTTPException(404, f"No weather data for {city}")
    return Weather(
        city=data.city,
        temperature_c=data.temp_c,
        humidity_pct=data.humidity,
        condition=data.condition,
        advice=advice_for(data.temp_c, data.condition),
    )

What to notice:

• @app.exception_handler(upstream.UpstreamError) catches the client's error wherever it is raised and turns it into a JSON response with the right status. Without it, an UpstreamError would become a bare 500 "Internal Server Error". One handler covers every route, now and later.
• Routes call upstream.fetch_courses(...) through the module, not with from app.upstream import fetch_courses. That small habit lets the tests on the next page swap the function for a fake one.
• Parameters are typed and limited with Annotated, Query and Path (see Pydantic Models, Validation and Error Responses): min_rating must be 0–5, course_id at least 1, city 2–40 characters, sort one of three words. Bad input never reaches the upstream.
• Return types are the response models. FastAPI validates and serialises the result and documents it in /docs; no response_model= needed.
• The reshaping is ours: temp_c becomes temperature_c, humidity becomes humidity_pct, and advice is new. If the upstream renames a field, only UpstreamWeather and these few lines change; our clients see nothing.
• Plain def routes, because requests blocks. FastAPI runs def routes in a thread pool, so one slow upstream call does not stop the server (the sync vs async section of What a Backend Does: Requests, Routes and Responses).

git add app/main.py
git commit -q -m "Add courses, summary and weather endpoints"

Step 5: Run it and explore

Start the server in the background and wait until it answers (in your own terminal you can simply run fastapi dev in a second window and stop it with Ctrl+C):

fastapi run --port 8000 > server.log 2>&1 &
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

The advanced courses, best rated first:

curl -s "http://localhost:8000/courses?level=advanced&sort=rating" | python -m json.tool

{
    "count": 3,
    "items": [
        {
            "id": 8,
            "title": "RAG Systems",
            "level": "advanced",
            "hours": 10,
            "rating": 4.9
        },
        {
            "id": 9,
            "title": "AI Agents",
            "level": "advanced",
            "hours": 12,
            "rating": 4.8
        },
        {
            "id": 10,
            "title": "LLMOps",
            "level": "advanced",
            "hours": 8,
            "rating": 4.3
        }
    ]
}

Courses rated 4.7 or more, the per-level summary, and the weather in two cities:

curl -s "http://localhost:8000/courses?min_rating=4.7" | python -c "import json, sys; d = json.load(sys.stdin); print(d['count'], [c['title'] for c in d['items']])"
curl -s http://localhost:8000/summary; echo
curl -s http://localhost:8000/weather/Sylhet; echo
curl -s http://localhost:8000/weather/rajshahi; echo

5 ['How AI Works', 'Python for AI', 'LLM APIs', 'RAG Systems', 'AI Agents']
[{"level":"beginner","courses":5,"total_hours":42,"average_rating":4.64},{"level":"intermediate","courses":2,"total_hours":11,"average_rating":4.55},{"level":"advanced","courses":3,"total_hours":30,"average_rating":4.67}]
{"city":"Sylhet","temperature_c":27.4,"humidity_pct":85,"condition":"Rain","advice":"Carry an umbrella."}
{"city":"Rajshahi","temperature_c":33.6,"humidity_pct":58,"condition":"Sunny","advice":"Hot day: drink water and avoid the midday sun."}

The summary is computed by our API, not the upstream: 5 beginner courses with 3 + 12 + 8 + 10 + 9 = 42 hours. That is the "reshape" part of the brief: the client asks one question and gets one small answer, instead of downloading every course and doing the maths itself. For an AI assistant that matters twice: fewer tokens in its context, and fewer chances to add numbers up wrong.

Open http://localhost:8000/docs in a browser: every endpoint is there with its parameters, the allowed values of level and sort, and the response shapes, all generated from your type hints. The same information as JSON:

curl -s http://localhost:8000/openapi.json | python -c "import json, sys; print(list(json.load(sys.stdin)['paths']))"

['/health', '/courses', '/courses/{course_id}', '/summary', '/weather/{city}']

Step 6: Break it on purpose

An error path you have never seen work is a guess. Check each one. First, client mistakes and missing data:

curl -s -w '  [%{http_code}]\n' http://localhost:8000/courses/99
curl -s -w '  [%{http_code}]\n' http://localhost:8000/weather/Comilla
curl -s -w '  [%{http_code}]\n' http://localhost:8000/courses/0
curl -s -w '  [%{http_code}]\n' "http://localhost:8000/courses?level=expert"

{"detail":"Course 99 not found"}  [404]
{"detail":"No weather data for Comilla"}  [404]
{"detail":[{"type":"greater_than_equal","loc":["path","course_id"],"msg":"Input should be greater than or equal to 1","input":"0","ctx":{"ge":1}}]}  [422]
{"detail":[{"type":"literal_error","loc":["query","level"],"msg":"Input should be 'beginner', 'intermediate' or 'advanced'","input":"expert","ctx":{"expected":"'beginner', 'intermediate' or 'advanced'"}}]}  [422]

-w ' [%{http_code}]\n' makes curl print the status code after the body. 404s carry our own short messages; the 422s come from FastAPI's validation and point exactly at the bad parameter (loc) and the allowed values. The Practice API was never called for those.

Now an upstream that is down. Start a second copy of the app on port 8001 whose PRACTICE_API_URL points at a port where nothing listens:

PRACTICE_API_URL=http://127.0.0.1:9 fastapi run --port 8001 > broken.log 2>&1 &
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s -w '  [%{http_code}]\n' http://localhost:8001/courses
curl -s -w '  [%{http_code}]\n' http://localhost:8001/weather/dhaka

{"status":"ok"}
{"detail":"Practice API is unreachable"}  [502]
{"detail":"Practice API is unreachable"}  [502]

/health still says ok, because it does not depend on the upstream. That is correct for a liveness check ("is the process alive?"); the advanced tutorial adds a separate readiness check that also tests dependencies. The data endpoints answer 502 with one clear sentence, and our server stays up. Look at the log of the broken copy:

grep -oE '"GET .*' broken.log

"GET /health HTTP/1.1" 200 OK
"GET /courses HTTP/1.1" 502 Bad Gateway
"GET /weather/dhaka HTTP/1.1" 502 Bad Gateway

No traceback, just three requests and their status codes. A 502 tells whoever reads the log, or a monitoring alert, "our code is fine, the thing behind it is not".

Where you are in Git

git status --short
git log --oneline --graph --decorate --all

* 3a8f75a (HEAD -> feature/data-explorer) Add courses, summary and weather endpoints
* 089f152 Add Practice API client with timeouts, pagination and validation
* 6bba991 Add response and upstream schemas
* 04b26f9 (main) Set up the Data Explorer project with uv

git status --short printed nothing: the working tree is clean, and server.log and broken.log stay out of it because *.log is ignored. main still has only the skeleton; the feature branch carries three commits, each one small enough to review on its own and each with a message that says what it adds. That is exactly what the pull request on the next page will show a reviewer. Your hashes will differ from the ones above: a commit hash includes the time you committed.

Where this shows up

• LLM backends: replace the Practice API with an LLM provider and this is the skeleton of every chat backend: one client module, a timeout on every call, provider errors turned into 502/504, and responses validated before they reach the user.
• Tools for AI agents: agents (and MCP servers) call small, typed endpoints like /summary and /weather/{city}. Short, predictable answers with clear errors are what make an agent reliable.
• "Backend for frontend" services: a mobile app or website rarely calls five upstream APIs itself; one backend calls them, combines and reshapes the data and hides their keys.
• Data pipelines: the same client code (session, timeout, pagination, validation) is how ETL jobs pull data from SaaS APIs into a database or a training set.

Common mistakes

• No timeout on upstream calls. One hung upstream ties up a worker per request until your whole server stops answering. Fix: timeout= on every call, from a setting.
• Passing upstream errors straight through. Returning the upstream's 500 (or letting requests exceptions become your 500) tells the client nothing and hides whose fault it is. Fix: translate into 502/504 with your own message, and 404 where "not found" is a normal answer.
• Trusting upstream data. A renamed or missing field becomes a KeyError deep in a route, or wrong numbers in the answer. Fix: validate with a Pydantic model at the boundary and fail clearly.
• Reading only the first page. The Practice API returns 5 courses by default; code that ignores next_page silently reports half the data. Fix: loop until next_page is null, with a page limit.
• Working directly on main in one giant commit. Nobody can review it and nothing can be undone in pieces. Fix: a feature branch and one commit per step.

Try it yourself

The answers add routes to the running app and call them with FastAPI's TestClient (from Testing Your API with pytest), so you do not have to restart the server. In your project, put the same functions in app/main.py.

1. Easy: Add GET /levels that returns the sorted list of course levels that exist, computed from the upstream courses.
2. Medium: Add GET /weather (no city) that returns the weather for Dhaka, Chattogram, Sylhet, Khulna and Rajshahi, hottest first. Reuse the existing weather() function.
3. Hard: The upstream sometimes answers 503 for a moment. Mount an HTTPAdapter with a urllib3 Retry on upstream.session that retries GET requests up to 3 times on 429, 502, 503 and 504 with a short backoff, then prove it with the Practice API's /v1/flaky?key=p20&fail_times=2 through upstream.get_json. Why is retrying only GET a sensible default?

Answers

# 1. GET /levels
from fastapi.testclient import TestClient
from app.main import app, weather

@app.get("/levels")
def levels() -> list[str]:
    return sorted({course.level for course in upstream.fetch_courses()})

client = TestClient(app)
print(client.get("/levels").json())

['advanced', 'beginner', 'intermediate']

# 2. GET /weather for all five cities, hottest first
from app.schemas import Weather

CITIES = ["Dhaka", "Chattogram", "Sylhet", "Khulna", "Rajshahi"]

@app.get("/weather")
def all_weather() -> list[Weather]:
    reports = [weather(city) for city in CITIES]
    return sorted(reports, key=lambda w: w.temperature_c, reverse=True)

for w in client.get("/weather").json():
    print(f"{w['city']:11} {w['temperature_c']:5} {w['advice']}")

Rajshahi     33.6 Hot day: drink water and avoid the midday sun.
Khulna       32.1 Hot day: drink water and avoid the midday sun.
Dhaka        31.5 Hot day: drink water and avoid the midday sun.
Chattogram   29.8 Carry an umbrella.
Sylhet       27.4 Carry an umbrella.

# 3. Retries for temporary upstream failures
from requests.adapters import HTTPAdapter
from urllib3.util import Retry

retry = Retry(total=3, backoff_factor=0.2, status_forcelist=[429, 502, 503, 504],
              allowed_methods=["GET"])
upstream.session.mount("http://", HTTPAdapter(max_retries=retry))
upstream.session.mount("https://", HTTPAdapter(max_retries=retry))

print(upstream.get_json("/v1/flaky", {"key": "p20", "fail_times": 2}))

{'ok': True, 'attempt': 3}

The first two calls got 503, the adapter waited and retried, and the third attempt succeeded; your code only saw the success. GET is safe to repeat because it does not change anything. Retrying a POST that creates something (a review, a payment, an LLM call you pay for) could do it twice, so those need idempotency keys first, which the advanced tutorial builds. Keep the total time in mind too: retries multiply the worst-case wait, so the timeout and the retry count must fit together within what your own client is willing to wait.

Summary

• Start a project by writing the contract: endpoints, response shapes, and what happens for each kind of error.
• Keep every upstream call in one module with a shared session, a timeout on every request, pagination with a limit and Pydantic validation of what comes back.
• Translate upstream trouble into one exception type and, with one exception handler, into 502 or 504 with a short message; treat "not found" as a normal answer and return 404.
• Type and limit every parameter with Annotated, Query, Path and Literal, so bad input is rejected with 422 before it costs an upstream call.
• Build on a feature branch in small commits, and test the error paths by breaking things on purpose.

Next: Project, Part 2: Test, Document and Publish It adds automated tests with the upstream replaced by fakes, a README, a review, a refactor and the merge into main through a pull-request workflow.