Build a Retry-Safe Payment API

Build two local APIs with idempotent payments, risk checks, and safe failures.

Introduction

30 Second Summary

A slow connection can make a customer press Pay twice. A safe payment system must recognize the repeated request before it creates a duplicate payment.

In this project, you will build a local Payments API that asks a separate Risk API to approve or decline each request. You'll protect the payment flow with idempotency across a REST service boundary.

What You'll Build

You'll demo a payment that crosses two local APIs while preserving one result across repeated requests.

By the end of this project, you'll have:

  • A documented payment contract that accepts a typed request. It returns a payment ID with an APPROVED or DECLINED status through an automatically generated OpenAPI contract.
  • A retry-safe payment flow that returns the same payment ID when a client repeats a request. Reusing the key with changed payment data produces a structured 409 conflict.
  • A traceable two-service call that carries one correlation ID through both service logs. Stopping the Risk API produces a controlled 503 response.
  • An optional Secret Mission to push your skills further.

Are there any prerequisites?

You do not need previous API experience. You need a Linux computer with internet access because the walkthrough includes every local setup check.

Before We Start

Before any setup, commit to a training-only fintech simulation where a Payments API asks a Risk API whether an amount should be approved or declined. Duplicate prevention matters because a retried client request must not create a second payment.

Set Up the Linux API Workspace

Your payment simulation will soon run as two local APIs. Machine-specific package drift can hide the API concepts behind setup failures.

A reproducible workspace removes that distraction. You'll prepare the exact editor toolchain and dependency set used throughout the project.

In this step, get ready to:
  • Verify Visual Studio Code on Linux.
  • Prepare uv version 0.13.6 with Python 3.12.
  • Create the local workspace with its locked project environment.
Verify Visual Studio Code

Visual Studio Code gives you one place to edit the service files. Its command-line tool also lets later terminal commands work with the same workspace.

  • Open a terminal from your Linux application menu.
  • Check whether the Visual Studio Code command is available by running:
code --version

What does this check prove?

The code --version command prints the installed Visual Studio Code version. A version in the terminal confirms that the code command is available from your shell.

✔️ I see a version

Good, your editor's command-line tool is ready. Continue to the uv setup below.

ⓧ Command not found

Visual Studio Code provides official Linux packages for the common distribution families. Choose the package that matches your system.

  • Download the Linux package for your distribution from the official Visual Studio Code page.
  • Open your Linux file manager's downloads folder.
  • Use the file manager's context menu to open a terminal in that folder.

Administrator access can feel sensitive. Linux hides your password while you type, so a blank-looking password prompt is expected.

Debian or Ubuntu

  • Replace <file> with the downloaded .deb filename.
  • Install the downloaded package by running:
sudo apt install ./<file>.deb
What does this command do?

The package manager installs the downloaded Microsoft package. The leading ./ points to the file in your current folder.

Fedora or RHEL

  • Replace <file> with the downloaded .rpm filename.
  • Install the downloaded package by running:
sudo dnf install ./<file>.rpm
What does this command do?

The package manager installs the downloaded Microsoft package. The leading ./ points to the file in your current folder.

  • Close the current terminal after the installation finishes.
  • Open a new terminal from your Linux application menu.
  • Verify the installation by running:
code --version

What should I see?

The terminal should print a Visual Studio Code version. This confirms that Linux can now find the code command.

Still unable to run the code command?

Close every open terminal after the package installation. A newly opened terminal reloads the command search path.

Confirm that you downloaded the package for your distribution family. A .deb package belongs to Debian or Ubuntu systems.

An .rpm package belongs to Fedora or RHEL systems.

Ask for help with this installation: Help me diagnose why the code command is unavailable after installing Visual Studio Code on Linux. Share the same terminal output in the NextWork community if you want a second set of eyes.

Install uv and Python 3.12

The uv tool installs the project runtime. It also keeps every dependency pinned to a known version.

  • Check whether uv is already installed by running:
uv --version

What does this check prove?

The uv --version command reports the installed release. This project uses exactly 0.13.6 for reproducible environment behavior.

✔️ I see the required version

Your pinned uv release is ready. Move to the Python check after this tab group.

ⓧ I see another version

The project pins uv to 0.13.6. The versioned installer sets the required release.

  • Install the pinned uv release by running:
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | sh

What does this installer do?

The command downloads the official installer for uv 0.13.6. The shell then installs that pinned release for your user account.

  • Close the current terminal after the installer finishes.
  • Open a new terminal from your Linux application menu.
  • Verify the pinned release by running:
uv --version

What should I see?

The version output should include 0.13.6. Your shell is now using the project's pinned uv release.

Still seeing a different uv version?

Make sure you opened a new terminal after running the installer. The new terminal reloads the path that points to the installed release.

Ask for help with the version mismatch: Help me find why uv --version does not show 0.13.6 after running the pinned Linux installer. Share the output in the NextWork community if the mismatch continues.

ⓧ Command not found

Uv is missing from your shell. The versioned Linux installer adds the required release.

  • Install the pinned uv release by running:
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | sh

What does this installer do?

The command downloads the official installer for uv 0.13.6. The shell then installs that pinned release for your user account.

  • Close the current terminal after the installer finishes.
  • Open a new terminal from your Linux application menu.
  • Verify the installation by running:
uv --version

What should I see?

The version output should include 0.13.6. This confirms that the new terminal can find uv.

Uv still unavailable?

Confirm that the installer completed without a network error. Open a new terminal before checking the command again.

Ask for help with the missing command: Help me troubleshoot why uv is unavailable after running the official Linux installer. Share the installer output in the NextWork community if you need more support.

  • List the Python releases visible to uv by running:
uv python list

What does this list show?

The list distinguishes Python releases that uv can manage. Use it to check whether the project's 3.12 line is already installed.

✔️ The required Python is installed

Your Python 3.12 runtime is ready for the project.

ⓧ Another Python version is installed

The project requires the Python 3.12 line. Uv can install it without replacing the other release.

  • Install Python 3.12 through uv by running:
uv python install 3.12

What does this command do?

Uv downloads a managed Python 3.12 runtime. The project can select this runtime through its version file.

ⓧ No Python release is installed

Uv can supply the required runtime directly. This keeps the project's Python release separate from the Linux system Python.

  • Install Python 3.12 through uv by running:
uv python install 3.12

What does this command do?

Uv downloads a managed Python 3.12 runtime. The project can select this runtime through its version file.

Python 3.12 missing after installation?

Run the installation again after confirming that your internet connection is active. The download must finish before the runtime appears in the Python list.

Ask for help with the runtime installation: Help me troubleshoot why uv python list does not show an installed Python 3.12 release. Share the command output in the NextWork community if the runtime remains missing.

Create and synchronize the workspace

The workspace files declare the exact runtime range and package versions. Uv reads those declarations to build the same environment on every machine.

  • Open your Linux application launcher.
  • Search for Visual Studio Code.
  • Select Visual Studio Code from the results.
  • Click File in the top menu.
  • Select Open Folder.
  • Select your Desktop in the folder dialog.
  • Create a folder named fintech-api-lab.
  • Select the fintech-api-lab folder.
  • Click Open.

You should see fintech-api-lab at the top of the Explorer sidebar. This confirms that VS Code has opened the correct workspace.

Seeing a workspace trust prompt?

VS Code may ask whether you trust this folder. You created this empty local workspace, so confirm that you trust its contents.

  • Use the new-file control in the Explorer sidebar to create .python-version inside fintech-api-lab.
  • Set the project's Python line by copying this content into the file:
3.12

What does this file do?

The .python-version file tells uv which Python line belongs to this workspace. It selects the managed 3.12 runtime you verified earlier.

  • Save .python-version.

You should now see .python-version in the Explorer sidebar.

  • Use the new-file control in the Explorer sidebar to create pyproject.toml inside fintech-api-lab.
  • Define the project and its pinned dependencies by copying this content into the file:
[project]
name = "fintech-api-lab"
version = "0.1.0"
description = "A retry-safe payment API learning project"
requires-python = ">=3.12,<3.13"
dependencies = [
    "fastapi==0.162.0",
    "httpx==0.28.1",
    "pydantic==2.14.2",
    "uvicorn==0.54.0",
]

What does this configuration do?

  • The Python range keeps the workspace on the 3.12 release line.
  • FastAPI 0.162.0 provides the API framework.
  • HTTPX 0.28.1 provides the client for service-to-service calls.
  • Pydantic 2.14.2 validates typed request data.
  • Uvicorn 0.54.0 runs each local API service.
  • Save pyproject.toml.

You should now see both project files in the Explorer sidebar. Their pinned values give uv a complete environment specification.

✔️ Awesome, I've got everything!

Great. Double check that both files are saved inside fintech-api-lab.

ⓧ I'd like to double check the full code

3.12

[project]
name = "fintech-api-lab"
version = "0.1.0"
description = "A retry-safe payment API learning project"
requires-python = ">=3.12,<3.13"
dependencies = [
    "fastapi==0.162.0",
    "httpx==0.28.1",
    "pydantic==2.14.2",
    "uvicorn==0.54.0",
]
  • Click Terminal in the VS Code top menu.
  • Select New Terminal.

Before you run the checks, which tool versions do you expect the terminal to confirm?

  • Verify the complete workspace by running these commands:
uv sync
uv python list
uv --version
code --version

What do these checks prove?

  • The uv sync command creates or updates the locked project environment from pyproject.toml.
  • The uv python list output shows an installed Python 3.12 release.
  • The uv version check shows 0.13.6.
  • The Visual Studio Code check prints its installed version.
  • The synchronization uses FastAPI 0.162.0, HTTPX 0.28.1, Pydantic 2.14.2, and Uvicorn 0.54.0.

That's the setup complete: your locked workspace now has the exact runtime and dependencies needed for both API services.

Workspace synchronization failed?

Confirm that the VS Code terminal starts inside the fintech-api-lab folder. The Explorer sidebar should show .python-version beside pyproject.toml.

Check both files against the full-code tab. A changed quote or version pin can prevent dependency resolution.

Ask for help with the failed synchronization: Help me diagnose why uv sync fails in my fintech-api-lab workspace using my terminal output and project files. Share the error output in the NextWork community if you need more support.

Your Linux API workspace is ready. Next, you'll build the first Payments API endpoint and expose the duplicate-payment problem.

Build a Payment Endpoint

Your pinned workspace is ready. It now has every package needed to run a local FastAPI Payments API.

Clients can retry a payment request when a response is delayed. This step lets you observe how a basic create route handles that retry before you add protection.

In this step, get ready to:
  • Define typed models for USD payment requests and responses.
  • Expose health and payment routes through a local API.
  • Test how the API handles the same payment request twice.
Define the payment contract

A typed contract specifies the data that your API accepts. Pydantic checks incoming request bodies against that contract before your route runs.

  • Create a file named payments_service.py inside the fintech-api-lab folder by using the VS Code Explorer sidebar.
  • Add the FastAPI application and request model by pasting this code into payments_service.py:
from typing import Literal
from uuid import uuid4

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Payments API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: Literal["USD"]

What Does This Code Do?

  • The app object creates the Payments API with version 1.0.0.
  • The PaymentRequest model defines the typed request body.
  • The ConfigDict setting rejects unexpected fields.
  • The Literal["USD"] annotation limits this learning API to USD payments.
  • Save payments_service.py.
  • Start the Payments API inside the synchronized environment by running this command in the VS Code terminal:
uv run uvicorn payments_service:app --port 8000 --reload

What Does This Command Do?

  • The uv run command uses the locked project environment.
  • Uvicorn loads the app object from payments_service.py.
  • The server listens at http://127.0.0.1:8000.
  • The --reload option restarts the API after each saved code change.

Your first API process is live. You should see terminal output confirming that the server is listening on port 8000.

Server Failed to Start?

  • Check that the terminal is inside the fintech-api-lab folder.
  • Check that the file is named payments_service.py.
  • Confirm that the app line matches the code above.

Help me fix the Payments API startup error.

The response needs its own model so every successful payment has the same documented shape. Each response carries a new UUID plus an approval status.

  • Add the response model below PaymentRequest by pasting this code:
class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED"]

What Does This Model Define?

  • The payment_id field identifies one created payment.
  • The status field accepts the training value APPROVED.
  • FastAPI uses this model to validate successful route responses.
  • Save payments_service.py.
  • Return to the server terminal to confirm that the automatic reload completes.

You should see the application restart successfully. That confirms the response model imports without a syntax error.

Reload Showing an Error?

  • Check that PaymentResponse starts at the left edge of the file.
  • Check that both field definitions are indented inside the class.
  • Check that the status annotation uses Literal["APPROVED"].

Help me fix my PaymentResponse model.

Expose the API routes

Routes connect HTTP requests to Python functions. FastAPI also turns those declarations into an OpenAPI contract that clients can inspect.

  • Add the health route below PaymentResponse by pasting this code:
@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}

What Does This Route Do?

  • The @app.get("/health") decorator registers a GET route.
  • The health() function returns a small service-status response.
  • The return annotation documents a dictionary with string keys and string values.
  • Save payments_service.py.
  • Open your web browser from the Linux application launcher.
  • Visit http://127.0.0.1:8000/health to check the new route.

You should see a JSON response containing {"status":"ok"}. This proves the API can receive a request and return a typed response.

Health Route Missing?

  • Confirm that the server terminal completed its automatic reload.
  • Check that the browser URL ends with /health.
  • Confirm that the route decorator starts with @app.get.

Help me debug the health route.

  • Add the payment creation route below health() by pasting this code:
@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(payment: PaymentRequest) -> PaymentResponse:
    return PaymentResponse(payment_id=str(uuid4()), status="APPROVED")

What Does This Route Do?

  • The @app.post("/payments") decorator registers the payment creation route.
  • The payment parameter makes FastAPI validate the request against PaymentRequest.
  • The response_model setting documents and validates the response as PaymentResponse.
  • The uuid4() call creates a fresh payment identifier every time the route runs.
  • Save payments_service.py.
  • Return to the browser.
  • Visit http://127.0.0.1:8000/openapi.json to inspect the generated API contract.

You should find /health and /payments inside the contract's paths. The payment route also shows a successful status code of 201.

Payment Route Missing?

  • Confirm that the payment route appears below the complete health() function.
  • Check that the decorator uses @app.post.
  • Confirm that the server terminal completed its automatic reload.

Help me find why the payment route is missing.

✔️ Awesome, I've got everything!

Your payments_service.py file now contains both typed models and both API routes.

ⓧ I'd like to double check the full code

from typing import Literal
from uuid import uuid4

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Payments API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED"]


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


@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(payment: PaymentRequest) -> PaymentResponse:
    return PaymentResponse(payment_id=str(uuid4()), status="APPROVED")

Reference Check

This is the complete payments_service.py file for this step. Its order matches the running API.

Test a repeated payment request

HTTPX acts as an HTTP client for your demo. It sends the same typed payment body twice so you can inspect how the route behaves during a retry.

  • Create a file named demo.py inside the fintech-api-lab folder by using the VS Code Explorer sidebar.
  • Add the repeated payment requests by pasting this code into demo.py:
from pprint import pprint

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}

first = httpx.post(PAYMENTS_URL, json=payload, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=payload, timeout=2.0)
print("first", first.status_code, first.json())
print("retry", retry.status_code, retry.json())

What Does This Client Do?

  • The payload dictionary matches the PaymentRequest contract.
  • The first httpx.post() call submits the payment.
  • The second httpx.post() call repeats the same request.
  • Each print statement exposes the response status and body.
  • Save demo.py.
  • Open a second integrated terminal from the VS Code top menu.

Before you run the client, do you think both requests will identify the same payment?

  • Execute the demo inside the synchronized environment by running this command:
uv run python demo.py

What Does This Command Do?

The command runs demo.py with the dependencies from the locked project environment. The client calls the API process that remains active in the first terminal.

You should see lines beginning with first 201 and retry 201. Each response contains a different payment_id.

You found the failure on purpose. The endpoint treats the retry as a new payment because every call generates another UUID.

Demo Could Not Reach the API?

  • Return to the first terminal to confirm that Uvicorn is still running.
  • Check that PAYMENTS_URL uses port 8000.
  • Confirm that the payment route is saved in payments_service.py.

Help me connect the demo to the Payments API.

The client can also read the generated contract. Sorting its paths gives you a compact check that both routes are published.

  • Add the OpenAPI check below the second print statement by pasting this code:
schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
pprint(sorted(schema_response.json()["paths"]))

What Does This Check Do?

  • The httpx.get() call downloads the generated OpenAPI document.
  • The raise_for_status() call stops the client if the schema request fails.
  • The final line prints the published route paths in sorted order.
  • Save demo.py.

Before you run the final check, which two route paths do you expect the schema to list?

  • Run the completed demo inside the synchronized environment with this command:
uv run python demo.py

What Does This Final Run Prove?

The client exercises the create route twice. It also reads the server-generated API contract during the same run.

You should again see two successful responses with different payment IDs. The final line should show ['/health', '/payments'].

Missing a Path or Response?

  • Check that both route decorators are saved in payments_service.py.
  • Check that the first terminal shows a completed reload after your latest save.
  • Confirm that demo.py uses the two local URLs shown above.

Help me verify the duplicate responses and OpenAPI paths.

✔️ Awesome, I've got everything!

Your demo.py file now reproduces the retry problem and checks the published API paths.

ⓧ I'd like to double check the full code

from pprint import pprint

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}

first = httpx.post(PAYMENTS_URL, json=payload, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=payload, timeout=2.0)
print("first", first.status_code, first.json())
print("retry", retry.status_code, retry.json())

schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
pprint(sorted(schema_response.json()["paths"]))

Reference Check

This is the complete demo.py file for this step. It preserves the repeated request before checking the OpenAPI paths.

Your Payments API now exposes a typed create route and a visible retry failure. Next, you will use those repeated IDs to make payment retries safe.

Make Payment Retries Idempotent

Your last run exposed the duplicate-payment risk. A client retry produced a second payment ID for the same request.

An idempotency key gives each intended operation a stable identity. FastAPI can require that key before the payment logic runs.

In this step, get ready to:
  • Require an Idempotency-Key header for every payment request.
  • Store each payment so matching retries return its original payment ID.
  • Reject changed payment data that reuses an existing key.
Require an idempotency key

The server needs two pieces of memory for each operation. It stores the original response beside a fingerprint made from the account, amount, and currency.

  • In payments_service.py, select everything from from typing import Literal through the closing line of PaymentResponse.
  • Replace that selection with this expanded application setup:
from typing import Annotated, Literal
from uuid import uuid4

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Payments API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED"]


payments_by_idempotency_key: dict[str, tuple[tuple[str, int, str], PaymentResponse]] = {}

What does this code add?

  • The Annotated and Header imports let the endpoint declare a required request header.
  • The HTTPException import lets the endpoint return a structured conflict response.
  • The payments_by_idempotency_key dictionary pairs each key with one fingerprint and one PaymentResponse.
  • Save payments_service.py.

Your running server reloads the file. Its terminal should complete the reload without showing a Python traceback.

Seeing a reload error?

  • Check that Annotated appears beside Literal in the first import.
  • Check that Header appears beside FastAPI in the FastAPI import.
  • Place payments_by_idempotency_key below the complete PaymentResponse class.

Help me fix the imports or store declaration.

The endpoint can now use that store to recognize a repeated operation. A matching fingerprint returns the saved response without generating another UUID.

  • Find the @app.post("/payments", response_model=PaymentResponse, status_code=201) decorator near the bottom of payments_service.py.
  • Replace that decorator and the existing create_payment() function with this code:
@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(
    payment: PaymentRequest,
    idempotency_key: Annotated[str, Header(min_length=1)],
) -> PaymentResponse:
    fingerprint = (payment.account_id, payment.amount_minor, payment.currency)
    previous = payments_by_idempotency_key.get(idempotency_key)
    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(status_code=409, detail={"code": "IDEMPOTENCY_KEY_REUSED"})
        return previous_response
    payment_response = PaymentResponse(payment_id=str(uuid4()), status="APPROVED")
    payments_by_idempotency_key[idempotency_key] = (fingerprint, payment_response)
    return payment_response

How does the retry check work?

  • The idempotency_key parameter requires a non-empty request header.
  • The fingerprint captures the payment data that defines the operation.
  • A matching stored fingerprint returns previous_response with its original payment ID.
  • A changed fingerprint raises a 409 conflict with the code IDEMPOTENCY_KEY_REUSED.
  • Save payments_service.py.

Before you run the existing demo, predict whether its requests satisfy the new contract. Pause for a moment before continuing.

  • Switch back to the terminal you used for the demo.
  • Test the existing client against the required header by running:
uv run python demo.py

What did this run prove?

Both payment responses now report that the required Idempotency-Key header is missing. FastAPI rejects each request before create_payment() can create a payment.

This is the intended intermediate result. The server contract is active, so the client must send the operation key next.

Did the demo fail differently?

  • Confirm the Payments API terminal still shows the server running on port 8000.
  • Confirm the demo terminal is inside the fintech-api-lab folder.
  • Review the Payments API terminal for a syntax error if the server stopped during reload.

Help me diagnose the unexpected demo result.

✔️ Awesome, I've got everything!

Your Payments API now requires an idempotency key. Double check that payments_service.py is saved.

ⓧ I'd like to double check the full code

from typing import Annotated, Literal
from uuid import uuid4

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Payments API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED"]


payments_by_idempotency_key: dict[str, tuple[tuple[str, int, str], PaymentResponse]] = {}


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


@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(
    payment: PaymentRequest,
    idempotency_key: Annotated[str, Header(min_length=1)],
) -> PaymentResponse:
    fingerprint = (payment.account_id, payment.amount_minor, payment.currency)
    previous = payments_by_idempotency_key.get(idempotency_key)
    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(status_code=409, detail={"code": "IDEMPOTENCY_KEY_REUSED"})
        return previous_response
    payment_response = PaymentResponse(payment_id=str(uuid4()), status="APPROVED")
    payments_by_idempotency_key[idempotency_key] = (fingerprint, payment_response)
    return payment_response

What should match?

Use this full-file reference to compare the imports, store declaration, and complete endpoint. Each name must match because the demo depends on this exact request contract.

Send the key from the demo

The client needs one fresh key for each demo run. Reusing that key across the first two calls simulates a retry of one intended payment.

  • In demo.py, select everything from from pprint import pprint through the existing print("retry", retry.status_code, retry.json()) line.
  • Replace that selection with this retry and conflict scenario:
from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
run_id = str(uuid4())
headers = {"Idempotency-Key": f"payment-{run_id}"}
payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}

first = httpx.post(PAYMENTS_URL, json=payload, headers=headers, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=payload, headers=headers, timeout=2.0)
conflict = httpx.post(PAYMENTS_URL, json={**payload, "amount_minor": 13_000}, headers=headers, timeout=2.0)
pprint((first.status_code, first.json()))
pprint((retry.status_code, retry.json()))
pprint((conflict.status_code, conflict.json()))

What does the demo test?

  • The run_id creates a fresh key whenever the script starts.
  • The first two calls send identical payment data with the same key.
  • The third call changes amount_minor while preserving the key.
  • The three printed tuples expose the successful creation, safe retry, and conflict response.
  • Save demo.py.

✔️ Awesome, I've got everything!

Your demo now covers the first request, matching retry, and changed-payload conflict. Double check that demo.py is saved.

ⓧ I'd like to double check the full code

from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
run_id = str(uuid4())
headers = {"Idempotency-Key": f"payment-{run_id}"}
payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}

first = httpx.post(PAYMENTS_URL, json=payload, headers=headers, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=payload, headers=headers, timeout=2.0)
conflict = httpx.post(PAYMENTS_URL, json={**payload, "amount_minor": 13_000}, headers=headers, timeout=2.0)
pprint((first.status_code, first.json()))
pprint((retry.status_code, retry.json()))
pprint((conflict.status_code, conflict.json()))

schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
pprint(sorted(schema_response.json()["paths"]))

What should match?

Use this reference to compare the generated key, shared headers, three requests, and existing schema check. Together they exercise the complete contract for this step.

Verify safe retries and key reuse

The completed demo now tests the same operation twice. It also tests one unsafe attempt to change that operation.

Before you run it, which two calls do you expect to share one payment ID? Pause before checking the output.

  • Run the completed demo from the fintech-api-lab folder with this command:
uv run python demo.py

What should you see?

  • The first two response tuples contain the same payment_id.
  • The third response starts with 409.
  • The third response body contains IDEMPOTENCY_KEY_REUSED.
  • The OpenAPI path list still contains /payments.

Seeing different payment IDs?

  • Confirm both successful requests pass the same headers dictionary.
  • Confirm payments_by_idempotency_key is declared outside create_payment().
  • Confirm the matching-key branch returns previous_response before creating a new UUID.

Help me find why my retry still creates a duplicate.

You closed the duplicate-payment gap. Matching retries now recover one stored result, while changed payment data receives a clear conflict response.

Next, you will move the approval decision into a separate Risk API. That service boundary introduces a new failure mode for the payment workflow.

Split Risk Decisions into a Second Service

Your Payments API already uses idempotency to prevent retries from creating duplicate payment IDs. It still decides every payment outcome inside one process.

A separate Risk API creates a service boundary around the approval policy. That boundary also makes network failure part of the payment flow.

In this step, get ready to:
  • Build a typed Risk API with a visible training rule.
  • Route payment decisions through the new service.
  • Expose the unhandled failure caused by an unavailable Risk API.
Build the Risk API

The Risk API uses FastAPI to publish a separate contract on port 8001. Its Pydantic models reject unexpected request fields before the training rule runs.

  • Create risk_service.py inside the fintech-api-lab folder by using the New File button in the Visual Studio Code Explorer.
  • Add the Risk API application and typed contracts by pasting this code into risk_service.py:
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Risk API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: str


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str

What does this code do?

  • The application title identifies this process as the Risk API in its generated documentation.
  • The RiskCheckRequest model defines the payment data accepted by the service.
  • The RiskCheckResponse model gives every decision a Boolean result plus a reason.
  • Add the health route and risk-check route below RiskCheckResponse by pasting this code:
@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/risk-checks", response_model=RiskCheckResponse)
def check_risk(risk: RiskCheckRequest) -> RiskCheckResponse:
    if risk.amount_minor <= 0:
        return RiskCheckResponse(approved=False, reason="amount must be positive")
    if risk.amount_minor > 50_000:
        return RiskCheckResponse(approved=False, reason="training risk limit exceeded")
    return RiskCheckResponse(approved=True, reason="within training risk limit")

How does the training rule work?

The check_risk() route declines non-positive amounts. It also declines amounts above 50_000 minor units.

This threshold exists only to make both decisions observable in the lab. The response reason lets the Payments API explain each result.

  • Save risk_service.py.
  • Create another terminal in the Visual Studio Code terminal panel.
  • Start the Risk API on port 8001 by running this command:
uv run uvicorn risk_service:app --port 8001 --reload

What does this command do?

The command runs the app object from risk_service.py inside the synchronized project environment. Auto-reload keeps the process aligned with saved code changes.

  • Visit http://127.0.0.1:8001/docs in your browser.
  • Expand the POST /risk-checks route to confirm its request and response schemas are documented.

You should see the Risk API documentation with POST /risk-checks listed. That route proves the second service is running independently.

Risk API documentation not loading?

Check the Risk API terminal for a Python syntax error. Confirm that the terminal is inside the fintech-api-lab folder.

Make sure port 8001 appears in the command. The Payments API already uses port 8000.

Help me diagnose my Risk API.

✔️ Awesome, I've got everything!

Your Risk API is running on port 8001. Keep its Uvicorn terminal running for the next substep.

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Risk API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: str


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str


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


@app.post("/risk-checks", response_model=RiskCheckResponse)
def check_risk(risk: RiskCheckRequest) -> RiskCheckResponse:
    if risk.amount_minor <= 0:
        return RiskCheckResponse(approved=False, reason="amount must be positive")
    if risk.amount_minor > 50_000:
        return RiskCheckResponse(approved=False, reason="training risk limit exceeded")
    return RiskCheckResponse(approved=True, reason="within training risk limit")
Connect Payments to Risk

The Payments API now needs an HTTP client to cross the service boundary. HTTPX sends the payment data to the Risk API with a two-second timeout.

  • Switch back to payments_service.py in Visual Studio Code.
  • Replace its current contents with this first section:
from typing import Annotated, Literal
from uuid import uuid4

import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict

RISK_API_URL = "http://127.0.0.1:8001/risk-checks"
app = FastAPI(title="Payments API", version="1.0.0")


class PaymentRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED", "DECLINED"]
    reason: str


payments_by_idempotency_key: dict[str, tuple[tuple[str, int, str], PaymentResponse]] = {}


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

What changed in the contracts?

  • The RISK_API_URL constant points to the second service's risk-check route.
  • The local RiskCheckResponse model validates the JSON returned by that route.
  • The payment response now supports both APPROVED and DECLINED outcomes.
  • The reason field carries the explanation from the Risk API.
  • Append the payment route below health() by pasting this code:
@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(payment: PaymentRequest, idempotency_key: Annotated[str, Header(min_length=1)]) -> PaymentResponse:
    fingerprint = (payment.account_id, payment.amount_minor, payment.currency)
    previous = payments_by_idempotency_key.get(idempotency_key)
    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(status_code=409, detail={"code": "IDEMPOTENCY_KEY_REUSED"})
        return previous_response
    risk_response = httpx.post(RISK_API_URL, json=payment.model_dump(), timeout=2.0)
    risk_response.raise_for_status()
    risk_result = RiskCheckResponse(**risk_response.json())
    payment_response = PaymentResponse(payment_id=str(uuid4()), status="APPROVED" if risk_result.approved else "DECLINED", reason=risk_result.reason)
    payments_by_idempotency_key[idempotency_key] = (fingerprint, payment_response)
    return payment_response

How does the service call work?

The existing fingerprint check still handles safe retries before another risk call occurs. A fresh payment sends its validated request data to RISK_API_URL.

The returned JSON is validated as RiskCheckResponse. Its Boolean decision selects the final payment status.

  • Save payments_service.py.
  • Switch back to the Payments API terminal from the previous step to confirm Uvicorn reloads the file without an error.
  • Run the existing demo from its separate terminal by running this command:
uv run python demo.py

What should you see?

The first payment should return APPROVED with the reason within training risk limit. The retry should return the original payment ID.

The changed payload should still return 409 with IDEMPOTENCY_KEY_REUSED. The service boundary has preserved your retry protection.

Seeing a failure during the first risk call?

Confirm the Risk API terminal is still running on port 8001. Confirm the Payments API terminal reloaded the saved file.

Check that RISK_API_URL ends with /risk-checks.

Help me debug the service call.

✔️ Awesome, I've got everything!

The Payments API now receives approval decisions from the separate Risk API. Your idempotency behavior still works across that boundary.

ⓧ I'd like to double check the full code

from typing import Annotated, Literal
from uuid import uuid4

import httpx
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, ConfigDict

RISK_API_URL = "http://127.0.0.1:8001/risk-checks"
app = FastAPI(title="Payments API", version="1.0.0")


class PaymentRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED", "DECLINED"]
    reason: str


payments_by_idempotency_key: dict[str, tuple[tuple[str, int, str], PaymentResponse]] = {}


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


@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(payment: PaymentRequest, idempotency_key: Annotated[str, Header(min_length=1)]) -> PaymentResponse:
    fingerprint = (payment.account_id, payment.amount_minor, payment.currency)
    previous = payments_by_idempotency_key.get(idempotency_key)
    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(status_code=409, detail={"code": "IDEMPOTENCY_KEY_REUSED"})
        return previous_response
    risk_response = httpx.post(RISK_API_URL, json=payment.model_dump(), timeout=2.0)
    risk_response.raise_for_status()
    risk_result = RiskCheckResponse(**risk_response.json())
    payment_response = PaymentResponse(payment_id=str(uuid4()), status="APPROVED" if risk_result.approved else "DECLINED", reason=risk_result.reason)
    payments_by_idempotency_key[idempotency_key] = (fingerprint, payment_response)
    return payment_response
Observe decisions and expose network failure

The demo needs a second payment above the training threshold. A fresh idempotency key keeps that declined request separate from the approved payment.

  • Switch back to demo.py in Visual Studio Code.
  • Replace its contents with this expanded demo:
from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
run_id = str(uuid4())
headers = {"Idempotency-Key": f"approved-{run_id}"}
approved_payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}
first = httpx.post(PAYMENTS_URL, json=approved_payload, headers=headers, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=approved_payload, headers=headers, timeout=2.0)
conflict = httpx.post(PAYMENTS_URL, json={**approved_payload, "amount_minor": 13_000}, headers=headers, timeout=2.0)
declined = httpx.post(PAYMENTS_URL, json={"account_id": "acct-2002", "amount_minor": 75_000, "currency": "USD"}, headers={"Idempotency-Key": f"declined-{run_id}"}, timeout=2.0)
pprint((first.status_code, first.json()))
pprint((retry.status_code, retry.json()))
pprint((conflict.status_code, conflict.json()))
pprint((declined.status_code, declined.json()))
schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
pprint(sorted(schema_response.json()["paths"]))

What does the expanded demo prove?

The approved request still exercises the first request plus retry behavior. The conflict request still tests changed payment data under the same key.

The new 75_000 request uses its own key. That amount reaches the declined branch in the Risk API.

  • Save demo.py.

Before you run the demo, which payment do you expect the Risk API to approve?

  • Run the expanded demo while both Uvicorn services are running by using this command:
uv run python demo.py

What should you see?

The 12_500 payment should return APPROVED. Its retry should show the same payment ID.

The 75_000 payment should return DECLINED with the training-limit reason. You now have two services producing distinct payment outcomes.

✔️ Awesome, I've got everything!

Your expanded demo covers approval, safe retry, conflict, and decline behavior. Keep both service terminals visible for the final check.

ⓧ I'd like to double check the full code

from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"
run_id = str(uuid4())
headers = {"Idempotency-Key": f"approved-{run_id}"}
approved_payload = {"account_id": "acct-1001", "amount_minor": 12_500, "currency": "USD"}
first = httpx.post(PAYMENTS_URL, json=approved_payload, headers=headers, timeout=2.0)
retry = httpx.post(PAYMENTS_URL, json=approved_payload, headers=headers, timeout=2.0)
conflict = httpx.post(PAYMENTS_URL, json={**approved_payload, "amount_minor": 13_000}, headers=headers, timeout=2.0)
declined = httpx.post(PAYMENTS_URL, json={"account_id": "acct-2002", "amount_minor": 75_000, "currency": "USD"}, headers={"Idempotency-Key": f"declined-{run_id}"}, timeout=2.0)
pprint((first.status_code, first.json()))
pprint((retry.status_code, retry.json()))
pprint((conflict.status_code, conflict.json()))
pprint((declined.status_code, declined.json()))
schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
pprint(sorted(schema_response.json()["paths"]))

Before the final run, do you think the Payments API can complete a fresh payment while its Risk API is stopped?

  • Switch to the terminal running the Risk API.
  • Stop only the Risk API by pressing Ctrl+C.
  • Return to the demo terminal.
  • Run the demo again with its freshly generated idempotency keys by using this command:
uv run python demo.py

This failure is intentional

The demo now stops during a fresh payment request. The Payments API terminal shows an unhandled downstream connection failure because the Risk API is unavailable.

This is the network boundary made visible. The client currently receives raw failure behavior instead of a controlled payment contract.

You have separated payment handling from risk policy. Next, you will trace the call across both services and turn this raw dependency failure into a safe response.

Trace Calls and Handle Service Failure

Your Payments API now gets approval decisions from a separate Risk API. The failure you triggered in the previous step showed how an unavailable dependency can leave clients with a raw connection error.

A correlation ID connects one request across service logs. In this step, you will also use HTTPX error handling to turn dependency failures into a stable API response.

In this step, get ready to:
  • Accept and log correlation IDs in the Risk API.
  • Forward correlation IDs through the Payments API.
  • Return a structured 503 response when the Risk API is unavailable.
Add correlation IDs to the Risk API

The Risk API needs one trace value for each decision. It accepts the caller's value when one exists. It generates a new value for direct requests.

  • In risk_service.py, replace the imports through the app declaration with this code:
from typing import Annotated
from uuid import uuid4

from fastapi import FastAPI, Header, Response
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Risk API", version="1.0.0")

What Do These Imports Enable?

  • The Annotated header input lets FastAPI read the incoming correlation ID.
  • The Response object lets the route add a response header.
  • The uuid4() fallback creates an ID when the caller does not provide one.
  • Replace the existing RiskCheckResponse model with this version:
class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str
    correlation_id: str

What Changes in the Contract?

Every validated risk response now includes correlation_id. The Payments API can associate the decision with one call chain.

  • Save risk_service.py.
  • Switch back to the Risk API terminal from the previous step.
  • Start the updated Risk API by running:
uv run uvicorn risk_service:app --port 8001 --reload

What Does This Command Start?

Uvicorn loads the Risk API on port 8001. The terminal remains occupied while the service runs.

You should see the Risk API complete its startup sequence without an import error. This proves the expanded model loads successfully.

  • Replace the existing check_risk() route in risk_service.py with this version:
@app.post("/risk-checks", response_model=RiskCheckResponse)
def check_risk(
    risk: RiskCheckRequest,
    response: Response,
    x_correlation_id: Annotated[str | None, Header()] = None,
) -> RiskCheckResponse:
    correlation_id = x_correlation_id or str(uuid4())
    response.headers["X-Correlation-ID"] = correlation_id

    if risk.amount_minor <= 0:
        approved = False
        reason = "amount must be positive"
    elif risk.amount_minor > 50_000:
        approved = False
        reason = "training risk limit exceeded"
    else:
        approved = True
        reason = "within training risk limit"

    print(
        f"risk-service correlation_id={correlation_id} "
        f"approved={approved} amount_minor={risk.amount_minor}"
    )

    return RiskCheckResponse(
        approved=approved,
        reason=reason,
        correlation_id=correlation_id,
    )

How Does the Trace Work?

  • The route reuses x_correlation_id when the caller supplies one.
  • The response header returns the selected ID to the caller.
  • The log line records the decision under the selected ID.
  • The typed response body carries the ID back to the Payments API.
  • Save risk_service.py.
  • Wait for the Risk API terminal to report a successful reload.
  • Switch back to the client terminal from earlier.
  • Confirm the updated route still processes payments by running:
uv run python demo.py

What Does This Check Prove?

The existing demo still reaches the Risk API. The Risk API terminal now prints a correlation_id= value for each new decision.

Your approved payment still succeeds. Your larger payment still returns a declined status.

Risk API Not Reloading?

Check that the import group includes Annotated. Confirm that it also includes uuid4.

Confirm that check_risk() passes correlation_id into RiskCheckResponse.

Help me debug the updated Risk API.

✔️ Awesome, I've got everything!

The Risk API now accepts a correlation ID. It returns the selected value in its header and body.

Keep the Risk API running in its terminal.

ⓧ I'd like to double check the full code

from typing import Annotated
from uuid import uuid4

from fastapi import FastAPI, Header, Response
from pydantic import BaseModel, ConfigDict

app = FastAPI(title="Risk API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: str


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str
    correlation_id: str


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


@app.post("/risk-checks", response_model=RiskCheckResponse)
def check_risk(
    risk: RiskCheckRequest,
    response: Response,
    x_correlation_id: Annotated[str | None, Header()] = None,
) -> RiskCheckResponse:
    correlation_id = x_correlation_id or str(uuid4())
    response.headers["X-Correlation-ID"] = correlation_id

    if risk.amount_minor <= 0:
        approved = False
        reason = "amount must be positive"
    elif risk.amount_minor > 50_000:
        approved = False
        reason = "training risk limit exceeded"
    else:
        approved = True
        reason = "within training risk limit"

    print(
        f"risk-service correlation_id={correlation_id} "
        f"approved={approved} amount_minor={risk.amount_minor}"
    )

    return RiskCheckResponse(
        approved=approved,
        reason=reason,
        correlation_id=correlation_id,
    )

How to Compare This File

Compare the imports first. Next, compare the response model and complete check_risk() route with your saved file.

Protect the Payments API boundary

The Payments API owns the public side of the call chain. It accepts a client correlation ID or creates one before contacting the Risk API.

A downstream request can fail before a usable response arrives. Catching httpx.HTTPError gives clients a controlled failure contract without exposing connection details.

  • In payments_service.py, replace the imports through the app declaration with this code:
import httpx
from fastapi import FastAPI, Header, HTTPException, Response
from pydantic import BaseModel, ConfigDict

RISK_API_URL = "http://127.0.0.1:8001/risk-checks"

app = FastAPI(title="Payments API", version="1.0.0")

Why Add Response?

The injected Response object lets the route set X-Correlation-ID. Response-model processing remains active.

  • Replace the opening of create_payment() through the previous lookup with this code:
@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(
    payment: PaymentRequest,
    response: Response,
    idempotency_key: Annotated[str, Header(min_length=1)],
    x_correlation_id: Annotated[str | None, Header()] = None,
) -> PaymentResponse:
    correlation_id = x_correlation_id or str(uuid4())
    response.headers["X-Correlation-ID"] = correlation_id

    fingerprint = (
        payment.account_id,
        payment.amount_minor,
        payment.currency,
    )
    previous = payments_by_idempotency_key.get(idempotency_key)

How Does the Call Chain Begin?

  • FastAPI maps the incoming header to x_correlation_id.
  • A new UUID fills the gap when the header is absent.
  • The selected value becomes the Payments API response header.
  • The payment fingerprint continues to protect idempotent retries.
  • Save payments_service.py.
  • Wait for the Payments API terminal to report a successful reload.
  • Switch back to the client terminal.
  • Confirm the route still handles the existing payment flow by running:
uv run python demo.py

What Does This Run Confirm?

The updated route signature loads successfully. The existing payment results still print in the client terminal.

  • Replace the existing idempotency branch in create_payment() with this structured error section:
    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "IDEMPOTENCY_KEY_REUSED",
                    "message": "This idempotency key was used with different payment data.",
                    "correlation_id": correlation_id,
                },
                headers={"X-Correlation-ID": correlation_id},
            )
        return previous_response

    if payment.amount_minor <= 0:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "INVALID_AMOUNT",
                "message": "amount_minor must be greater than zero.",
                "correlation_id": correlation_id,
            },
            headers={"X-Correlation-ID": correlation_id},
        )

How Do Local Errors Stay Traceable?

Conflict responses now contain a safe public message. The body and header carry the same correlation ID.

The amount check gives invalid requests the same traceable error structure.

  • Replace the existing Risk API request lines with this protected downstream call:
    print(
        f"payments-service correlation_id={correlation_id} "
        "calling=risk-service"
    )

    try:
        risk_response = httpx.post(
            RISK_API_URL,
            json={
                "account_id": payment.account_id,
                "amount_minor": payment.amount_minor,
                "currency": payment.currency,
            },
            headers={"X-Correlation-ID": correlation_id},
            timeout=2.0,
        )
        risk_response.raise_for_status()
    except httpx.HTTPError as exc:
        raise HTTPException(
            status_code=503,
            detail={
                "code": "RISK_SERVICE_UNAVAILABLE",
                "message": "The payment cannot be assessed right now. Retry later.",
                "correlation_id": correlation_id,
            },
            headers={"X-Correlation-ID": correlation_id},
        ) from exc

How Is Dependency Failure Controlled?

  • The Payments API logs the ID before contacting the Risk API.
  • The downstream request forwards the same ID in X-Correlation-ID.
  • The 2.0 second timeout limits the wait for the dependency.
  • The exception handler converts HTTPX failures into a structured 503 response.
  • Save payments_service.py.
  • Wait for the Payments API terminal to report a successful reload.
  • Switch back to the client terminal.
  • Confirm the healthy downstream call still works by running:
uv run python demo.py

What Does the Healthy Run Prove?

The Payments API still receives approved and declined decisions. Both service terminals now log the same correlation ID for each request that reaches the Risk API.

  • Replace the response models through the idempotency-store declaration with this code:
class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str
    correlation_id: str


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED", "DECLINED"]
    reason: str
    correlation_id: str


payments_by_idempotency_key: dict[
    str,
    tuple[tuple[str, int, str], PaymentResponse],
] = {}

What Do the Models Preserve?

The downstream model validates the Risk API's correlation ID. The public payment model exposes the call-chain ID to the client.

The idempotency store keeps the original payment response. A safe retry therefore keeps its original trace data.

  • Replace the payment-response creation through the end of create_payment() with this code:
    risk_result = RiskCheckResponse(**risk_response.json())
    payment_response = PaymentResponse(
        payment_id=str(uuid4()),
        status="APPROVED" if risk_result.approved else "DECLINED",
        reason=risk_result.reason,
        correlation_id=correlation_id,
    )

    payments_by_idempotency_key[idempotency_key] = (
        fingerprint,
        payment_response,
    )
    return payment_response

How Does the Payment Keep Its Trace?

The payment result receives the correlation ID selected at the start of the request. The idempotency store preserves that result for a safe retry.

  • Save payments_service.py.
  • Switch back to the client terminal.
  • Confirm every successful payment body now contains trace data by running:
uv run python demo.py

What Should Change?

Each successful payment body now contains a correlation_id field. The first request and safe retry still share one payment ID.

Payments API Not Reloading?

Check that Response appears in the FastAPI import. Confirm that both response models include correlation_id.

Check the indentation around try: and except httpx.HTTPError as exc:. Both blocks belong inside create_payment().

Help me debug the Payments API changes.

✔️ Awesome, I've got everything!

The Payments API now forwards trace headers. It also converts downstream HTTP failures into a controlled response.

ⓧ I'd like to double check the full code

from typing import Annotated, Literal
from uuid import uuid4

import httpx
from fastapi import FastAPI, Header, HTTPException, Response
from pydantic import BaseModel, ConfigDict

RISK_API_URL = "http://127.0.0.1:8001/risk-checks"

app = FastAPI(title="Payments API", version="1.0.0")


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

    account_id: str
    amount_minor: int
    currency: Literal["USD"]


class RiskCheckResponse(BaseModel):
    approved: bool
    reason: str
    correlation_id: str


class PaymentResponse(BaseModel):
    payment_id: str
    status: Literal["APPROVED", "DECLINED"]
    reason: str
    correlation_id: str


payments_by_idempotency_key: dict[
    str,
    tuple[tuple[str, int, str], PaymentResponse],
] = {}


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


@app.post("/payments", response_model=PaymentResponse, status_code=201)
def create_payment(
    payment: PaymentRequest,
    response: Response,
    idempotency_key: Annotated[str, Header(min_length=1)],
    x_correlation_id: Annotated[str | None, Header()] = None,
) -> PaymentResponse:
    correlation_id = x_correlation_id or str(uuid4())
    response.headers["X-Correlation-ID"] = correlation_id

    fingerprint = (
        payment.account_id,
        payment.amount_minor,
        payment.currency,
    )
    previous = payments_by_idempotency_key.get(idempotency_key)

    if previous is not None:
        previous_fingerprint, previous_response = previous
        if previous_fingerprint != fingerprint:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "IDEMPOTENCY_KEY_REUSED",
                    "message": "This idempotency key was used with different payment data.",
                    "correlation_id": correlation_id,
                },
                headers={"X-Correlation-ID": correlation_id},
            )
        return previous_response

    if payment.amount_minor <= 0:
        raise HTTPException(
            status_code=400,
            detail={
                "code": "INVALID_AMOUNT",
                "message": "amount_minor must be greater than zero.",
                "correlation_id": correlation_id,
            },
            headers={"X-Correlation-ID": correlation_id},
        )

    print(
        f"payments-service correlation_id={correlation_id} "
        "calling=risk-service"
    )

    try:
        risk_response = httpx.post(
            RISK_API_URL,
            json={
                "account_id": payment.account_id,
                "amount_minor": payment.amount_minor,
                "currency": payment.currency,
            },
            headers={"X-Correlation-ID": correlation_id},
            timeout=2.0,
        )
        risk_response.raise_for_status()
    except httpx.HTTPError as exc:
        raise HTTPException(
            status_code=503,
            detail={
                "code": "RISK_SERVICE_UNAVAILABLE",
                "message": "The payment cannot be assessed right now. Retry later.",
                "correlation_id": correlation_id,
            },
            headers={"X-Correlation-ID": correlation_id},
        ) from exc

    risk_result = RiskCheckResponse(**risk_response.json())
    payment_response = PaymentResponse(
        payment_id=str(uuid4()),
        status="APPROVED" if risk_result.approved else "DECLINED",
        reason=risk_result.reason,
        correlation_id=correlation_id,
    )

    payments_by_idempotency_key[idempotency_key] = (
        fingerprint,
        payment_response,
    )
    return payment_response

How to Compare This File

Compare both response models first. Next, compare the complete create_payment() route with your saved file.

Update the demo and test both outcomes

The client needs to display each status code clearly. It also needs to surface the correlation header so you can compare healthy calls with dependency failures.

  • In demo.py, replace the imports through the code before run_id with this section:
from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"


def show(label: str, response: httpx.Response) -> None:
    print(f"\n{label}")
    print(f"status={response.status_code}")
    print(f"correlation_id={response.headers.get('x-correlation-id')}")
    try:
        pprint(response.json())
    except ValueError:
        print(response.text)

What Does the Helper Show?

The show() helper prints a label for each response. It then displays the status code and correlation header.

The JSON body appears next. Plain text remains readable if a response does not contain JSON.

  • Replace the existing schema output at the bottom of demo.py with this section:
schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
print("\nOpenAPI paths")
pprint(sorted(schema_response.json()["paths"]))

Why Keep the Schema Check?

The final request confirms that the Payments API still publishes its OpenAPI contract. The listed route proves the reliability changes kept the payment endpoint available.

  • Save demo.py.
  • Switch back to the client terminal.
  • Confirm the partial demo changes remain runnable by executing:
uv run python demo.py

What Should Appear?

The existing payment results still print. The schema section now begins with the OpenAPI paths label.

  • Replace the old request section that starts with run_id = str(uuid4()) by adding this first section:
run_id = str(uuid4())
approved_headers = {
    "Idempotency-Key": f"approved-{run_id}",
    "X-Correlation-ID": f"trace-{run_id}",
}
approved_payload = {
    "account_id": "acct-1001",
    "amount_minor": 12_500,
    "currency": "USD",
}

first = httpx.post(
    PAYMENTS_URL,
    json=approved_payload,
    headers=approved_headers,
    timeout=2.0,
)
retry = httpx.post(
    PAYMENTS_URL,
    json=approved_payload,
    headers=approved_headers,
    timeout=2.0,
)
show("first request", first)
show("safe retry", retry)

Why Reuse These Headers?

Both calls reuse one idempotency key. They also reuse one correlation ID so the retry remains part of the same traceable operation.

  • Add this conflict and declined-payment section immediately below the retry section:
conflicting_payload = {**approved_payload, "amount_minor": 13_000}
conflict = httpx.post(
    PAYMENTS_URL,
    json=conflicting_payload,
    headers=approved_headers,
    timeout=2.0,
)
show("same key with changed data", conflict)

declined = httpx.post(
    PAYMENTS_URL,
    json={
        "account_id": "acct-2002",
        "amount_minor": 75_000,
        "currency": "USD",
    },
    headers={
        "Idempotency-Key": f"declined-{run_id}",
        "X-Correlation-ID": f"declined-trace-{run_id}",
    },
    timeout=2.0,
)
show("risk-declined payment", declined)

What Paths Does This Cover?

The changed payload exercises unsafe idempotency-key reuse. The larger payment exercises the Risk API's training decline path.

Every result now shows its status and correlation header before its body. The client output makes each contract visible.

  • Save demo.py.
  • Switch back to the client terminal.
  • Run the completed demo while both services are active:
uv run python demo.py

What Does the Traced Run Prove?

The client displays each status code and correlation header. Both service terminals print matching IDs for calls that reach the Risk API.

That is the healthy tracing path complete. One payment request can now be followed through both local services.

Demo Not Printing Correlation IDs?

Check that approved_headers includes X-Correlation-ID.

Confirm that show() reads x-correlation-id from the response headers.

Help me display the correlation IDs.

✔️ Awesome, I've got everything!

The demo now labels every result. It displays status codes, correlation headers, response bodies, and the published API path.

ⓧ I'd like to double check the full code

from pprint import pprint
from uuid import uuid4

import httpx

PAYMENTS_URL = "http://127.0.0.1:8000/payments"
OPENAPI_URL = "http://127.0.0.1:8000/openapi.json"


def show(label: str, response: httpx.Response) -> None:
    print(f"\n{label}")
    print(f"status={response.status_code}")
    print(f"correlation_id={response.headers.get('x-correlation-id')}")
    try:
        pprint(response.json())
    except ValueError:
        print(response.text)


run_id = str(uuid4())
approved_headers = {
    "Idempotency-Key": f"approved-{run_id}",
    "X-Correlation-ID": f"trace-{run_id}",
}
approved_payload = {
    "account_id": "acct-1001",
    "amount_minor": 12_500,
    "currency": "USD",
}

first = httpx.post(
    PAYMENTS_URL,
    json=approved_payload,
    headers=approved_headers,
    timeout=2.0,
)
retry = httpx.post(
    PAYMENTS_URL,
    json=approved_payload,
    headers=approved_headers,
    timeout=2.0,
)
show("first request", first)
show("safe retry", retry)

conflicting_payload = {**approved_payload, "amount_minor": 13_000}
conflict = httpx.post(
    PAYMENTS_URL,
    json=conflicting_payload,
    headers=approved_headers,
    timeout=2.0,
)
show("same key with changed data", conflict)

declined = httpx.post(
    PAYMENTS_URL,
    json={
        "account_id": "acct-2002",
        "amount_minor": 75_000,
        "currency": "USD",
    },
    headers={
        "Idempotency-Key": f"declined-{run_id}",
        "X-Correlation-ID": f"declined-trace-{run_id}",
    },
    timeout=2.0,
)
show("risk-declined payment", declined)

schema_response = httpx.get(OPENAPI_URL, timeout=2.0)
schema_response.raise_for_status()
print("\nOpenAPI paths")
pprint(sorted(schema_response.json()["paths"]))

How to Compare This File

Compare the show() helper first. Next, compare the request headers and payment calls with your saved file.

Before you test the dependency failure, do you expect the Payments API process to stop or return a structured response?

  • Switch back to the Risk API terminal.
  • Press Ctrl+C to stop only the Risk API.
  • Return to the client terminal.
  • Run the demo while the Risk API is stopped:
uv run python demo.py

What Does the Failure Reveal?

The fresh payment calls return status=503 while the Payments API stays active. Each failure response includes an X-Correlation-ID header.

The JSON detail contains RISK_SERVICE_UNAVAILABLE with a safe public message. Clients now receive a failure contract they can handle.

Not Receiving the Structured 503?

Confirm that you stopped the Risk API on port 8001. Keep the Payments API active on port 8000.

If the client receives a raw connection error, check that httpx.post() remains inside the try: block. Confirm that raise_for_status() is inside the same block.

Help me get the structured failure response.

  • Return to the Risk API terminal.
  • Restart the Risk API by running:
uv run uvicorn risk_service:app --port 8001 --reload

What Does Restarting Restore?

The Risk API begins accepting risk checks on port 8001 again. Its terminal is ready to log forwarded correlation IDs.

Before the final run, which payment results do you expect to replace the dependency-failure responses?

  • Return to the client terminal.
  • Run the demo again with both services active:
uv run python demo.py

What Does Success Look Like?

  • The first request and safe retry return 201 with the same payment ID.
  • The changed payload returns 409 with IDEMPOTENCY_KEY_REUSED.
  • The larger payment returns DECLINED.
  • The schema output lists /payments.

Compare one trace- value in the Payments API terminal with the Risk API terminal. You should see the exact same correlation ID in both logs.

Correlation IDs Do Not Match?

Check that the Payments API forwards headers={"X-Correlation-ID": correlation_id} in its Risk API request.

Check that the Risk API uses x_correlation_id before generating a new UUID.

Help me trace the mismatched IDs.

That is the reliability boundary complete. Your two services now share trace data and return a controlled response when the Risk API is unavailable.

Secret mission

Add Payment Lookup

Your Payments API can create retry-safe payments. Clients also need a reliable way to retrieve a payment after creation. In this mission, you will add a lookup endpoint with a documented success response and a structured not-found response.

Clean Up Your Resources

Clean Up Your Resources

Everything you built stays on your Linux machine at no ongoing cost. Decide whether to keep it running, pause the local services, or delete the lab entirely.

Resources you used:

  • The fintech-api-lab folder on your Desktop.
  • The synchronized Python project environment inside fintech-api-lab.
  • The Payments API process on port 8000. It holds the payment records in memory. It also holds the idempotency records in memory.
  • The Risk API process on port 8001.

Keep everything running

No action is needed. Choose this option if you want to keep testing the payment workflow or extend its API contracts.

  • Your fintech-api-lab folder stays on your Desktop.
  • The Payments API continues running on port 8000.
  • The Risk API continues running on port 8001.
  • Your in-memory payment records remain available while the Payments API process stays running.
  • Everything remains local at no ongoing cost.

Pause - I'll come back to this later

Shut down both service processes to free their ports. Your project files remain on your machine for another session.

  • Switch back to the Payments API terminal from earlier.
  • Press Ctrl+C to stop the Payments API.
  • Switch back to the Risk API terminal from earlier.
  • Press Ctrl+C to stop the Risk API.

That frees both service ports while preserving every project file. Stopping the Payments API clears its in-memory payment records and idempotency records.

Delete - I don't want to use this again

Removing this folder is permanent. The command targets only your local fintech-api-lab project.

  • Switch back to the Payments API terminal from earlier.
  • Press Ctrl+C to stop the Payments API.
  • Switch back to the Risk API terminal from earlier.
  • Press Ctrl+C to stop the Risk API.
  • Delete the fintech-api-lab folder from your Desktop by running these commands:
cd ~/Desktop
rm -rf fintech-api-lab

What Gets Removed?

  • The first line moves your terminal to the Desktop where the project was created.
  • The second line permanently removes the project folder. This includes the Python files and the synchronized project environment.
  • Confirm that fintech-api-lab is absent by viewing your Desktop in your Linux file manager.

Nice Work!

Nice Work!

Strong finish. You built two local FastAPI services that process typed payments through a documented REST contract.

You've learned how to:

  • Build typed Payments and Risk APIs with automatically generated OpenAPI contracts. Connect the services through synchronous HTTP calls.
  • Reproduce duplicate payment IDs caused by a retry. Add an idempotency key that returns the original payment. Reject changed payment data with a structured 409 conflict.
  • Trace a service call with one correlation ID across the client response. Find that same ID in both service logs. Convert Risk API outages into a controlled failure contract with a 503 response.
  • Secret Mission: Add GET /payments/{payment_id} for payment lookup. Return a stored PaymentResponse for known IDs. Return a structured 404 response with PAYMENT_NOT_FOUND for unknown IDs.

Ready to quiz yourself?