Payment Processing API with Stripe

Build a FastAPI payment service with idempotency, webhooks, and state tracking.

Introduction

30 Second Summary

Every time you tap your card or click "Pay now" online, a complex chain of events fires behind the scenes. Your payment gets created, verified, confirmed, and tracked through multiple states before the money moves.

In this project, you will build a payment processing API using Python and Stripe that handles the full payment lifecycle. You'll create endpoints that process payments, prevent duplicate charges, and receive asynchronous event notifications.

What You'll Build

A backend payment processing service built with FastAPI and Stripe that creates payments, enforces idempotency, tracks state transitions in a local SQLite database, and processes webhook events.

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

  • A FastAPI application with payment endpoints that create, retrieve, and list payment transactions.
  • An idempotency layer that prevents duplicate charges by detecting repeated requests.
  • A webhook endpoint with Stripe signature verification that processes asynchronous payment events and updates your database.
  • Secret Mission: Add a refund endpoint that handles full and partial refunds with proper state tracking.

Are there any prerequisites?

You should be comfortable with basic Python (variables, functions, and classes) and have some familiarity with using a terminal. No prior experience with payment systems or APIs is required.

Before We Start

Before you can build a payment processing API, you need the right tools installed and accounts ready.

In this step, you will install Python, set up a virtual environment with the required packages, create a free Stripe account, and install the Stripe CLI for local testing.

In this step, get ready to:
  • Install Python and set up a virtual environment with project dependencies.
  • Create a free Stripe account and locate your test API keys.
  • Install and authenticate the Stripe CLI.
Install Python and create a virtual environment

This project requires Python 3.10 or higher. Let's check whether you have it installed.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), then type Terminal and press Enter.
  • Check your Python version by running this command:
python3 --version

Windows users

On Windows, use python instead of python3 for all commands in this project. The same applies to pip instead of pip3.

✔️ I see version 3.10 or higher

You're all set. Move on to creating your project folder below.

ⓧ I see an older version

You have Python installed, but this project needs 3.10 or higher. Download the latest version from the official Python website:

  • Visit python.org/downloads and download the latest installer for your operating system.
  • Run the installer. On Windows, make sure to check Add Python to PATH before clicking Install Now.
  • Close and reopen your terminal, then run the version check command again to confirm.

ⓧ Command not found

Python is not installed yet. Download it from the official website:

  • Visit python.org/downloads and download the latest installer for your operating system.
  • Run the installer. On Windows, make sure to check Add Python to PATH before clicking Install Now.
  • Close and reopen your terminal, then run the version check command again to confirm you see 3.10 or higher.

Now let's create a project folder, set up a virtual environment, and install the packages you need.

  • 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
  • Create a virtual environment by running:
python3 -m venv .venv
  • Activate the virtual environment by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

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

What is a virtual environment?

A virtual environment is an isolated Python setup that keeps your project's packages separate from your system Python. This prevents version conflicts between projects.

  • Create a requirements.txt file that lists your project dependencies by running:
echo 'fastapi[standard]==0.136.3
stripe==15.2.0' > requirements.txt
  • Install the dependencies by running:
pip install -r requirements.txt

This installs FastAPI (the web framework for your API) and the Stripe Python SDK (for communicating with Stripe's payment services).

Installation failed?

Make sure you see (.venv) in your terminal prompt. If not, rerun the activate command above. Also confirm you are using pip (Windows) or pip3 (macOS) depending on your setup.

Help me troubleshoot my pip install failing inside a virtual environment.

  • Confirm the packages installed correctly by running:
pip show fastapi stripe

You should see version 0.136.3 for fastapi and 15.2.0 for stripe in the output.

✔️ Awesome, I've got everything!

Great. Double check you have requirements.txt saved in your payment-api folder.

ⓧ I'd like to double check the full code

Your requirements.txt should contain exactly this:

fastapi[standard]==0.136.3
stripe==15.2.0
Create a Stripe account and find your test API keys

Stripe provides a free sandbox environment for testing payments. No credit card is required to sign up, and no real charges are made in test mode.

  • Open your browser and go to dashboard.stripe.com/register.
  • Fill in your email, full name, and password, then click Create account.
  • Verify your email address when prompted.
  • Once signed in, confirm you are in the sandbox by looking at the top of the Dashboard. You should see a Test mode toggle or a Sandbox label.

Now find your test API keys.

  • In the Stripe Dashboard, click Developers in the left sidebar.
  • Click API keys.
  • You will see two keys listed:
  • Copy your secret key and save it somewhere safe (a notes app or password manager). You will use it in the next step when writing code.

What are these keys?

The publishable key identifies your account on the frontend. The secret key authenticates server-side API calls to Stripe. Because test keys start with sk_test_, they only interact with Stripe's sandbox. No real money moves.

Don't see API keys?

Make sure you are viewing the sandbox/test environment, not live mode. Look for a toggle or tab at the top of the Stripe Dashboard that says Test mode or Sandbox.

Help me find my Stripe test API keys in the Dashboard.

Install and authenticate the Stripe CLI

The Stripe CLI lets you test webhooks locally by forwarding Stripe events to your development server. You will need it for Step 4.

  • Install the Stripe CLI using the instructions for your operating system:

macOS

The easiest way to install on macOS is with Homebrew. Run this command:

brew install stripe/stripe-cli/stripe

Don't have Homebrew?

If you don't have Homebrew installed, visit brew.sh and follow the one-line install command. Then rerun the command above.

Alternatively, download the macOS archive directly from the Stripe CLI GitHub releases page, extract it, and move the stripe binary to /usr/local/bin.

Windows

Download the Stripe CLI directly from the GitHub releases page:

  • Visit the Stripe CLI GitHub releases page and download the latest windows zip file.
  • Unzip the file.
  • Add the folder containing stripe.exe to your Path environment variable.
  • Close and reopen your terminal for the change to take effect.

Windows antivirus warning

Windows antivirus scanners occasionally flag the Stripe CLI as unsafe. This is a known false positive. If your antivirus blocks the download, allow the file to proceed.

  • Verify the Stripe CLI is installed by running:
stripe --version

You should see a version number printed in the terminal.

  • Authenticate the CLI with your Stripe account by running:
stripe login

What happens during login?

This command opens your browser and asks you to confirm that you want to give the CLI access to your Stripe account. Press Enter in the terminal when prompted, then click Allow access in the browser window that opens.

Once authenticated, you will see a confirmation message in your terminal.

Stripe command not found after install?

On macOS, make sure Homebrew's bin directory is in your PATH. Try opening a new terminal tab and running the command again. On Windows, confirm you added the folder containing stripe.exe to your Path environment variable and reopened the terminal.

Help me fix the stripe command not being found after installation.

Your development environment is ready. Next up, you will build the database layer and data models that track payments through their lifecycle.

Build the Payment Database and Models

Your environment is ready and Stripe is connected. Now you need a way to persist payment data locally so your API can track every payment through its lifecycle.

Payment systems must record every state change a transaction goes through. In this step, you will create a SQLite database layer that stores payment records and a set of Pydantic models that validate incoming requests and shape outgoing responses.

In this step, get ready to:
  • Create the project files and set up a SQLite database with a payments table.
  • Define Pydantic request and response models for the API.
  • Write helper functions to create, update, and retrieve payment records.
Create the project files and database schema

Your payment API will be split across three files. Each file has a single responsibility: database operations, data models, and the API endpoints.

  • Create the three project files by running this command:
touch database.py models.py main.py
  • Confirm the files exist by running:
ls

You should see database.py, models.py, main.py, and requirements.txt in the output.

Why three separate files?

Separating database logic, data models, and API routes keeps each file focused on one job. This makes the codebase easier to test and modify as it grows.

Now you will write the database setup code. This creates the connection helper and the payments table schema.

  • Open database.py in Cursor and add the following code:
import sqlite3
from datetime import datetime, timezone


# Database file stored in the project directory
DB_NAME = "payments.db"


def get_connection():
    # Connect to SQLite and enable dict-like row access
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    # Create the payments table if it doesn't already exist
    conn = get_connection()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS payments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            stripe_payment_intent_id TEXT,
            amount INTEGER NOT NULL,
            currency TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            idempotency_key TEXT UNIQUE,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)
    conn.commit()
    conn.close()

What does this code do?

The get_connection() function opens a connection to payments.db and sets row_factory = sqlite3.Row so query results behave like dictionaries. You can access columns by name instead of by index.

The init_db() function creates the payments table with columns for tracking Stripe IDs, amounts, status, and timestamps. The idempotency_key column has a UNIQUE constraint to prevent duplicate payment records at the database level.

  • Save database.py.
  • Test that the database initializes correctly by running:
python3 -c "from database import init_db; init_db(); print('Database created!')"

You should see Database created! printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is still active. You should see the environment name in parentheses at the start of your terminal prompt. If it is missing, reactivate it.

Also confirm you are running the command from the same folder that contains database.py.

Help me debug a ModuleNotFoundError when importing database

Define the Pydantic request and response models

Your API needs structured data for incoming payment requests and outgoing responses. Pydantic models handle validation automatically, rejecting malformed requests before they reach your business logic.

  • Open models.py and add the following code:
from pydantic import BaseModel


# Request model for creating a new payment
class CreatePaymentRequest(BaseModel):
    amount: int
    currency: str = "usd"
    idempotency_key: str


# Response model returned by all payment endpoints
class PaymentResponse(BaseModel):
    id: int
    stripe_payment_intent_id: str | None
    amount: int
    currency: str
    status: str
    idempotency_key: str | None
    created_at: str
    updated_at: str
    client_secret: str | None = None


# Request model for refunding a payment (used in Secret Mission)
class RefundRequest(BaseModel):
    amount: int | None = None

What do these models do?

Each class defines a strict data shape. CreatePaymentRequest requires an amount (in cents) and an idempotency_key that callers must provide. The currency defaults to "usd" if omitted.

FastAPI uses these models to auto-validate request bodies and auto-generate API documentation. If a caller sends a request missing the required amount field, FastAPI returns a 422 error automatically.

  • Save models.py.

You won't see a change yet. These models are used in the next step when you build the API endpoints.

Write the database helper functions

Your API endpoints will need to create new payment records, look them up by different keys, and update their status as payments progress through the lifecycle. Each helper function handles one database operation.

  • Return to database.py and add the create_payment function below init_db():
def create_payment(amount: int, currency: str, idempotency_key: str, stripe_payment_intent_id: str, status: str):
    conn = get_connection()
    # Record the current UTC time for both timestamps
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        """INSERT INTO payments (amount, currency, idempotency_key, stripe_payment_intent_id, status, created_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (amount, currency, idempotency_key, stripe_payment_intent_id, status, now, now)
    )
    # Get the auto-generated ID of the new row
    payment_id = cursor.lastrowid
    conn.commit()
    # Fetch and return the complete payment record
    payment = conn.execute("SELECT * FROM payments WHERE id = ?", (payment_id,)).fetchone()
    conn.close()
    return dict(payment)

What does this code do?

This function inserts a new payment row into the database with all the provided fields plus UTC timestamps. It then fetches the complete row back (including the auto-generated id) and returns it as a dictionary. The ? placeholders prevent SQL injection by parameterizing the query.

  • Add the retrieval and update functions below create_payment:
def get_payment_by_idempotency_key(idempotency_key: str):
    # Look up a payment by its unique idempotency key
    conn = get_connection()
    payment = conn.execute(
        "SELECT * FROM payments WHERE idempotency_key = ?", (idempotency_key,)
    ).fetchone()
    conn.close()
    return dict(payment) if payment else None


def get_payment_by_id(payment_id: int):
    # Look up a payment by its database ID
    conn = get_connection()
    payment = conn.execute("SELECT * FROM payments WHERE id = ?", (payment_id,)).fetchone()
    conn.close()
    return dict(payment) if payment else None


def get_payment_by_stripe_id(stripe_payment_intent_id: str):
    # Look up a payment by the Stripe PaymentIntent ID
    conn = get_connection()
    payment = conn.execute(
        "SELECT * FROM payments WHERE stripe_payment_intent_id = ?", (stripe_payment_intent_id,)
    ).fetchone()
    conn.close()
    return dict(payment) if payment else None


def update_payment_status(payment_id: int, status: str):
    # Update a payment's status and refresh the updated_at timestamp
    conn = get_connection()
    now = datetime.now(timezone.utc).isoformat()
    conn.execute(
        "UPDATE payments SET status = ?, updated_at = ? WHERE id = ?",
        (status, now, payment_id)
    )
    conn.commit()
    conn.close()


def list_payments():
    # Return all payments, newest first
    conn = get_connection()
    payments = conn.execute("SELECT * FROM payments ORDER BY created_at DESC").fetchall()
    conn.close()
    return [dict(p) for p in payments]

What do these functions do?

You now have five retrieval and mutation helpers. The three get_payment_by_* functions each query by a different column. Your webhook handler will use get_payment_by_stripe_id to match incoming Stripe events to your local records. The idempotency endpoint will use get_payment_by_idempotency_key to detect duplicates.

Each getter returns None when no matching row exists, making it easy to branch in your API logic.

  • Save database.py.
  • Verify the full database module works by running this quick integration test:
python3 -c "
from database import init_db, create_payment, get_payment_by_id
init_db()
payment = create_payment(1000, 'usd', 'test-key-1', 'pi_test123', 'pending')
print(f'Created payment with ID: {payment["id"]}')
looked_up = get_payment_by_id(payment['id'])
print(f'Retrieved payment status: {looked_up["status"]}')
"

You should see output confirming the payment was created and retrieved successfully.

Seeing an OperationalError about the table?

If you see an error mentioning no such table: payments, make sure your init_db() call comes before create_payment() in the test script. The init_db function creates the table.

Help me fix an OperationalError with my payments database

  • Clean up the test database so you start fresh when the API runs by running:
rm payments.db

✔️ Awesome, I've got everything!

Great. Double check you have saved both database.py and models.py.

ⓧ I'd like to double check the full code

Here is the complete database.py:

import sqlite3
from datetime import datetime, timezone


DB_NAME = "payments.db"


def get_connection():
    conn = sqlite3.connect(DB_NAME)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_connection()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS payments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            stripe_payment_intent_id TEXT,
            amount INTEGER NOT NULL,
            currency TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            idempotency_key TEXT UNIQUE,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)
    conn.commit()
    conn.close()


def create_payment(amount: int, currency: str, idempotency_key: str, stripe_payment_intent_id: str, status: str):
    conn = get_connection()
    now = datetime.now(timezone.utc).isoformat()
    cursor = conn.execute(
        """INSERT INTO payments (amount, currency, idempotency_key, stripe_payment_intent_id, status, created_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?)""",
        (amount, currency, idempotency_key, stripe_payment_intent_id, status, now, now)
    )
    payment_id = cursor.lastrowid
    conn.commit()
    payment = conn.execute("SELECT * FROM payments WHERE id = ?", (payment_id,)).fetchone()
    conn.close()
    return dict(payment)


def get_payment_by_idempotency_key(idempotency_key: str):
    conn = get_connection()
    payment = conn.execute(
        "SELECT * FROM payments WHERE idempotency_key = ?", (idempotency_key,)
    ).fetchone()
    conn.close()
    return dict(payment) if payment else None


def get_payment_by_id(payment_id: int):
    conn = get_connection()
    payment = conn.execute("SELECT * FROM payments WHERE id = ?", (payment_id,)).fetchone()
    conn.close()
    return dict(payment) if payment else None


def get_payment_by_stripe_id(stripe_payment_intent_id: str):
    conn = get_connection()
    payment = conn.execute(
        "SELECT * FROM payments WHERE stripe_payment_intent_id = ?", (stripe_payment_intent_id,)
    ).fetchone()
    conn.close()
    return dict(payment) if payment else None


def update_payment_status(payment_id: int, status: str):
    conn = get_connection()
    now = datetime.now(timezone.utc).isoformat()
    conn.execute(
        "UPDATE payments SET status = ?, updated_at = ? WHERE id = ?",
        (status, now, payment_id)
    )
    conn.commit()
    conn.close()


def list_payments():
    conn = get_connection()
    payments = conn.execute("SELECT * FROM payments ORDER BY created_at DESC").fetchall()
    conn.close()
    return [dict(p) for p in payments]

And here is the complete models.py:

from pydantic import BaseModel


class CreatePaymentRequest(BaseModel):
    amount: int
    currency: str = "usd"
    idempotency_key: str


class PaymentResponse(BaseModel):
    id: int
    stripe_payment_intent_id: str | None
    amount: int
    currency: str
    status: str
    idempotency_key: str | None
    created_at: str
    updated_at: str
    client_secret: str | None = None


class RefundRequest(BaseModel):
    amount: int | None = None

Your data layer is complete. Next up, you will wire these database functions and models into live API endpoints that talk to Stripe.

Create Payment Endpoints with Idempotency

Your database layer and Pydantic models are ready to store and validate payment data.

Now it's time to build the API endpoints that tie everything together. In this step, you'll create a POST endpoint that prevents duplicate charges using idempotency keys, plus GET endpoints for retrieving payment records.

In this step, get ready to:
  • Build a POST endpoint that creates payments with idempotency protection.
  • Build GET endpoints to retrieve individual and all payment records.
  • Test the API using FastAPI's auto-generated interactive docs.
Set up main.py with app configuration

Your API needs a central file that initializes the FastAPI app, connects to Stripe, and wires up the database. The lifespan pattern ensures your database table gets created before the app starts handling requests.

  • Create a new file called main.py in your project folder.
  • Add the following code to set up the app configuration:
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, HTTPException
from stripe import StripeClient

from database import (
    init_db,
    create_payment,
    get_payment_by_idempotency_key,
    get_payment_by_id,
    get_payment_by_stripe_id,
    update_payment_status,
    list_payments,
)
from models import CreatePaymentRequest, PaymentResponse, RefundRequest


# Load Stripe keys from environment variables with fallback placeholders
STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY", "sk_test_your-key-here")
STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_your-webhook-secret-here")

# Initialize the Stripe client with your secret key
client = StripeClient(STRIPE_SECRET_KEY)


# Run init_db() once before the app starts accepting requests
@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(title="Payment Processing API", lifespan=lifespan)

What does this code do?

  • os.environ.get() reads your Stripe keys from environment variables. The fallback strings are just placeholders for when the variable isn't set.
  • StripeClient(STRIPE_SECRET_KEY) creates a client object you'll use to call the Stripe API.
  • The lifespan function is an async context manager that runs init_db() before the app starts serving requests. This creates the payments table in SQLite automatically.
  • FastAPI(title=..., lifespan=...) creates the app instance with a name that shows up in the auto-generated docs.
  • Save main.py.

Why environment variables for keys?

Storing secrets in environment variables keeps them out of your source code. If you accidentally push your code to GitHub, the keys won't be exposed. You'll set the actual environment variable when you start the server later in this step.

Build the POST /payments endpoint with idempotency

In real payment systems, network failures can cause a client to retry a request. Without protection, that retry creates a second charge. Idempotency keys solve this by letting the server recognize a repeated request and return the original result instead of processing it again.

  • In main.py, add the following endpoint below the app = FastAPI(...) line:
@app.post("/payments", response_model=PaymentResponse)
def create_payment_endpoint(payment_request: CreatePaymentRequest):
    # Check if a payment with this idempotency key already exists
    existing = get_payment_by_idempotency_key(payment_request.idempotency_key)
    if existing:
        return PaymentResponse(**existing)

    # No duplicate found - create a new PaymentIntent via Stripe
    payment_intent = client.v1.payment_intents.create(params={
        "amount": payment_request.amount,
        "currency": payment_request.currency,
        "automatic_payment_methods": {"enabled": True},
    })

    # Store the payment record in the local database
    payment = create_payment(
        amount=payment_request.amount,
        currency=payment_request.currency,
        idempotency_key=payment_request.idempotency_key,
        stripe_payment_intent_id=payment_intent.id,
        status=payment_intent.status,
    )

    return PaymentResponse(**payment, client_secret=payment_intent.client_secret)

What does this endpoint do?

This endpoint handles two scenarios:

  • Duplicate request: get_payment_by_idempotency_key() checks the database for an existing payment with the same key. If found, it returns the existing payment immediately without calling Stripe again.
  • New request: client.v1.payment_intents.create() calls Stripe to create a new PaymentIntent. The result is saved to the database with the idempotency key, then returned along with the client_secret that a frontend would use to confirm the payment.

What is a PaymentIntent?

A PaymentIntent is Stripe's object that tracks a payment through its lifecycle. It starts in a requires_payment_method state and transitions to succeeded or failed as the payment is processed. This state machine pattern is exactly how enterprise payment platforms track transactions.

  • Save main.py.

Seeing a red underline under the Stripe imports?

Make sure your virtual environment is active and you've installed the dependencies from requirements.txt. If you're using Cursor or VS Code, select your virtual environment's Python interpreter from the bottom status bar.

Help me fix import errors in my payment API project.

Build GET endpoints and test your API

With payments being created, you need a way to look them up. These read endpoints let you retrieve a single payment by ID or list all payments.

  • In main.py, add the following endpoints below create_payment_endpoint:
@app.get("/payments/{payment_id}", response_model=PaymentResponse)
def get_payment_endpoint(payment_id: int):
    # Look up the payment by its database ID
    payment = get_payment_by_id(payment_id)
    if not payment:
        raise HTTPException(status_code=404, detail="Payment not found")
    return PaymentResponse(**payment)


@app.get("/payments", response_model=list[PaymentResponse])
def list_payments_endpoint():
    # Return all payments ordered by creation date
    return [PaymentResponse(**p) for p in list_payments()]

What do these endpoints do?

  • GET /payments/{payment_id} fetches a single payment. If the ID doesn't exist, it returns a 404 error.
  • GET /payments returns all payments in the database, sorted newest first.
  • Save main.py.

Now start the server and test your endpoints.

  • Export your Stripe secret key as an environment variable by running this command (replace sk_test_your-key-here with your actual test key from the Stripe Dashboard):
export STRIPE_SECRET_KEY=sk_test_your-key-here
  • Start the FastAPI development server by running:
fastapi dev main.py

What should I see?

Your terminal should show the server starting at http://127.0.0.1:8000 with auto-reload enabled. FastAPI automatically generates interactive API docs you can use to test your endpoints without any extra tools.

  • Open your browser and go to http://127.0.0.1:8000/docs.
  • Click on the POST /payments endpoint to expand it.
  • Click Try it out.
  • Replace the request body with:
{
  "amount": 2000,
  "currency": "usd",
  "idempotency_key": "test-payment-001"
}
  • Click Execute.

You should see a 200 response with your payment record, including a stripe_payment_intent_id starting with pi_ and a client_secret value.

  • Click Execute again with the same request body.

You get back the exact same payment record. The idempotency key prevented a duplicate charge.

Getting an authentication error from Stripe?

Make sure you exported your real STRIPE_SECRET_KEY environment variable before starting the server. The key must start with sk_test_. If you forgot to export it, stop the server with Ctrl+C, export the variable, and restart.

Help me fix Stripe authentication errors in my FastAPI payment API.

✔️ 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 for this step:

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, HTTPException
from stripe import StripeClient

from database import (
    init_db,
    create_payment,
    get_payment_by_idempotency_key,
    get_payment_by_id,
    get_payment_by_stripe_id,
    update_payment_status,
    list_payments,
)
from models import CreatePaymentRequest, PaymentResponse, RefundRequest


STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY", "sk_test_your-key-here")
STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_your-webhook-secret-here")

client = StripeClient(STRIPE_SECRET_KEY)


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(title="Payment Processing API", lifespan=lifespan)


@app.post("/payments", response_model=PaymentResponse)
def create_payment_endpoint(payment_request: CreatePaymentRequest):
    existing = get_payment_by_idempotency_key(payment_request.idempotency_key)
    if existing:
        return PaymentResponse(**existing)

    payment_intent = client.v1.payment_intents.create(params={
        "amount": payment_request.amount,
        "currency": payment_request.currency,
        "automatic_payment_methods": {"enabled": True},
    })

    payment = create_payment(
        amount=payment_request.amount,
        currency=payment_request.currency,
        idempotency_key=payment_request.idempotency_key,
        stripe_payment_intent_id=payment_intent.id,
        status=payment_intent.status,
    )

    return PaymentResponse(**payment, client_secret=payment_intent.client_secret)


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
def get_payment_endpoint(payment_id: int):
    payment = get_payment_by_id(payment_id)
    if not payment:
        raise HTTPException(status_code=404, detail="Payment not found")
    return PaymentResponse(**payment)


@app.get("/payments", response_model=list[PaymentResponse])
def list_payments_endpoint():
    return [PaymentResponse(**p) for p in list_payments()]

Your payment API can now create and retrieve payments with built-in duplicate protection. Next up, you'll add webhook handling so your API can receive real-time payment status updates from Stripe.

Handle Webhooks and State Transitions

Your payment endpoints can create and retrieve payments, but there's a gap in the flow. Stripe processes payments asynchronously, meaning the final result (success or failure) arrives later as an event sent to your server.

This async notification pattern is called a webhook. In this step, you'll build a webhook endpoint that receives events from Stripe, verifies they're authentic, and updates your payment records accordingly.

In this step, get ready to:
  • Build a webhook endpoint that verifies Stripe event signatures.
  • Handle payment success and failure events by updating the database.
  • Test the full end-to-end webhook flow with the Stripe CLI.
Build the webhook endpoint

When a payment succeeds or fails, Stripe sends an HTTP POST request to a URL you specify. Your server needs to read the raw request body, verify the signature using your webhook secret, then react to the event type.

  • In main.py, add the following webhook endpoint at the bottom of the file, below the list_payments_endpoint function:
@app.post("/webhook")
async def stripe_webhook(request: Request):
    # Read the raw request body sent by Stripe
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    # Verify the event signature to confirm it's from Stripe
    try:
        event = client.construct_event(payload, sig_header, STRIPE_WEBHOOK_SECRET)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Webhook signature verification failed: {str(e)}")

    # Handle specific payment event types
    if event.type == "payment_intent.succeeded":
        payment_intent = event.data.object
        payment = get_payment_by_stripe_id(payment_intent["id"])
        if payment:
            update_payment_status(payment["id"], "succeeded")
            print(f"Payment {payment['id']} succeeded")

    elif event.type == "payment_intent.payment_failed":
        payment_intent = event.data.object
        payment = get_payment_by_stripe_id(payment_intent["id"])
        if payment:
            update_payment_status(payment["id"], "failed")
            print(f"Payment {payment['id']} failed")

    else:
        print(f"Unhandled event type: {event.type}")

    return {"status": "ok"}

What does this code do?

This endpoint handles the entire webhook lifecycle in one function:

  • payload = await request.body() reads the raw bytes Stripe sent. You need the unmodified body for signature verification to work correctly.
  • client.construct_event(payload, sig_header, STRIPE_WEBHOOK_SECRET) uses HMAC-SHA256 to verify the request is genuinely from Stripe, not from an attacker.
  • The if/elif/else block is a state machine. It transitions each payment from its current status to succeeded or failed based on the event type Stripe sends.
  • Unhandled event types are logged so you can monitor what other events Stripe sends without crashing your server.

Why webhooks instead of polling?

You could repeatedly call Stripe's API to check payment status, but that's wasteful and slow. Webhooks flip the model: Stripe pushes updates to you the instant something happens.

This is the same event-driven pattern used in enterprise payment systems where milliseconds matter and millions of transactions flow through daily.

  • Save main.py.

Seeing a syntax error?

Make sure the @app.post("/webhook") decorator is at the same indentation level as your other route decorators (no extra indentation).

Check that the f" strings use regular double quotes, not curly or smart quotes from copy-paste.

Help me debug a syntax error in my webhook endpoint.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py before moving on.

ⓧ I'd like to double check the full code

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, HTTPException
from stripe import StripeClient

from database import (
    init_db,
    create_payment,
    get_payment_by_idempotency_key,
    get_payment_by_id,
    get_payment_by_stripe_id,
    update_payment_status,
    list_payments,
)
from models import CreatePaymentRequest, PaymentResponse, RefundRequest


STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY", "sk_test_your-key-here")
STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_your-webhook-secret-here")

client = StripeClient(STRIPE_SECRET_KEY)


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    yield


app = FastAPI(title="Payment Processing API", lifespan=lifespan)


@app.post("/payments", response_model=PaymentResponse)
def create_payment_endpoint(payment_request: CreatePaymentRequest):
    existing = get_payment_by_idempotency_key(payment_request.idempotency_key)
    if existing:
        return PaymentResponse(**existing)

    payment_intent = client.v1.payment_intents.create(params={
        "amount": payment_request.amount,
        "currency": payment_request.currency,
        "automatic_payment_methods": {"enabled": True},
    })

    payment = create_payment(
        amount=payment_request.amount,
        currency=payment_request.currency,
        idempotency_key=payment_request.idempotency_key,
        stripe_payment_intent_id=payment_intent.id,
        status=payment_intent.status,
    )

    return PaymentResponse(**payment, client_secret=payment_intent.client_secret)


@app.get("/payments/{payment_id}", response_model=PaymentResponse)
def get_payment_endpoint(payment_id: int):
    payment = get_payment_by_id(payment_id)
    if not payment:
        raise HTTPException(status_code=404, detail="Payment not found")
    return PaymentResponse(**payment)


@app.get("/payments", response_model=list[PaymentResponse])
def list_payments_endpoint():
    return [PaymentResponse(**p) for p in list_payments()]


@app.post("/webhook")
async def stripe_webhook(request: Request):
    payload = await request.body()
    sig_header = request.headers.get("stripe-signature")

    try:
        event = client.construct_event(payload, sig_header, STRIPE_WEBHOOK_SECRET)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Webhook signature verification failed: {str(e)}")

    if event.type == "payment_intent.succeeded":
        payment_intent = event.data.object
        payment = get_payment_by_stripe_id(payment_intent["id"])
        if payment:
            update_payment_status(payment["id"], "succeeded")
            print(f"Payment {payment['id']} succeeded")

    elif event.type == "payment_intent.payment_failed":
        payment_intent = event.data.object
        payment = get_payment_by_stripe_id(payment_intent["id"])
        if payment:
            update_payment_status(payment["id"], "failed")
            print(f"Payment {payment['id']} failed")

    else:
        print(f"Unhandled event type: {event.type}")

    return {"status": "ok"}
Test the full webhook flow

Testing webhooks locally requires three terminal windows running simultaneously. The Stripe CLI acts as a bridge, forwarding events from Stripe's servers to your local machine.

  • In your first terminal, start the Stripe event listener by running this command:
stripe listen --forward-to localhost:8000/webhook

You should see output like this:

> Ready! Your webhook signing secret is whsec_... (^C to quit)

What is the webhook signing secret?

The whsec_ value printed by stripe listen is the secret your app uses to verify that incoming webhook requests are genuinely from Stripe. Your code already reads this from the STRIPE_WEBHOOK_SECRET environment variable.

  • Copy the whsec_ value from the terminal output.
  • Open a second terminal window.
  • Export the webhook secret and start your server by running these commands:
export STRIPE_WEBHOOK_SECRET=whsec_your-secret-from-stripe-listen
fastapi dev main.py

Windows users

On Windows, use set STRIPE_WEBHOOK_SECRET=whsec_your-secret-from-stripe-listen instead of export.

  • Open a third terminal window.
  • Trigger a test payment success event by running:
stripe trigger payment_intent.succeeded
  • Switch back to your second terminal (where the FastAPI server is running).

You should see a log line printed in your server output confirming the webhook was processed. In your first terminal (the Stripe listener), you should see the event forwarded with a 200 response status.

Seeing a 400 error or signature verification failed?

The most common cause is a mismatch between the webhook secret printed by stripe listen and the value your app is using. Make sure you copied the full whsec_ value and exported it in the same terminal where your FastAPI server is running.

If you started the server before setting the environment variable, stop it with Ctrl+C, export the variable, then restart with fastapi dev main.py.

Help me fix webhook signature verification failures in my Stripe integration.

Your API now handles the full payment lifecycle. Payments flow from creation through to final confirmation via webhooks, with every state transition tracked in your database. Next up, you'll take on the Secret Mission and add refund processing to complete the picture.

Secret mission

Add Refund Processing

Your payment API handles creation and confirmation, but what happens when a customer wants their money back? In this secret mission, you'll add a refund endpoint that handles both full and partial refunds with proper state tracking.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally with a free Stripe test-mode account, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Stripe account (free, no charges in test mode).
  • Local project folder with SQLite database file.
  • Stripe CLI authenticated session.

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the project further.

  • Your Stripe test account is free and persists for future projects.
  • Your local project folder and payments.db file stay on your machine until you delete them manually.
  • The Stripe CLI session remains authenticated until the key expires or you log out.

Pause - I'll come back to this later

Stop your local server to free up the terminal. Everything else is already idle.

  • In the terminal running fastapi dev main.py, press Ctrl+C to stop the server.
  • In the terminal running stripe listen, press Ctrl+C to stop forwarding webhooks.
  • Your project files and Stripe account remain intact for when you return.

Delete - I don't want to use this again

Remove all project resources and deauthenticate the Stripe CLI.

  • Stop any running processes (fastapi dev and stripe listen) by pressing Ctrl+C in each terminal.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the project folder by running:
cd ~/Desktop
rm -rf payment-api

Windows users

On Windows, use these commands instead to remove the folder:

cd ~\Desktop
rmdir /s /q payment-api
  • Log out of the Stripe CLI by running:
stripe logout
  • If you no longer need your Stripe account, you can close it from the Stripe Dashboard. Go to Settings, then navigate to the account information tab to close your account.

Nice Work!

Nice Work!

You just built a production-grade payment processing API from scratch using FastAPI and Stripe. This service demonstrates the exact patterns that power enterprise payment systems at scale.

Here's what you accomplished:

  • Built a FastAPI payment processing service with endpoints to create, retrieve, and list payments through the full lifecycle.
  • Implemented idempotency keys that prevent duplicate charges, a critical safety pattern in enterprise payment platforms.
  • Created a webhook endpoint with Stripe signature verification that processes async payment events and updates transaction state in your SQLite database.
  • Secret Mission: Added refund processing with partial refund support and async webhook confirmation via the charge.refunded event.

Ready to quiz yourself?