Build a Payment Processing API

Build a FastAPI payment system with idempotency, state machines, and webhooks.

Introduction

30 Second Summary

Every time you tap your card or hit "Pay now," a system behind the scenes ensures you're charged exactly once and the payment moves through the correct stages. These reliability patterns are what payment companies test for in system design interviews.

In this project, you will build a payment processing API that implements idempotency keys, a payment state machine, and webhook dispatch. You'll walk away with a working system and the ability to articulate why each design decision exists.

What You'll Build

A payment processing API built with Python and FastAPI that handles payment creation, state transitions, idempotency protection against double charges, and webhook notifications for async events.

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

  • A FastAPI application with clean payment endpoints protected by idempotency middleware that prevents duplicate charges.
  • A state machine enforcing valid payment lifecycle transitions and a webhook dispatch system with retry logic for notifying external services.
  • A test script that simulates failures and proves your system handles them correctly, plus the vocabulary to explain your design trade-offs in an interview.
  • Secret Mission: Add a circuit breaker pattern that protects your payment processor from cascading failures.

Are there any prerequisites?

You should be comfortable with basic Python concepts like functions, dictionaries, and classes. No prior API development or system design experience is needed.

Before We Start

Before We Start...

Set Up Your Development Environment

To build a payment processing API, you need a local development environment with Python and FastAPI ready to serve requests.

In this step, you'll install Python, create an isolated project folder with a virtual environment, and get a minimal FastAPI server running with interactive documentation.

In this step, get ready to:
  • Install and verify Python 3.9+.
  • Create a project directory with a virtual environment and install FastAPI.
  • Create a minimal FastAPI app and confirm it runs with interactive docs.
Install Python and verify your version

Python is the language you'll use to build the payment API. You need version 3.9 or higher because FastAPI requires it.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

✔️ I see version 3.9 or higher

You should see output like Python 3.11.4 or any version at or above 3.9. Python is ready to go.

ⓧ I see an older version

Your system has a version of Python below 3.9. macOS often ships with an older system Python that shadows newer installs.

  • Download and install Python 3.9 or higher from the official site: https://www.python.org/downloads/.
  • After installing, close and reopen your terminal, then run the version check again.
  • If your system still shows the old version, try running python3.11 --version (or whichever version you installed) to confirm the new install is accessible.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download and install Python 3.9 or higher from the official site: https://www.python.org/downloads/.
  • During installation on Windows, check the box that says Add Python to PATH.
  • After installing, close and reopen your terminal, then run the version check again.
Create your project directory and install FastAPI

A virtual environment keeps your project's dependencies isolated from other Python projects on your machine. This prevents version conflicts between projects.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir payment-api
cd payment-api
  • Confirm the folder was created by running:
ls

The folder will be empty for now. Next, you'll create a virtual environment inside it.

  • Create and activate a virtual environment by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran both commands. The first command creates the environment, and the second activates it.

On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned first, then try activating again.

Help me activate my virtual environment.

  • Install FastAPI and its standard dependencies by running:
pip install "fastapi[standard]"

What does this install?

The fastapi[standard] package includes FastAPI itself plus common tools you'll need: an HTTP server (Uvicorn), a test client (httpx), and other utilities.

You should see output ending with a success message listing the installed packages.

Seeing a permission error or install failure?

Make sure (venv) appears in your prompt. If it does not, your virtual environment is not active. Go back and run the activation command.

Help me install FastAPI in my virtual environment.

Create a minimal FastAPI app and run the dev server

With FastAPI installed, you can create a minimal app to confirm everything is wired up correctly.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type the name of your code editor (such as Cursor or VS Code) and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the payment-api folder on your Desktop, then click Open.
  • Create a new file called main.py in the payment-api folder.
  • Paste the following code into main.py:
from fastapi import FastAPI

# Create the application instance with a title and version for the docs
app = FastAPI(title="Payment Processing API", version="1.0.0")

What does this code do?

  • from fastapi import FastAPI imports the FastAPI framework.
  • app = FastAPI(...) creates your application instance. The title and version show up in the auto-generated API documentation.
  • Save main.py.
  • Switch back to your terminal and start the development server by running:
fastapi dev main.py

You should see output indicating the server is running at http://127.0.0.1:8000.

  • Open your browser and navigate to http://127.0.0.1:8000/docs.

You should see the FastAPI interactive documentation page with your API title "Payment Processing API" displayed at the top.

Page not loading?

Make sure the server is still running in your terminal. If you see an error about the port being in use, another process is occupying port 8000. Close any other servers you may have running.

If you see a ModuleNotFoundError, your virtual environment may not be active. Check for (venv) in your terminal prompt and reactivate if needed.

Help me get my FastAPI server running.

Why interactive docs?

FastAPI auto-generates an interactive API explorer from your code. As you add endpoints in the next steps, they'll appear here automatically. You can test every endpoint directly in the browser without needing a separate tool.

✔️ Awesome, I've got everything!

Great. Make sure you've saved main.py and your server is running.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

from fastapi import FastAPI

app = FastAPI(title="Payment Processing API", version="1.0.0")

Your development environment is ready. Next up, you'll design the payment data models and build your first API endpoints.

Design the Payment API and Data Models

Your development environment is ready and your FastAPI server is running. Now it's time to define what a payment looks like in your system and build the first endpoints that create and retrieve payments.

Every payment API starts with a clear data contract. The contract defines exactly what fields a client sends when creating a payment and what the server sends back. This is the foundation that Stripe interviewers probe when they ask you to design a payment API from scratch.

In this step, get ready to:
  • Define data models that form the payment API contract.
  • Build the POST /payments endpoint to create new payments.
  • Add the GET /payments/{payment_id} endpoint and test both via interactive docs.
Define the payment data models

Before you can create or retrieve payments, your API needs to know what a payment looks like. Pydantic models enforce the shape of incoming requests and outgoing responses at the framework level.

  • Replace the entire contents of main.py with the following code to define your payment data models:
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str

What does this code do?

This code defines the data contract for your payment API:

  • PaymentStatus is an enum that lists every state a payment can be in. Using an enum means your code will reject any status value that isn't in this list.
  • PaymentRequest defines what a client must send to create a payment. The Field(gt=0) constraint means the amount must be greater than zero.
  • PaymentResponse defines what the server sends back. It includes server-generated fields like id and created_at that the client never sends.
  • Save main.py.

You won't see a change in the browser yet. The next chunk adds the first endpoint that uses these models.

Why define the full lifecycle upfront?

The PaymentStatus enum includes states like processing and refunded that you won't use until later steps. Defining the full lifecycle now is intentional.

In a system design interview, showing that you've thought through the complete state space before writing endpoint logic signals maturity. It's the same approach Stripe uses internally.

Create the payments store and POST endpoint

With your data models defined, you need somewhere to store payments and an endpoint that accepts new ones. For this project, an in-memory dictionary keeps things focused on the patterns rather than database setup.

  • In main.py, add the following code below your PaymentResponse class:
# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}


# --- API Endpoints ---

@app.post("/payments", response_model=PaymentResponse, status_code=201)
async def create_payment(payment: PaymentRequest):
    payment_id = f"pay_{uuid.uuid4().hex[:16]}"
    now = datetime.now(timezone.utc).isoformat()

    payment_record = {
        "id": payment_id,
        "status": PaymentStatus.PENDING,
        "amount": payment.amount,
        "currency": payment.currency,
        "customer_id": payment.customer_id,
        "description": payment.description,
        "created_at": now,
    }
    payments_store[payment_id] = payment_record
    return payment_record

What does this code do?

  • payments_store is a dictionary that maps payment IDs to payment records. In production you'd use a database, but the in-memory store lets you focus on the logic.
  • The POST /payments endpoint generates a unique ID with the pay_ prefix, timestamps the creation, and stores the record with a pending status.
  • response_model=PaymentResponse tells FastAPI to validate and serialize the output. status_code=201 signals that a new resource was created.
  • Save main.py.
  • Switch back to your browser and refresh the interactive docs at http://127.0.0.1:8000/docs. You should see the POST /payments endpoint listed under default.

Don't see the endpoint in /docs?

  • Check that your dev server terminal isn't showing an error. A missing import or syntax error prevents the reload.
  • Make sure you saved main.py. The dev server only reloads on file save.
  • Verify the new code is placed below the PaymentResponse class, not inside it.

Help me debug why my FastAPI endpoint isn't showing in /docs

Add the GET endpoint and test both endpoints

Creating payments is only useful if you can retrieve them later. The GET endpoint lets clients check a payment's current status by its ID.

  • In main.py, add the following code below the create_payment function:
@app.get("/payments/{payment_id}", response_model=PaymentResponse)
async def get_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")
    return payments_store[payment_id]

What does this code do?

  • The path parameter {payment_id} captures the ID from the URL and passes it to the function.
  • If the payment doesn't exist in the store, the endpoint returns a 404 with a clear error message.
  • This is the pattern real payment APIs use. Clients poll this endpoint to check whether an async payment has succeeded or failed.
  • Save main.py.
  • Refresh http://127.0.0.1:8000/docs in your browser. You should now see both POST /payments and GET /payments/{payment_id} listed.

Now test the endpoints using the interactive docs:

  • Click on POST /payments to expand it.
  • Click Try it out.
  • Replace the example request body with this JSON:
{
  "amount": 5000,
  "currency": "usd",
  "customer_id": "cus_123",
  "description": "Test payment"
}
  • Click Execute.

You should see a 201 response with a generated payment ID, pending status, and a UTC timestamp.

  • Copy the id value from the response (it starts with pay_).
  • Click on GET /payments/{payment_id} to expand it.
  • Click Try it out, paste the payment ID into the payment_id field, and click Execute.

You should see the same payment returned with all its fields intact.

Getting a 422 validation error?

  • Make sure amount is a positive integer (no decimals, no zero). The Field(gt=0) constraint rejects zero and negative values.
  • Check that currency is exactly 3 lowercase letters (like usd). Uppercase or longer strings fail the regex pattern.
  • Confirm customer_id is present. It's a required field with no default.

Help me fix a 422 error from my POST /payments endpoint

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and both endpoints are responding correctly in the interactive docs.

ⓧ I'd like to double check the full code

Here's the complete main.py at this point:

import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str


# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}


# --- API Endpoints ---

@app.post("/payments", response_model=PaymentResponse, status_code=201)
async def create_payment(payment: PaymentRequest):
    payment_id = f"pay_{uuid.uuid4().hex[:16]}"
    now = datetime.now(timezone.utc).isoformat()

    payment_record = {
        "id": payment_id,
        "status": PaymentStatus.PENDING,
        "amount": payment.amount,
        "currency": payment.currency,
        "customer_id": payment.customer_id,
        "description": payment.description,
        "created_at": now,
    }
    payments_store[payment_id] = payment_record
    return payment_record


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
async def get_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")
    return payments_store[payment_id]

Your payment API now has a clear data contract and two working endpoints. Next up, you'll add a state machine that controls how payments transition through their lifecycle, preventing invalid moves like processing an already-succeeded payment.

Implement the Payment State Machine

Your payment API can create and retrieve payments, but right now a payment stays in "pending" forever. Real payment systems like Stripe move payments through a strict lifecycle: pending, processing, succeeded, or failed.

A state machine controls which transitions are allowed. A pending payment can move to processing, but a failed payment cannot jump straight to succeeded. In this step, you'll encode those rules and build an endpoint that processes payments while enforcing them.

In this step, get ready to:
  • Define the valid state transitions for your payment lifecycle.
  • Build a process payment endpoint that enforces the state machine.
  • Test that invalid transitions are rejected with 409 Conflict.
Define the valid state transitions

A state machine needs an explicit map of which states can lead to which other states. Without this, your API would let clients move a payment from "failed" to "succeeded" directly, which makes no sense in the real world.

  • In main.py, find the # --- In-Memory Stores --- comment.
  • Add the following state machine definition directly above that comment:
# --- State Machine ---

VALID_TRANSITIONS: dict[PaymentStatus, list[PaymentStatus]] = {
    PaymentStatus.PENDING: [PaymentStatus.PROCESSING],
    PaymentStatus.PROCESSING: [PaymentStatus.SUCCEEDED, PaymentStatus.FAILED],
    PaymentStatus.SUCCEEDED: [PaymentStatus.REFUNDED],
    PaymentStatus.FAILED: [],
    PaymentStatus.REFUNDED: [],
}

What does this code do?

Each key in VALID_TRANSITIONS is a current payment status. The value is a list of statuses that payment is allowed to move to next.

  • A pending payment can only move to processing.
  • A processing payment resolves to either succeeded or failed.
  • A succeeded payment can be refunded, but a failed or refunded payment goes nowhere (empty list).
  • Save main.py.

You won't see a visible change yet. The next chunk uses this dictionary to enforce transitions in the process endpoint.

Build the process payment endpoint

The process endpoint simulates what happens when your API talks to an external payment provider. It transitions a payment from pending through processing, then resolves to succeeded or failed based on a simulated success rate.

  • At the top of main.py, find this line:
import uuid
  • Add import random on a new line directly above it, so your imports look like this:
import random
import uuid
  • Now scroll to the bottom of main.py and add the process endpoint below the get_payment function:
@app.post("/payments/{payment_id}/process", response_model=PaymentResponse)
async def process_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")

    payment = payments_store[payment_id]
    current_status = payment["status"]

    # Validate transition to processing
    if PaymentStatus.PROCESSING not in VALID_TRANSITIONS.get(current_status, []):
        raise HTTPException(
            status_code=409,
            detail=f"Cannot transition from {current_status} to processing",
        )

    # Transition to processing
    payment["status"] = PaymentStatus.PROCESSING

    # Simulate payment provider (80% success rate)
    if random.random() < 0.8:
        payment["status"] = PaymentStatus.SUCCEEDED
    else:
        payment["status"] = PaymentStatus.FAILED

    return payment

What does this code do?

  • The endpoint looks up the payment and checks its current status against VALID_TRANSITIONS. If processing is not in the allowed list, it returns a 409 Conflict.
  • If the transition is valid, it moves the payment to "processing" first, then simulates calling an external provider with an 80% success rate.
  • The payment lands on either "succeeded" or "failed" depending on the random outcome. This mirrors how real providers return asynchronous results.

Why 409 Conflict?

HTTP 409 means the request conflicts with the current state of the resource. This is the standard status code for telling clients "this payment has already moved past the point where you can do that." Stripe interviewers expect you to choose status codes that communicate intent clearly.

  • Save main.py.
  • Your FastAPI dev server should auto-reload. Check the terminal for confirmation that it reloaded without errors.

Seeing a syntax error in the terminal?

  • Make sure import random is on its own line at the top of the file, not inside another statement.
  • Check that the process_payment function has the same indentation as the other endpoint functions above it.
  • Verify the f-string in the HTTPException uses curly braces around current_status: f"Cannot transition from {current_status} to processing".

Why is my server not reloading after save?

Test the state machine in the interactive docs

The best way to prove your state machine works is to create a payment, process it once (should succeed), then try to process it again (should be rejected).

  • Open http://127.0.0.1:8000/docs in your browser.
  • Expand the POST /payments endpoint and click Try it out.
  • Replace the request body with this and click Execute:
{
  "amount": 5000,
  "currency": "usd",
  "customer_id": "cus_123",
  "description": "Test payment"
}
  • Copy the id value from the response (it looks like pay_abc123...).
  • Expand the POST /payments/{payment_id}/process endpoint and click Try it out.
  • Paste the payment ID into the payment_id field and click Execute.

You should see the payment status change to either "succeeded" or "failed" in the response body.

  • Click Execute again with the same payment ID.

This time you should see a 409 response with the message "Cannot transition from succeeded to processing" (or "failed" depending on your previous result). Your state machine is working.

Why does this matter in interviews?

Stripe interviewers specifically ask: "What happens if the client retries a payment that already succeeded?" Your state machine gives you a precise answer. The system rejects invalid transitions with 409, preventing double charges. This is reliability through correctness, not just through retries.

Getting 404 instead of 409?

  • Make sure you're using the payment ID from the create response, not an example like string or pay_123.
  • If you restarted the server, the in-memory store is empty. Create a new payment and try again.

Help me debug my process payment endpoint

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

import random
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str


# --- State Machine ---

VALID_TRANSITIONS: dict[PaymentStatus, list[PaymentStatus]] = {
    PaymentStatus.PENDING: [PaymentStatus.PROCESSING],
    PaymentStatus.PROCESSING: [PaymentStatus.SUCCEEDED, PaymentStatus.FAILED],
    PaymentStatus.SUCCEEDED: [PaymentStatus.REFUNDED],
    PaymentStatus.FAILED: [],
    PaymentStatus.REFUNDED: [],
}


# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}


# --- API Endpoints ---

@app.post("/payments", response_model=PaymentResponse, status_code=201)
async def create_payment(payment: PaymentRequest):
    payment_id = f"pay_{uuid.uuid4().hex[:16]}"
    now = datetime.now(timezone.utc).isoformat()

    payment_record = {
        "id": payment_id,
        "status": PaymentStatus.PENDING,
        "amount": payment.amount,
        "currency": payment.currency,
        "customer_id": payment.customer_id,
        "description": payment.description,
        "created_at": now,
    }
    payments_store[payment_id] = payment_record
    return payment_record


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
async def get_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")
    return payments_store[payment_id]


@app.post("/payments/{payment_id}/process", response_model=PaymentResponse)
async def process_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")

    payment = payments_store[payment_id]
    current_status = payment["status"]

    # Validate transition to processing
    if PaymentStatus.PROCESSING not in VALID_TRANSITIONS.get(current_status, []):
        raise HTTPException(
            status_code=409,
            detail=f"Cannot transition from {current_status} to processing",
        )

    # Transition to processing
    payment["status"] = PaymentStatus.PROCESSING

    # Simulate payment provider (80% success rate)
    if random.random() < 0.8:
        payment["status"] = PaymentStatus.SUCCEEDED
    else:
        payment["status"] = PaymentStatus.FAILED

    return payment

Your state machine now guarantees that payments can only follow valid paths through their lifecycle. Next up, you'll add idempotency keys to prevent duplicate charges when clients retry failed network requests.

Add Idempotency Key Protection

Your payment API now handles state transitions correctly. But there's a critical reliability problem that every payment system must solve: network retries.

When a customer clicks "Pay" and the network drops, their browser retries the request. Without protection, your API creates a second payment and charges them twice. Idempotency keys solve this by letting clients attach a unique identifier to each request so your server can recognize and safely replay duplicates.

In this step, get ready to:
  • Build an idempotency middleware that intercepts POST requests and caches responses.
  • Add body fingerprinting to detect mismatched request bodies reusing the same key.
  • Test that duplicate requests return the same payment and mismatched bodies are rejected.
Create the idempotency middleware

Middleware in FastAPI runs on every request before it reaches your endpoint and on every response before it goes back to the client. This is the perfect place to intercept duplicate requests.

  • In main.py, find the import lines at the top of the file.
  • Replace the entire import section and add the idempotency store, fingerprint helper, and middleware by updating the top portion of your file to match the following code:
import hashlib
import json
import random
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")

What changed in the imports?

  • hashlib provides SHA-256 hashing to fingerprint request bodies.
  • json lets you serialize and deserialize response bodies for caching.
  • Request gives the middleware access to the raw request object (headers, body, URL).
  • JSONResponse lets the middleware return cached responses directly without hitting the endpoint.
  • Next, find the # --- In-Memory Stores --- section and add the idempotency_store below payments_store. Then add the full idempotency middleware section between the stores and the # --- API Endpoints --- comment. Copy this code:
# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}
idempotency_store: dict[str, dict] = {}


# --- Idempotency Middleware ---

def fingerprint_body(body: bytes) -> str:
    return hashlib.sha256(body).hexdigest()


@app.middleware("http")
async def idempotency_middleware(request: Request, call_next):
    if request.method != "POST":
        return await call_next(request)

    idempotency_key = request.headers.get("Idempotency-Key")
    if not idempotency_key:
        return await call_next(request)

    body = await request.body()
    fp = fingerprint_body(body)
    cache_key = f"{request.url.path}:{idempotency_key}"

    if cache_key in idempotency_store:
        cached = idempotency_store[cache_key]
        if cached["fingerprint"] != fp:
            return JSONResponse(
                status_code=422,
                content={"error": "Idempotency key reused with different request body"},
            )
        return JSONResponse(
            status_code=cached["status_code"],
            content=cached["body"],
        )

    response = await call_next(request)

    response_body = b""
    async for chunk in response.body_iterator:
        response_body += chunk

    try:
        body_json = json.loads(response_body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        body_json = response_body.decode()

    idempotency_store[cache_key] = {
        "fingerprint": fp,
        "status_code": response.status_code,
        "body": body_json,
    }

    return JSONResponse(
        status_code=response.status_code,
        content=body_json,
    )

What does this middleware do?

The middleware follows a straightforward flow for every incoming POST request:

  • If the request has no Idempotency-Key header, it passes through normally with no caching.
  • If the key has been seen before AND the body fingerprint matches, the middleware replays the cached response without touching the endpoint.
  • If the key has been seen before but the body is different, it returns 422 to flag the misuse.
  • For new keys, it lets the request proceed, captures the response, and stores it keyed by path:idempotency_key.
  • Save main.py.
  • Check your terminal where the dev server is running. FastAPI auto-reloads on save, so you should see a reload message with no errors.

Seeing an import error on reload?

  • Make sure Request is included in the from fastapi import line alongside FastAPI and HTTPException.
  • Make sure from fastapi.responses import JSONResponse is on its own line below the main FastAPI import.
  • Check that import hashlib and import json appear at the very top of the file.

help me fix import errors in my FastAPI middleware

Understand body fingerprinting and response caching

The middleware uses two mechanisms working together to prevent both duplicate charges and key misuse.

Why fingerprint the request body?

Stripe's official API docs specify that idempotency keys are compared against the original request parameters. If a client reuses a key with different parameters, Stripe rejects the request. This prevents a subtle bug where a retry accidentally creates a completely different payment.

The fingerprint_body function hashes the raw request body with SHA-256. Two identical JSON payloads always produce the same hash. A payload with a different amount, currency, or customer produces a different hash. The middleware compares these fingerprints to catch misuse.

Why scope the cache key to the path?

The cache_key combines the endpoint path with the idempotency key: /payments:abc-123. This means the same idempotency key can be safely used on different endpoints without collision. A key used for POST /payments won't interfere with the same key on POST /webhooks/register.

Test idempotency by sending duplicate requests

The interactive docs at /docs don't make it easy to set custom headers. You'll use curl in a second terminal to test the idempotency protection directly.

  • Open a new terminal window (keep the FastAPI dev server running in the first one).
  • Send a payment request with an Idempotency-Key header by running this command:
curl -X POST http://127.0.0.1:8000/payments \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: test-key-001" \
  -d '{"amount": 5000, "currency": "usd", "customer_id": "cus_123"}'

What should I see?

You should see a JSON response with a payment id (like pay_abc123...), status of pending, and the amount 5000. Note the payment id value for the next test.

  • Now send the exact same request again (same key, same body) to simulate a network retry:
curl -X POST http://127.0.0.1:8000/payments \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: test-key-001" \
  -d '{"amount": 5000, "currency": "usd", "customer_id": "cus_123"}'

You should see the exact same payment id returned. The middleware replayed the cached response instead of creating a duplicate payment.

  • Finally, test the fingerprint protection by reusing the same key with a different amount:
curl -X POST http://127.0.0.1:8000/payments \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: test-key-001" \
  -d '{"amount": 9999, "currency": "usd", "customer_id": "cus_123"}'

You should see a 422 response with the error message indicating the idempotency key was reused with a different request body. This proves your fingerprinting protection works.

Getting a 201 instead of a cached response?

  • Make sure the Idempotency-Key header is spelled exactly right (capital I, capital K, hyphen between words).
  • Confirm the middleware is placed before the # --- API Endpoints --- section. FastAPI registers middleware in order.
  • Check that the dev server reloaded after you saved main.py. Look for a reload message in the server terminal.

help me debug why my idempotency middleware isn't caching responses

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and confirmed idempotency works with your curl tests.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

import hashlib
import json
import random
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str


# --- State Machine ---

VALID_TRANSITIONS: dict[PaymentStatus, list[PaymentStatus]] = {
    PaymentStatus.PENDING: [PaymentStatus.PROCESSING],
    PaymentStatus.PROCESSING: [PaymentStatus.SUCCEEDED, PaymentStatus.FAILED],
    PaymentStatus.SUCCEEDED: [PaymentStatus.REFUNDED],
    PaymentStatus.FAILED: [],
    PaymentStatus.REFUNDED: [],
}


# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}
idempotency_store: dict[str, dict] = {}


# --- Idempotency Middleware ---

def fingerprint_body(body: bytes) -> str:
    return hashlib.sha256(body).hexdigest()


@app.middleware("http")
async def idempotency_middleware(request: Request, call_next):
    if request.method != "POST":
        return await call_next(request)

    idempotency_key = request.headers.get("Idempotency-Key")
    if not idempotency_key:
        return await call_next(request)

    body = await request.body()
    fp = fingerprint_body(body)
    cache_key = f"{request.url.path}:{idempotency_key}"

    if cache_key in idempotency_store:
        cached = idempotency_store[cache_key]
        if cached["fingerprint"] != fp:
            return JSONResponse(
                status_code=422,
                content={"error": "Idempotency key reused with different request body"},
            )
        return JSONResponse(
            status_code=cached["status_code"],
            content=cached["body"],
        )

    response = await call_next(request)

    response_body = b""
    async for chunk in response.body_iterator:
        response_body += chunk

    try:
        body_json = json.loads(response_body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        body_json = response_body.decode()

    idempotency_store[cache_key] = {
        "fingerprint": fp,
        "status_code": response.status_code,
        "body": body_json,
    }

    return JSONResponse(
        status_code=response.status_code,
        content=body_json,
    )


# --- API Endpoints ---

@app.post("/payments", response_model=PaymentResponse, status_code=201)
async def create_payment(payment: PaymentRequest):
    payment_id = f"pay_{uuid.uuid4().hex[:16]}"
    now = datetime.now(timezone.utc).isoformat()

    payment_record = {
        "id": payment_id,
        "status": PaymentStatus.PENDING,
        "amount": payment.amount,
        "currency": payment.currency,
        "customer_id": payment.customer_id,
        "description": payment.description,
        "created_at": now,
    }
    payments_store[payment_id] = payment_record
    return payment_record


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
async def get_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")
    return payments_store[payment_id]


@app.post("/payments/{payment_id}/process", response_model=PaymentResponse)
async def process_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")

    payment = payments_store[payment_id]
    current_status = payment["status"]

    # Validate transition to processing
    if PaymentStatus.PROCESSING not in VALID_TRANSITIONS.get(current_status, []):
        raise HTTPException(
            status_code=409,
            detail=f"Cannot transition from {current_status} to processing",
        )

    # Transition to processing
    payment["status"] = PaymentStatus.PROCESSING

    # Simulate payment provider (80% success rate)
    if random.random() < 0.8:
        payment["status"] = PaymentStatus.SUCCEEDED
    else:
        payment["status"] = PaymentStatus.FAILED

    return payment

Your API now has the same idempotency protection that Stripe uses to prevent double charges. Next up, you'll build a webhook dispatch system so your payment events can notify external services in real time.

Build the Webhook Dispatch System

Your payment API now enforces valid state transitions and prevents duplicate charges. There is one piece missing from a real payment system: notifying other services when something happens.

When a payment succeeds or fails, downstream services (billing, fulfillment, analytics) need to know. Payment systems solve this with webhooks. A webhook is an HTTP callback that fires automatically when an event occurs, pushing data to registered listener URLs.

In this step, get ready to:
  • Define webhook data models and in-memory stores for event tracking.
  • Build a dispatch function with retry logic and wire it into the payment processing flow.
  • Add endpoints to register webhook URLs and inspect the event log.
Define the webhook data models and stores

Before dispatching events, you need models that describe what an event looks like and a place to track registered listener URLs.

  • In main.py, find the existing import block at the top of the file. Add time and httpx to the imports, then add the two new Pydantic models and the two new store lists by copying the following code:
import hashlib
import json
import random
import time
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str


class WebhookEvent(BaseModel):
    id: str
    type: str
    payment_id: str
    data: dict
    created_at: str


class WebhookRegistration(BaseModel):
    url: str


# --- State Machine ---

VALID_TRANSITIONS: dict[PaymentStatus, list[PaymentStatus]] = {
    PaymentStatus.PENDING: [PaymentStatus.PROCESSING],
    PaymentStatus.PROCESSING: [PaymentStatus.SUCCEEDED, PaymentStatus.FAILED],
    PaymentStatus.SUCCEEDED: [PaymentStatus.REFUNDED],
    PaymentStatus.FAILED: [],
    PaymentStatus.REFUNDED: [],
}


# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}
idempotency_store: dict[str, dict] = {}
webhook_urls: list[str] = []
webhook_events: list[dict] = []

What did you just add?

  • import time gives you time.sleep() for implementing backoff delays between retry attempts.
  • import httpx is a modern HTTP client that comes bundled with fastapi[standard]. You will use it to POST event payloads to registered webhook URLs.
  • WebhookEvent defines the shape of an event record: a unique ID, event type (like payment.succeeded), the related payment ID, a data payload, and a timestamp.
  • WebhookRegistration accepts a single url field when a client registers a listener.
  • webhook_urls stores the list of registered listener URLs. webhook_events logs every event that fires, whether delivery succeeded or not.
  • Save main.py.
  • Check your terminal where the dev server is running. It should auto-reload without errors.

Seeing an import error?

If the server shows ModuleNotFoundError for httpx, make sure you installed FastAPI with the standard extras (pip install "fastapi[standard]"). The httpx library is included in that bundle.

help me fix this httpx import error in my FastAPI project

Create the dispatch function with retry logic

Real webhook systems do not give up after a single failed delivery. If the receiving server is temporarily down, the sender retries a few times with increasing delays between attempts. This is called exponential backoff.

  • In main.py, find the comment # --- Idempotency Middleware ---. Directly above it, add a new section for webhook dispatch by copying the following code:
# --- Webhook Dispatch ---

def dispatch_webhook(event_type: str, payment_id: str, data: dict):
    # Build the event record
    event = {
        "id": f"evt_{uuid.uuid4().hex[:16]}",
        "type": event_type,
        "payment_id": payment_id,
        "data": data,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    # Always log the event, even if no URLs are registered
    webhook_events.append(event)

    # Attempt delivery to each registered URL with retries
    for url in webhook_urls:
        for attempt in range(3):
            try:
                with httpx.Client(timeout=5.0) as client:
                    resp = client.post(url, json=event)
                    # Stop retrying if we got a non-server-error response
                    if resp.status_code < 500:
                        break
            except httpx.RequestError:
                pass
            # Exponential backoff: 0.1s, 0.2s, 0.4s
            time.sleep(0.1 * (2**attempt))

Why retry with exponential backoff?

Network requests fail for many reasons: a downstream server might be restarting, a load balancer might be rotating, or there might be a brief connectivity blip. Retrying immediately floods the target while it is still recovering.

Exponential backoff spaces out retries (0.1s, 0.2s, 0.4s) so the target has time to recover. The function also only retries on server errors (status 500+) or connection failures. A 4xx response means the request itself was bad, so retrying would not help.

  • Now wire the dispatch function into the payment processing endpoint. In main.py, scroll down to the process_payment function and replace the entire function with this updated version:
@app.post("/payments/{payment_id}/process", response_model=PaymentResponse)
async def process_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")

    payment = payments_store[payment_id]
    current_status = payment["status"]

    # Validate transition to processing
    if PaymentStatus.PROCESSING not in VALID_TRANSITIONS.get(current_status, []):
        raise HTTPException(
            status_code=409,
            detail=f"Cannot transition from {current_status} to processing",
        )

    # Transition to processing
    payment["status"] = PaymentStatus.PROCESSING
    dispatch_webhook("payment.processing", payment_id, {"status": "processing"})

    # Simulate payment provider (80% success rate)
    if random.random() < 0.8:
        payment["status"] = PaymentStatus.SUCCEEDED
        dispatch_webhook("payment.succeeded", payment_id, {"status": "succeeded"})
    else:
        payment["status"] = PaymentStatus.FAILED
        dispatch_webhook("payment.failed", payment_id, {"status": "failed"})

    return payment

What changed in process_payment?

You added three dispatch_webhook calls that fire at each state transition:

  • payment.processing fires when the payment moves from pending to processing.
  • payment.succeeded fires when the simulated provider returns success.
  • payment.failed fires when the simulated provider returns a failure.

External services (like a fulfillment system) can subscribe to these events and react without polling your API.

  • Save main.py. Your dev server should auto-reload without errors.

Server crashing on reload?

Make sure the dispatch_webhook function is defined above the process_payment endpoint. Python resolves names top-to-bottom, so if dispatch_webhook appears below where it is called, you will get a NameError.

help me fix this NameError for dispatch_webhook in my payment API

Add webhook management endpoints and test the system

You need two endpoints: one for external services to register their listener URL, and one for you to inspect which events have fired.

  • In main.py, scroll to the bottom of the file (after the process_payment function) and add the following two endpoints:
@app.post("/webhooks/register", status_code=201)
async def register_webhook(registration: WebhookRegistration):
    # Add the URL to the list of registered listeners
    webhook_urls.append(registration.url)
    return {"message": "Webhook registered", "url": registration.url}


@app.get("/webhooks/events")
async def list_webhook_events():
    # Return all logged events for inspection
    return webhook_events

What do these endpoints do?

  • POST /webhooks/register accepts a JSON body with a url field and appends it to webhook_urls. Any future payment events will be dispatched to that URL.
  • GET /webhooks/events returns the full event log. This lets you verify that events fired correctly even when the registered URL is unreachable.
  • Save main.py.
  • Open http://127.0.0.1:8000/docs in your browser to test the new endpoints.
  • Expand POST /webhooks/register, click Try it out, paste {"url": "http://localhost:9999/webhook"} into the body, and click Execute.
  • Expand POST /payments, create a new payment with any amount and customer ID, then use POST /payments/{payment_id}/process to process it.
  • Expand GET /webhooks/events and click Execute. You should see at least two events: payment.processing and either payment.succeeded or payment.failed.

Events list is empty?

Make sure you registered a webhook URL before processing the payment. Also confirm that you are calling POST /payments/{payment_id}/process (not just creating a payment). Events only fire during the processing step.

help me debug why my webhook events list is empty after processing a payment

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and your dev server is running without errors.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point. Compare it against yours:

import hashlib
import json
import random
import time
import uuid
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Payment Processing API", version="1.0.0")


# --- Data Models ---

class PaymentStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    REFUNDED = "refunded"


class PaymentRequest(BaseModel):
    amount: int = Field(gt=0, description="Amount in cents")
    currency: str = Field(default="usd", pattern="^[a-z]{3}$")
    customer_id: str
    description: Optional[str] = None


class PaymentResponse(BaseModel):
    id: str
    status: PaymentStatus
    amount: int
    currency: str
    customer_id: str
    description: Optional[str]
    created_at: str


class WebhookEvent(BaseModel):
    id: str
    type: str
    payment_id: str
    data: dict
    created_at: str


class WebhookRegistration(BaseModel):
    url: str


# --- State Machine ---

VALID_TRANSITIONS: dict[PaymentStatus, list[PaymentStatus]] = {
    PaymentStatus.PENDING: [PaymentStatus.PROCESSING],
    PaymentStatus.PROCESSING: [PaymentStatus.SUCCEEDED, PaymentStatus.FAILED],
    PaymentStatus.SUCCEEDED: [PaymentStatus.REFUNDED],
    PaymentStatus.FAILED: [],
    PaymentStatus.REFUNDED: [],
}


# --- In-Memory Stores ---

payments_store: dict[str, dict] = {}
idempotency_store: dict[str, dict] = {}
webhook_urls: list[str] = []
webhook_events: list[dict] = []


# --- Idempotency Middleware ---

def fingerprint_body(body: bytes) -> str:
    return hashlib.sha256(body).hexdigest()


@app.middleware("http")
async def idempotency_middleware(request: Request, call_next):
    if request.method != "POST":
        return await call_next(request)

    idempotency_key = request.headers.get("Idempotency-Key")
    if not idempotency_key:
        return await call_next(request)

    body = await request.body()
    fp = fingerprint_body(body)
    cache_key = f"{request.url.path}:{idempotency_key}"

    if cache_key in idempotency_store:
        cached = idempotency_store[cache_key]
        if cached["fingerprint"] != fp:
            return JSONResponse(
                status_code=422,
                content={"error": "Idempotency key reused with different request body"},
            )
        return JSONResponse(
            status_code=cached["status_code"],
            content=cached["body"],
        )

    response = await call_next(request)

    response_body = b""
    async for chunk in response.body_iterator:
        response_body += chunk

    try:
        body_json = json.loads(response_body)
    except (json.JSONDecodeError, UnicodeDecodeError):
        body_json = response_body.decode()

    idempotency_store[cache_key] = {
        "fingerprint": fp,
        "status_code": response.status_code,
        "body": body_json,
    }

    return JSONResponse(
        status_code=response.status_code,
        content=body_json,
    )


# --- Webhook Dispatch ---

def dispatch_webhook(event_type: str, payment_id: str, data: dict):
    event = {
        "id": f"evt_{uuid.uuid4().hex[:16]}",
        "type": event_type,
        "payment_id": payment_id,
        "data": data,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    webhook_events.append(event)

    for url in webhook_urls:
        for attempt in range(3):
            try:
                with httpx.Client(timeout=5.0) as client:
                    resp = client.post(url, json=event)
                    if resp.status_code < 500:
                        break
            except httpx.RequestError:
                pass
            time.sleep(0.1 * (2**attempt))


# --- API Endpoints ---

@app.post("/payments", response_model=PaymentResponse, status_code=201)
async def create_payment(payment: PaymentRequest):
    payment_id = f"pay_{uuid.uuid4().hex[:16]}"
    now = datetime.now(timezone.utc).isoformat()

    payment_record = {
        "id": payment_id,
        "status": PaymentStatus.PENDING,
        "amount": payment.amount,
        "currency": payment.currency,
        "customer_id": payment.customer_id,
        "description": payment.description,
        "created_at": now,
    }
    payments_store[payment_id] = payment_record
    return payment_record


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
async def get_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")
    return payments_store[payment_id]


@app.post("/payments/{payment_id}/process", response_model=PaymentResponse)
async def process_payment(payment_id: str):
    if payment_id not in payments_store:
        raise HTTPException(status_code=404, detail="Payment not found")

    payment = payments_store[payment_id]
    current_status = payment["status"]

    # Validate transition to processing
    if PaymentStatus.PROCESSING not in VALID_TRANSITIONS.get(current_status, []):
        raise HTTPException(
            status_code=409,
            detail=f"Cannot transition from {current_status} to processing",
        )

    # Transition to processing
    payment["status"] = PaymentStatus.PROCESSING
    dispatch_webhook("payment.processing", payment_id, {"status": "processing"})

    # Simulate payment provider (80% success rate)
    if random.random() < 0.8:
        payment["status"] = PaymentStatus.SUCCEEDED
        dispatch_webhook("payment.succeeded", payment_id, {"status": "succeeded"})
    else:
        payment["status"] = PaymentStatus.FAILED
        dispatch_webhook("payment.failed", payment_id, {"status": "failed"})

    return payment


@app.post("/webhooks/register", status_code=201)
async def register_webhook(registration: WebhookRegistration):
    webhook_urls.append(registration.url)
    return {"message": "Webhook registered", "url": registration.url}


@app.get("/webhooks/events")
async def list_webhook_events():
    return webhook_events

Your payment API now fires events on every state transition and delivers them to registered listeners with automatic retries. Next, you will write a test script that simulates failures and proves your entire system handles them correctly.

Simulate Failures and Prove Correctness

Your payment API has idempotency protection, a state machine, and webhook dispatch. But how do you prove these systems actually work under failure conditions?

In a system design interview, stating that your idempotency layer prevents double charges is not enough. You need to demonstrate it. In this step, you will write a test script that simulates network retries, invalid transitions, and mismatched requests to verify your system handles each failure correctly.

In this step, get ready to:
  • Create a test script that proves idempotency prevents duplicate payments.
  • Simulate invalid state transitions and body mismatches.
  • Practice articulating your design decisions in interview style.
Create the test script with payment creation and idempotency retries

The test script acts as a simulated client that hits your running API. It uses httpx (already installed with FastAPI) to send HTTP requests and Python assert statements to verify responses.

  • Create a new file called test_client.py in the same folder as main.py.
  • Add the imports, base URL, and the first test function by pasting this code:
import httpx
import uuid

BASE_URL = "http://127.0.0.1:8000"


def test_create_payment():
    """Test basic payment creation."""
    response = httpx.post(
        f"{BASE_URL}/payments",
        json={
            "amount": 5000,
            "currency": "usd",
            "customer_id": "cus_123",
            "description": "Test payment",
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["status"] == "pending"
    assert data["amount"] == 5000
    print(f"✅ Created payment: {data['id']}")
    return data["id"]

What does this code do?

This sets up the test infrastructure. BASE_URL points at your running server.

The test_create_payment function creates a $50.00 payment (5000 cents) and asserts the response comes back with status 201 and a pending state. It returns the payment ID so other tests can reuse it.

  • Next, add the idempotency retry test below test_create_payment:

def test_idempotency_prevents_duplicates():
    """Test that the same idempotency key returns the same result."""
    idempotency_key = str(uuid.uuid4())
    payload = {
        "amount": 3000,
        "currency": "usd",
        "customer_id": "cus_456",
        "description": "Idempotent payment",
    }

    # First request
    resp1 = httpx.post(
        f"{BASE_URL}/payments",
        json=payload,
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp1.status_code == 201

    # Retry with same key (simulating network retry)
    resp2 = httpx.post(
        f"{BASE_URL}/payments",
        json=payload,
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp2.status_code == 201

    # Both should return the exact same payment
    assert resp1.json()["id"] == resp2.json()["id"]
    print(f"✅ Idempotency works: same key returned same payment {resp1.json()['id']}")

What does this code do?

This simulates what happens when a client retries a payment request after a network timeout. Both requests use the same Idempotency-Key header and identical body.

The critical assertion is that both responses return the same payment ID. Without idempotency protection, the second request would create a duplicate charge.

  • Save test_client.py.

Seeing an import error for httpx?

The httpx library is included when you install fastapi[standard]. If you see an import error, make sure your virtual environment is activated (you should see (venv) in your terminal prompt).

Help me fix my httpx import error in test_client.py.

Add test scenarios for failure conditions

The real value of these tests is proving your system rejects bad requests. A payment system that only works on the happy path is dangerous.

  • Add the body mismatch test below test_idempotency_prevents_duplicates:

def test_idempotency_rejects_different_body():
    """Test that reusing a key with a different body returns 422."""
    idempotency_key = str(uuid.uuid4())

    # First request
    resp1 = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 1000, "currency": "usd", "customer_id": "cus_789"},
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp1.status_code == 201

    # Same key, different body
    resp2 = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 2000, "currency": "usd", "customer_id": "cus_789"},
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp2.status_code == 422
    print("✅ Idempotency rejects different body with same key (422)")

Why is this test important?

This catches a subtle attack or bug: a client reusing the same idempotency key but changing the payment amount. Your body fingerprinting detects the mismatch and returns a 422 error instead of silently processing the wrong amount.

  • Add the state machine tests below test_idempotency_rejects_different_body:

def test_state_machine_valid_transitions():
    """Test that processing works for pending payments."""
    payment_id = test_create_payment()

    response = httpx.post(f"{BASE_URL}/payments/{payment_id}/process")
    assert response.status_code == 200
    status = response.json()["status"]
    assert status in ("succeeded", "failed")
    print(f"✅ Payment processed: {status}")
    return payment_id, status


def test_state_machine_invalid_transition():
    """Test that re-processing a completed payment returns 409."""
    payment_id, status = test_state_machine_valid_transitions()

    # Try to process again
    response = httpx.post(f"{BASE_URL}/payments/{payment_id}/process")
    assert response.status_code == 409
    print(f"✅ Invalid transition rejected (409): cannot process a {status} payment")

What does this code do?

The first function creates a payment and processes it. Since the simulated provider has an 80% success rate, the final status will be either succeeded or failed. Both are valid terminal states.

The second function then tries to process the same payment again. The state machine rejects this with a 409 Conflict because neither succeeded nor failed can transition back to processing. This prevents double-processing of the same payment.

  • Add the webhook events test below test_state_machine_invalid_transition:

def test_webhook_events_logged():
    """Test that processing a payment generates webhook events."""
    # Register a webhook (will fail to deliver but events are still logged)
    httpx.post(
        f"{BASE_URL}/webhooks/register",
        json={"url": "http://localhost:9999/webhook"},
    )

    # Create and process a payment
    resp = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 7500, "currency": "usd", "customer_id": "cus_webhook"},
    )
    payment_id = resp.json()["id"]
    httpx.post(f"{BASE_URL}/payments/{payment_id}/process")

    # Check events
    events_resp = httpx.get(f"{BASE_URL}/webhooks/events")
    events = events_resp.json()
    payment_events = [e for e in events if e["payment_id"] == payment_id]
    assert len(payment_events) >= 2  # processing + succeeded/failed
    print(f"✅ Webhook events logged: {[e['type'] for e in payment_events]}")

What does this code do?

This registers a fake webhook URL (nothing is actually listening on port 9999). The delivery will fail, but the events are still logged in memory.

After creating and processing a payment, the test checks that at least 2 events were logged: one for payment.processing and one for either payment.succeeded or payment.failed.

  • Save test_client.py.

Unsure where to place each function?

Each test function goes directly after the previous one in the same file. There is no class or test framework wrapping them. Use the full-code tab at the end of this step to double check your file structure.

Help me check my test_client.py structure.

Run the tests and practice articulating your design decisions

The final piece is tying all the tests together with a runner, executing them, and practicing how you would explain these design decisions in an interview.

  • Add the test runner block at the bottom of test_client.py:

if __name__ == "__main__":
    print("\n🧪 Running Payment API Tests\n" + "=" * 40)
    test_create_payment()
    test_idempotency_prevents_duplicates()
    test_idempotency_rejects_different_body()
    test_state_machine_valid_transitions()
    test_state_machine_invalid_transition()
    test_webhook_events_logged()
    print("\n" + "=" * 40)
    print("🎉 All tests passed!")

What does this code do?

The if __name__ == "__main__": block runs all test functions in sequence when you execute the file directly. If any assert statement fails, the script stops immediately with an error showing which assertion broke.

  • Save test_client.py.
  • Make sure your FastAPI server is still running in your other terminal (the one running fastapi dev main.py).
  • Open a new terminal tab and activate your virtual environment by running:
source venv/bin/activate
  • Run your test script by executing:
python test_client.py
  • You should see all six tests pass with checkmark outputs and a final "All tests passed" message.

Seeing a ConnectionRefusedError?

This means your FastAPI server is not running. Switch back to your other terminal and start it with fastapi dev main.py. Then switch back and run the test script again.

Help me fix my test script connection error.

Why practice explaining this aloud?

In a system design interview (especially at companies like Stripe), you are expected to articulate WHY your design decisions exist, not just what they are. Practice saying these three things out loud:

  • "Idempotency keys prevent double charges because the server caches the response of the first request and replays it for retries with the same key."
  • "The state machine prevents invalid transitions because each state defines its allowed next states. A succeeded payment cannot go back to processing."
  • "At scale, I would swap the in-memory dictionaries for Redis (idempotency cache with TTL) and PostgreSQL (durable payment records with row-level locking)."

✔️ Awesome, I've got everything!

Great. Double check you have saved both files.

ⓧ I'd like to double check the full code

Here is the complete test_client.py file:

import httpx
import uuid

BASE_URL = "http://127.0.0.1:8000"


def test_create_payment():
    """Test basic payment creation."""
    response = httpx.post(
        f"{BASE_URL}/payments",
        json={
            "amount": 5000,
            "currency": "usd",
            "customer_id": "cus_123",
            "description": "Test payment",
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["status"] == "pending"
    assert data["amount"] == 5000
    print(f"✅ Created payment: {data['id']}")
    return data["id"]


def test_idempotency_prevents_duplicates():
    """Test that the same idempotency key returns the same result."""
    idempotency_key = str(uuid.uuid4())
    payload = {
        "amount": 3000,
        "currency": "usd",
        "customer_id": "cus_456",
        "description": "Idempotent payment",
    }

    # First request
    resp1 = httpx.post(
        f"{BASE_URL}/payments",
        json=payload,
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp1.status_code == 201

    # Retry with same key (simulating network retry)
    resp2 = httpx.post(
        f"{BASE_URL}/payments",
        json=payload,
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp2.status_code == 201

    # Both should return the exact same payment
    assert resp1.json()["id"] == resp2.json()["id"]
    print(f"✅ Idempotency works: same key returned same payment {resp1.json()['id']}")


def test_idempotency_rejects_different_body():
    """Test that reusing a key with a different body returns 422."""
    idempotency_key = str(uuid.uuid4())

    # First request
    resp1 = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 1000, "currency": "usd", "customer_id": "cus_789"},
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp1.status_code == 201

    # Same key, different body
    resp2 = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 2000, "currency": "usd", "customer_id": "cus_789"},
        headers={"Idempotency-Key": idempotency_key},
    )
    assert resp2.status_code == 422
    print("✅ Idempotency rejects different body with same key (422)")


def test_state_machine_valid_transitions():
    """Test that processing works for pending payments."""
    payment_id = test_create_payment()

    response = httpx.post(f"{BASE_URL}/payments/{payment_id}/process")
    assert response.status_code == 200
    status = response.json()["status"]
    assert status in ("succeeded", "failed")
    print(f"✅ Payment processed: {status}")
    return payment_id, status


def test_state_machine_invalid_transition():
    """Test that re-processing a completed payment returns 409."""
    payment_id, status = test_state_machine_valid_transitions()

    # Try to process again
    response = httpx.post(f"{BASE_URL}/payments/{payment_id}/process")
    assert response.status_code == 409
    print(f"✅ Invalid transition rejected (409): cannot process a {status} payment")


def test_webhook_events_logged():
    """Test that processing a payment generates webhook events."""
    # Register a webhook (will fail to deliver but events are still logged)
    httpx.post(
        f"{BASE_URL}/webhooks/register",
        json={"url": "http://localhost:9999/webhook"},
    )

    # Create and process a payment
    resp = httpx.post(
        f"{BASE_URL}/payments",
        json={"amount": 7500, "currency": "usd", "customer_id": "cus_webhook"},
    )
    payment_id = resp.json()["id"]
    httpx.post(f"{BASE_URL}/payments/{payment_id}/process")

    # Check events
    events_resp = httpx.get(f"{BASE_URL}/webhooks/events")
    events = events_resp.json()
    payment_events = [e for e in events if e["payment_id"] == payment_id]
    assert len(payment_events) >= 2  # processing + succeeded/failed
    print(f"✅ Webhook events logged: {[e['type'] for e in payment_events]}")


if __name__ == "__main__":
    print("\n🧪 Running Payment API Tests\n" + "=" * 40)
    test_create_payment()
    test_idempotency_prevents_duplicates()
    test_idempotency_rejects_different_body()
    test_state_machine_valid_transitions()
    test_state_machine_invalid_transition()
    test_webhook_events_logged()
    print("\n" + "=" * 40)
    print("🎉 All tests passed!")

Your main.py should remain unchanged from the previous step.

You now have a provably correct payment API. Every reliability pattern (idempotency, state machine, webhooks) is backed by automated assertions. Next up, the Secret Mission challenges you to add a circuit breaker pattern for cascading failure protection.

Secret mission

Add a Circuit Breaker Pattern

Your payment API handles retries and state transitions, but what happens when the payment provider itself goes down? Right now, every request still attempts to reach the provider, piling up failures and wasting resources. Add a circuit breaker that detects consecutive failures, stops sending doomed requests, and self-heals once the provider recovers.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • FastAPI dev server (running process in your terminal).
  • Python virtual environment (venv folder inside your project directory).
  • Project files (main.py and test_client.py).

Keep everything running

No action needed. Choose this if you're still actively building or want to keep experimenting with your payment API.

  • Your FastAPI dev server stays running in the terminal.
  • Your virtual environment stays active.
  • Your project files remain on your Desktop.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files stay intact so you can pick up where you left off.

  • Switch to the terminal running your FastAPI server and press Ctrl+C to stop it.
  • Deactivate your virtual environment by running:
deactivate

To resume later, navigate back to your project folder, activate the virtual environment with source venv/bin/activate, and start the server again with fastapi dev main.py.

Delete - I don't want to use this again

Remove all project files and start fresh. This permanently deletes your code.

  • Switch to the terminal running your FastAPI server and press Ctrl+C to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the project folder.

On macOS/Linux:

cd ~/Desktop
rm -rf payment-api

On Windows:

cd ~\Desktop
rmdir /s /q payment-api

Replace with your folder name

If you named your project folder something different, replace payment-api with whatever you called it.

Nice Work!

Nice Work!

Nice work! You just built a payment processing API that implements the exact reliability patterns used by real payment systems like Stripe.

You've learned how to:

  • Build an API-first payment service with FastAPI that uses idempotency keys to prevent double charges on retried requests.
  • Create a state machine that enforces valid payment lifecycle transitions and returns 409 Conflict for invalid ones.
  • Implement a webhook dispatch system with exponential backoff retry logic and write test scenarios that prove your system handles failures correctly.
  • Add a circuit breaker pattern to protect against cascading failures from an unreliable payment provider.

Ready to quiz yourself?