Build a Webhook Delivery Service

Introduction

30 Second Summary

Every time Stripe processes a payment or GitHub merges a pull request, those services don't wait for you to refresh the page. They push the update to your server automatically through a webhook.

In this project, you will build and deploy a webhook delivery service using Python and FastAPI. Your service will accept incoming events, persist them to a database, and deliver them to target URLs with automatic retries and exponential backoff.

What You'll Build

You'll fire off a webhook event from your browser, watch it land at a target URL, break the target on purpose, and see the retry engine kick in with exponential backoff while you track every delivery attempt.

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

• A production-structured Python project using uv, src/ layout, and pyproject.toml that you can reuse as a template for any future service.
• A webhook delivery engine where you send an event, watch it get delivered or retried with exponential backoff, and query the full attempt history to see every timestamp and status code.
• A live deployed service on Render backed by PostgreSQL, where your data survives container restarts and redeploys.
• An optional Secret Mission to push your skills further.

Are there any prerequisites?

You'll need Python, a code editor, Git, a GitHub account, and Docker installed on your machine. Step 1 covers everything else.

Before We Start

Before any code hits your editor, let's lock in what you're about to build and why it matters.

Set Up the Modern Python Toolchain

You have committed to building a webhook delivery service. Before you can write a single endpoint, you need the foundation that separates a production Python project from a loose script.

In this step you will install uv, Astral's Rust-based Python project manager that replaces pip, venv, and poetry with a single tool. You will scaffold a project with the modern src/ layout, add your dependencies, configure a linter, and get a minimal FastAPI server running.

In this step, get ready to:

• Install uv and scaffold the hookshot project with a src/ layout.
• Add production and dev dependencies and configure Ruff for linting.
• Create the project's module files and run a minimal FastAPI server.

Install uv and scaffold the project

uv is a single tool for creating environments, managing dependencies, and locking versions. It is the 2026 standard for Python project management.

• Open your terminal. 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 if uv is already installed by running:

uv --version

✔️ I see a version number

You are ready to go. Move on to the next instruction below.

ⓧ Command not found

Install uv using the official installer for your platform.

macOS

• Install uv by running this command:

curl -LsSf https://astral.sh/uv/install.sh | sh

• Close and reopen your terminal so the new PATH takes effect.
• Confirm the install by running:

uv --version

You should see a version number printed.

Windows

• Install uv by running this command in PowerShell:

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

• Close and reopen PowerShell so the new PATH takes effect.
• Confirm the install by running:

uv --version

You should see a version number printed.

Still seeing command not found?

The installer adds uv to your shell profile, but the change only applies to new terminal sessions. Make sure you closed and reopened your terminal after installing.

help me troubleshoot why uv is not found after installing it

• Move to your Desktop by running this command:

cd ~/Desktop

• Scaffold the hookshot project by running this command:

uv init hookshot

What does uv init do?

uv init hookshot creates a new hookshot/ directory on your Desktop with a production-ready project skeleton. Inside you get a pyproject.toml (your project's single config file), a src/hookshot/__init__.py entry point, a .python-version pin, and a README.md.

The src/ layout is the modern Python standard. It prevents import confusion by keeping your source code separate from the project root. This is what every production project uses.

• Move into the new project folder by running:

cd hookshot

Before you add anything, confirm the scaffold is wired correctly. What do you think will happen when you run the project entry point?

• Run the project entry point by running:

uv run hookshot

You should see Hello from hookshot! printed in your terminal. This confirms uv, the project structure, and the entry point all work.

What is uv run?

uv run ensures your lockfile and virtual environment are up to date, then runs the command using your locked dependency versions. You will use uv run for every command in this project instead of activating a virtual environment manually.

Don't see the greeting?

Make sure you ran cd hookshot first. uv run looks for a pyproject.toml in the current directory.

help me troubleshoot why uv run hookshot is not working

Add dependencies and configure Ruff

Your project needs a web framework, an HTTP client for delivering webhooks, and a settings library. You will also add dev tools for linting and testing.

• Add the production dependencies by running:

uv add fastapi uvicorn httpx pydantic-settings aiosqlite

You should see uv resolve and install five packages. Your pyproject.toml now lists them under dependencies.

What are these packages?

• fastapi is the web framework for your API endpoints.
• uvicorn is the ASGI server that runs FastAPI.
• httpx is the async HTTP client you will use to deliver webhooks to target URLs.
• pydantic-settings loads configuration from environment variables.
• aiosqlite is the async driver for SQLite, your local development database.

• Add the dev dependencies by running:

uv add --dev pytest httpx ruff

You should see uv add these under a [dependency-groups] dev section in pyproject.toml.

Why is httpx listed twice?

httpx appears in both production and dev dependencies. In production, your delivery engine uses it to POST webhooks. In development, FastAPI's TestClient uses it under the hood for testing. Listing it in both groups makes each group's purpose self-documenting.

Now configure Ruff, the linter that keeps your code clean. You will add a small config section to pyproject.toml.

• Open pyproject.toml in your code editor.
• Add the following [tool.ruff] section at the bottom of the file:

[tool.ruff]
line-length = 88
select = ["E", "F", "I"]

What does this Ruff config do?

• line-length = 88 sets the max line width, matching the Black formatter's default.
• select = ["E", "F", "I"] enables pycodestyle errors (E), Pyflakes checks (F), and isort import sorting (I).

• Save pyproject.toml.
• Verify Ruff works on the scaffold by running:

uv run ruff check src/

You should see All checks passed! (or no output at all, meaning zero issues found). Ruff is ready.

Seeing linting errors?

If Ruff reports errors, check that you pasted the [tool.ruff] section on its own line, not nested inside another section. Each TOML section header must start on a new line.

help me fix Ruff linting errors in my hookshot project

Create the project file structure and run the server

Your project needs several module files to organize the webhook service. You will create them now, then add a minimal FastAPI app to confirm everything is wired up.

• Create the routes/ subdirectory and all the module files by running these commands:

mkdir src/hookshot/routes
touch src/hookshot/main.py src/hookshot/config.py src/hookshot/models.py src/hookshot/database.py
touch src/hookshot/routes/__init__.py src/hookshot/routes/webhooks.py src/hookshot/routes/targets.py

• Confirm the files exist by running:

ls src/hookshot/

You should see __init__.py, config.py, database.py, main.py, models.py, and a routes directory listed.

• Open src/hookshot/main.py in your code editor and paste the following code:

from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI(title="Hookshot")

# Root endpoint to confirm the service is running
@app.get("/")
async def root():
    return {"status": "ok"}

What does this code do?

• FastAPI(title="Hookshot") creates your application instance. The title appears in the auto-generated API docs.
• @app.get("/") registers an async handler for GET requests to the root URL. It returns a JSON response you can use to quickly check whether the service is alive.

• Save main.py.
• Open src/hookshot/__init__.py in your code editor. Replace its entire contents with:

from hookshot.main import app

def main():
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

What does this code do?

• from hookshot.main import app imports the FastAPI instance so it is available at the package level.
• The main() function is the entry point defined in pyproject.toml under [project.scripts]. When you run uv run hookshot, this function starts the Uvicorn server.

• Save __init__.py.

Time to see it all come together. Before you run the server, what do you expect to see at http://localhost:8000?

• Start the development server by running:

uv run uvicorn hookshot.main:app --reload

What does this command do?

• uv run ensures the environment is synced, then runs the command.
• uvicorn hookshot.main:app tells Uvicorn to find the app object inside hookshot.main.
• --reload makes Uvicorn restart automatically whenever you save a file. This keeps your dev loop fast.

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

You should see {"status":"ok"} displayed in the browser. Your FastAPI server is alive.

• Navigate to http://localhost:8000/docs to see the auto-generated Swagger UI.

You should see a page titled Hookshot with your GET / endpoint listed. This interactive documentation is generated automatically by FastAPI from your code. You will use it heavily to test endpoints in upcoming steps.

Browser showing an error or nothing loading?

Check your terminal for error messages. A common cause is a typo in main.py or __init__.py. Make sure app = FastAPI(title="Hookshot") is at the module level, not indented inside a function.

If you see Address already in use, another process is using port 8000. Stop it with Ctrl+C and try again.

help me troubleshoot why my FastAPI server is not responding at localhost:8000

✔️ Awesome, I've got everything!

Great. Make sure you have saved both main.py and __init__.py.

ⓧ I'd like to double check the full code

Here is what each file should contain at this point.

src/hookshot/main.py

from fastapi import FastAPI

app = FastAPI(title="Hookshot")

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

src/hookshot/__init__.py

from hookshot.main import app

def main():
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

pyproject.toml should contain these sections (it may also have a [build-system] section auto-generated by uv. That is fine to leave in place):

[project]
name = "hookshot"
version = "0.1.0"
description = "A webhook delivery service"
requires-python = ">=3.11"
dependencies = [
    "fastapi",
    "uvicorn",
    "httpx",
    "pydantic-settings",
    "aiosqlite",
]

[project.scripts]
hookshot = "hookshot:main"

[tool.ruff]
line-length = 88
select = ["E", "F", "I"]

[dependency-groups]
dev = [
    "pytest",
    "httpx",
    "ruff",
]

All other module files (config.py, database.py, models.py, routes/__init__.py, routes/webhooks.py, routes/targets.py) should be empty for now. You will fill them in during the next steps.

• Leave the dev server running. Press Ctrl+C to stop it when you need the terminal back.

Your toolchain is locked in. You have a production-structured Python project with a running API server, a linter, and every dependency you need. Next up, you will build the webhook ingestion layer that accepts events and persists them to a database.

Build Webhook Ingestion and Persistence

Your project skeleton is running and the dev server responds at localhost:8000. Now it is time to give the service a real purpose.

The heart of any webhook service is persist-first design. Events must be written to the database before any delivery attempt. If delivery crashes or the target is down, the event is safe because it is already stored. In this step, you will build the database layer and the API endpoints for registering targets and accepting events.

In this step, get ready to:

• Set up an SQLite database with raw SQL to store targets and events.
• Build API endpoints for registering webhook targets and ingesting events.
• Verify that events persist with status "pending" through the Swagger UI.

Set up SQLite with plain SQL and aiosqlite

Before events can be delivered, they need to be stored. Two tables capture everything the service needs. A targets table records where webhooks should go. An events table records what should be delivered, with every new event starting at status "pending".

• Create the database module by pasting this code into database.py:

import aiosqlite

DB_PATH = "./hookshot.db"

async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        # Table for registered webhook target URLs
        await db.execute("""
            CREATE TABLE IF NOT EXISTS targets (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
        """)
        # Table for incoming webhook events
        await db.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                target_id INTEGER NOT NULL,
                payload TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'pending',
                created_at TEXT NOT NULL,
                FOREIGN KEY (target_id) REFERENCES targets(id)
            )
        """)
        await db.commit()

def get_db():
    return aiosqlite.connect(DB_PATH)

What does this code do?

• init_db() opens a connection, runs two CREATE TABLE IF NOT EXISTS statements, then commits. This is idempotent, so restarting the server never duplicates tables.
• The events table defaults every new row to status = 'pending' so the service knows which events still need delivery.
• get_db() returns a fresh connection context manager. Each route opens and closes its own connection, keeping things simple.

• Save database.py.

The database module is ready, but FastAPI does not know about it yet. You need to call init_db() when the app starts. FastAPI uses a lifespan handler for this. Code before yield runs at startup. Code after yield runs at shutdown.

• Replace the entire contents of main.py with this code to wire up database initialization:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from hookshot.database import init_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create database tables when the app starts
    await init_db()
    yield

app = FastAPI(title="Hookshot", lifespan=lifespan)

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

What is a lifespan handler?

The lifespan function is an async context manager that wraps the entire life of your FastAPI app. Everything before yield runs once at startup, and everything after yield runs once at shutdown.

By calling await init_db() before yield, you guarantee the tables exist before any request hits the API.

• Save main.py.
• If you have the dev server running, stop it with Ctrl+C.
• Start the dev server by running:

uv run uvicorn hookshot.main:app --reload

• Confirm the database file was created by running this in a second terminal (inside the hookshot folder):

ls *.db

You should see hookshot.db listed. The lifespan handler ran init_db() at startup and created your SQLite file.

Don't see hookshot.db?

• Make sure you saved both database.py and main.py before restarting the server.
• Check the terminal where uvicorn is running for import errors. A typo in the import path hookshot.database is the most common cause.
• Make sure you are running ls *.db from inside the hookshot project folder, not a subfolder.

Still stuck? Why isn't my hookshot.db file being created when I start the server?

✔️ Awesome, I've got everything!

Great. Double check you have saved database.py and main.py.

ⓧ I'd like to double check the full code

Here is the full database.py for comparison:

import aiosqlite

DB_PATH = "./hookshot.db"

async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS targets (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
        """)
        await db.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                target_id INTEGER NOT NULL,
                payload TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'pending',
                created_at TEXT NOT NULL,
                FOREIGN KEY (target_id) REFERENCES targets(id)
            )
        """)
        await db.commit()

def get_db():
    return aiosqlite.connect(DB_PATH)

Build the target registration endpoints

With the database ready, you need an API that lets callers register target URLs where webhooks should be delivered. You will create Pydantic models to define the request and response shapes, then build the route handlers.

• Define the target request and response schemas by pasting this code into models.py:

from pydantic import BaseModel

# Request body for creating a target
class TargetCreate(BaseModel):
    url: str

# Response body for target endpoints
class TargetResponse(BaseModel):
    id: int
    url: str
    created_at: str

What do these models do?

Pydantic models define the shape of data flowing in and out of your API. TargetCreate tells FastAPI that incoming POST bodies must include a url string. TargetResponse defines the JSON shape callers receive back.

FastAPI uses these models for automatic request validation and for generating the Swagger UI docs at /docs.

• Save models.py.
• Build the target routes by pasting this code into routes/targets.py:

import aiosqlite
import json
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import TargetCreate, TargetResponse

router = APIRouter()

@router.post("/targets", response_model=TargetResponse)
async def create_target(target: TargetCreate):
    now = datetime.now(timezone.utc).isoformat()
    async with get_db() as db:
        # Insert the target and get its auto-generated ID
        cursor = await db.execute(
            "INSERT INTO targets (url, created_at) VALUES (?, ?)",
            (target.url, now)
        )
        await db.commit()
        return TargetResponse(id=cursor.lastrowid, url=target.url, created_at=now)

@router.get("/targets", response_model=list[TargetResponse])
async def list_targets():
    async with get_db() as db:
        # Row factory enables dict-like access by column name
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, url, created_at FROM targets")
        rows = await cursor.fetchall()
        return [TargetResponse(id=r["id"], url=r["url"], created_at=r["created_at"]) for r in rows]

What does this code do?

• POST /api/targets accepts a URL, inserts it into the targets table with a UTC timestamp, and returns the new target with its auto-generated ID.
• GET /api/targets fetches all targets. Setting db.row_factory = aiosqlite.Row lets you access columns by name (r["url"]) instead of by index.
• The ? placeholders in the SQL string are SQLite-style parameterized queries that prevent SQL injection.

• Save routes/targets.py.

The router exists but FastAPI does not know about it yet. You need to import it and register it in main.py.

• In main.py, find this import line:

from hookshot.database import init_db

• Add the following line directly below it:

from hookshot.routes import targets

• Then find this line:

app = FastAPI(title="Hookshot", lifespan=lifespan)

• Add this line directly below it:

app.include_router(targets.router, prefix="/api")

• Save main.py. The dev server should auto-reload.
• Open http://localhost:8000/docs in your browser.

You should see POST /api/targets and GET /api/targets listed in the Swagger UI alongside the existing root endpoint.

• Expand POST /api/targets, click Try it out, and paste this request body:

{
  "url": "https://httpbin.org/post"
}

• Click Execute.

You should see a 200 response with the target's id, url, and created_at fields populated.

• Now expand GET /api/targets, click Try it out, then Execute.

The response should be a list containing the target you just created.

Seeing a 422 or 500 error?

• A 422 means the request body did not match TargetCreate. Make sure the JSON has a "url" key with a string value.
• A 500 usually means a database error. Check the terminal running uvicorn for the full traceback. A NameError: name 'aiosqlite' is not defined means the import aiosqlite line is missing at the top of routes/targets.py.
• If you do not see the target endpoints at all, confirm that app.include_router(targets.router, prefix="/api") is in main.py.

Still stuck? Why am I getting errors when I try to create a target?

✔️ Awesome, I've got everything!

Great. Double check you have saved routes/targets.py.

ⓧ I'd like to double check the full code

Here is the full routes/targets.py for comparison:

import aiosqlite
import json
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import TargetCreate, TargetResponse

router = APIRouter()

@router.post("/targets", response_model=TargetResponse)
async def create_target(target: TargetCreate):
    now = datetime.now(timezone.utc).isoformat()
    async with get_db() as db:
        cursor = await db.execute(
            "INSERT INTO targets (url, created_at) VALUES (?, ?)",
            (target.url, now)
        )
        await db.commit()
        return TargetResponse(id=cursor.lastrowid, url=target.url, created_at=now)

@router.get("/targets", response_model=list[TargetResponse])
async def list_targets():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, url, created_at FROM targets")
        rows = await cursor.fetchall()
        return [TargetResponse(id=r["id"], url=r["url"], created_at=r["created_at"]) for r in rows]

Build the event ingestion endpoint

Targets define where webhooks go. Events define what gets sent. You need an endpoint that accepts an event payload, ties it to a target, and stores it with status "pending". No delivery happens yet. That comes in the next step.

• In models.py, find the last line of the TargetResponse class.
• Add from typing import Any to the import section at the top of the file, then add the following Event models below TargetResponse:

# Request body for creating an event
class EventCreate(BaseModel):
    target_id: int
    payload: dict[str, Any]

# Response body for event endpoints
class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: str

What does dict[str, Any] mean?

The payload field accepts any JSON object. dict[str, Any] tells Pydantic the value must be a dictionary with string keys and any type of value. This lets callers send arbitrary event data like {"event": "user.created", "user_id": 42}.

• Save models.py.
• Build the event routes by pasting this code into routes/webhooks.py:

import aiosqlite
import json
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import EventCreate, EventResponse

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc).isoformat()
    payload_str = json.dumps(event.payload)  # Serialize payload for storage
    async with get_db() as db:
        cursor = await db.execute(
            "INSERT INTO events (target_id, payload, status, created_at) VALUES (?, ?, 'pending', ?)",
            (event.target_id, payload_str, now)
        )
        await db.commit()
        event_id = cursor.lastrowid
    return EventResponse(id=event_id, target_id=event.target_id, payload=event.payload, status="pending", created_at=now)

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row  # Enable dict-like access
        cursor = await db.execute("SELECT id, target_id, payload, status, created_at FROM events")
        rows = await cursor.fetchall()
        return [EventResponse(id=r["id"], target_id=r["target_id"], payload=json.loads(r["payload"]), status=r["status"], created_at=r["created_at"]) for r in rows]

What does this code do?

• POST /api/events accepts a target_id and a JSON payload. It serializes the payload to a string with json.dumps() (SQLite stores text, not native JSON), inserts it with status 'pending', and returns the stored event.
• GET /api/events fetches all events. It deserializes each payload back from a string to a Python dict with json.loads() so the API response returns proper JSON.

• Save routes/webhooks.py.

Just like the targets router, you need to register the webhooks router in main.py.

• In main.py, find this import line:

from hookshot.routes import targets

• Update it to also import webhooks, so the line now looks like this:

from hookshot.routes import targets, webhooks

• Then find this line:

app.include_router(targets.router, prefix="/api")

• Add this line directly below it:

app.include_router(webhooks.router, prefix="/api")

• Save main.py. The dev server should auto-reload.

Time to test the full flow. Before you do, think about this: you are about to send an event to a target. What status do you expect the event to have when you query it back?

• Open http://localhost:8000/docs in your browser.
• Expand POST /api/events, click Try it out, and paste this request body:

{
  "target_id": 1,
  "payload": {"event": "user.created", "user_id": 42}
}

• Click Execute.

You should see a 200 response with the event stored and status set to "pending".

• Now expand GET /api/events, click Try it out, then Execute.

The response is a list containing your event with "status": "pending". The event is stored and waiting. Nothing has tried to deliver it yet.

Seeing a 500 error on the events endpoint?

• If the error mentions FOREIGN KEY constraint failed, the target_id you sent does not match any target. Make sure you created a target first in substep 2 and use its id value (usually 1).
• If the error is a NameError, check that import aiosqlite is at the top of routes/webhooks.py.
• If you do not see the event endpoints at all, confirm app.include_router(webhooks.router, prefix="/api") is in main.py.

Still stuck? Why am I getting errors when I try to create a webhook event?

✔️ Awesome, I've got everything!

Great. Scroll up and confirm each file is saved and matches.

ⓧ I'd like to double check the full code

Here is the full routes/webhooks.py for comparison:

import aiosqlite
import json
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import EventCreate, EventResponse

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc).isoformat()
    payload_str = json.dumps(event.payload)
    async with get_db() as db:
        cursor = await db.execute(
            "INSERT INTO events (target_id, payload, status, created_at) VALUES (?, ?, 'pending', ?)",
            (event.target_id, payload_str, now)
        )
        await db.commit()
        event_id = cursor.lastrowid
    return EventResponse(id=event_id, target_id=event.target_id, payload=event.payload, status="pending", created_at=now)

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, target_id, payload, status, created_at FROM events")
        rows = await cursor.fetchall()
        return [EventResponse(id=r["id"], target_id=r["target_id"], payload=json.loads(r["payload"]), status=r["status"], created_at=r["created_at"]) for r in rows]

Here is the full models.py for comparison:

from pydantic import BaseModel
from typing import Any

class TargetCreate(BaseModel):
    url: str

class TargetResponse(BaseModel):
    id: int
    url: str
    created_at: str

class EventCreate(BaseModel):
    target_id: int
    payload: dict[str, Any]

class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: str

Here is the full main.py for comparison:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from hookshot.database import init_db
from hookshot.routes import targets, webhooks

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

app = FastAPI(title="Hookshot", lifespan=lifespan)

app.include_router(targets.router, prefix="/api")
app.include_router(webhooks.router, prefix="/api")

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

Your events are safely stored and waiting for delivery. Next up, you will build the engine that actually sends them to their targets. It will not go perfectly.

Deliver Webhooks and Watch Them Fail

In the previous step, you built the ingestion layer. Targets and events are sitting in your database, and every event has status "pending". But nothing actually delivers them. A webhook service that stores events without sending them is just a database.

In this step, you'll build naive delivery that fires an HTTP POST to the target URL the moment an event is created. This works when the target is up. But what happens when it's not?

In this step, get ready to:

• Build a delivery function that POSTs events to target URLs and logs every attempt.
• Wire delivery into the event creation flow so events are delivered immediately.
• Test with a working target and a broken one to see what happens when delivery fails.

Add a delivery attempts table

Every delivery attempt needs to be logged, whether it succeeds or fails. Before writing the delivery function itself, you need a table to store these records.

• In database.py, find the second await db.execute call that creates the events table.
• Add the following code directly below it, before the await db.commit() line:

        await db.execute("""
            CREATE TABLE IF NOT EXISTS delivery_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id INTEGER NOT NULL,
                attempted_at TEXT NOT NULL,
                status_code INTEGER,
                success INTEGER NOT NULL,
                error_message TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id)
            )
        """)

What does this table store?

Each row in delivery_attempts records one delivery try for a specific event. It captures the HTTP status code the target returned, whether the attempt succeeded, and any error message if the request itself failed (like a timeout or connection error).

• Save database.py.
• Your dev server should auto-reload. Check the terminal for any errors.

What about the old database?

Because the SQL uses CREATE TABLE IF NOT EXISTS, the new table is added alongside the existing targets and events tables. Any data from Step 2 is preserved.

Create the delivery function

The delivery function is the core of this step. It takes an event, POSTs it to a target URL, logs the attempt, and updates the event's status.

• Create a new file delivery.py inside src/hookshot/ and add this code:

import json
from datetime import datetime, timezone
import httpx
from hookshot.database import get_db

async def deliver_event(event_id: int, target_url: str, payload: dict):
    now = datetime.now(timezone.utc).isoformat()
    status_code = None
    success = False
    error_message = None
    try:
        # POST the event payload to the target URL
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(target_url, json=payload)
            status_code = response.status_code
            if response.is_success:
                success = True
    except Exception as e:
        error_message = str(e)

What does this code do?

• httpx.AsyncClient sends an async HTTP POST with the event payload as JSON. The 10-second timeout prevents hanging on unresponsive targets.
• response.is_success checks if the status code is in the 2xx range. Anything else (4xx, 5xx) counts as a failure.
• The broad except Exception catches network-level failures like timeouts, DNS errors, and connection refused. The error message is stored for debugging.

• Add the second half of the function below the except block. This logs the attempt and updates the event status:

    # Log the delivery attempt and update event status
    async with get_db() as db:
        await db.execute(
            "INSERT INTO delivery_attempts (event_id, attempted_at, status_code, success, error_message) VALUES (?, ?, ?, ?, ?)",
            (event_id, now, status_code, int(success), error_message)
        )
        new_status = "delivered" if success else "failed"
        await db.execute(
            "UPDATE events SET status = ? WHERE id = ?",
            (new_status, event_id)
        )
        await db.commit()

Why log the attempt before updating the status?

The attempt record captures the raw result: what status code came back, whether it succeeded, and any error message. The event status update ("delivered" or "failed") is the summary that the API returns. Both happen in a single database transaction so they stay in sync.

• Save delivery.py.

Add the response model and fix an import

Before wiring delivery into the API, you need a response model for delivery attempt data. You also need to fix a missing import in targets.py that would cause errors when listing targets.

• In models.py, add this class at the bottom of the file, below EventResponse:

class DeliveryAttemptResponse(BaseModel):
    id: int
    event_id: int
    attempted_at: str
    status_code: int | None
    success: bool
    error_message: str | None

• Save models.py.

• In routes/targets.py, find this import line at the top:

import json

• Replace it with:

import aiosqlite

• Save targets.py. Check the terminal. The dev server should reload without errors.

Why this import swap?

The list_targets function uses aiosqlite.Row as the row factory, but the file was importing json (which it never uses) instead of aiosqlite. This fix prevents a NameError when you call GET /api/targets.

Server not reloading?

• Make sure your dev server is still running with the --reload flag. If you stopped it, restart it by running uv run uvicorn hookshot.main:app --reload from the hookshot project folder.
• If you see an ImportError, double-check that delivery.py is inside src/hookshot/, not in the project root.

Why is my dev server crashing on reload?

Wire delivery into the event creation flow

The delivery function exists, but nothing calls it. You need to update the event creation endpoint so it looks up the target URL, persists the event, delivers it, and then returns the updated status.

• Replace the entire content of routes/webhooks.py with the following imports and updated create_event function:

import json
import asyncio
import aiosqlite
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import EventCreate, EventResponse, DeliveryAttemptResponse
from hookshot.delivery import deliver_event

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc).isoformat()
    payload_str = json.dumps(event.payload)
    async with get_db() as db:
        # Look up the target URL before delivery
        db.row_factory = aiosqlite.Row
        target_row = await db.execute("SELECT url FROM targets WHERE id = ?", (event.target_id,))
        target = await target_row.fetchone()
        cursor = await db.execute(
            "INSERT INTO events (target_id, payload, status, created_at) VALUES (?, ?, 'pending', ?)",
            (event.target_id, payload_str, now)
        )
        await db.commit()
        event_id = cursor.lastrowid

What changed?

• The imports now include deliver_event, DeliveryAttemptResponse, and aiosqlite for the row factory.
• Before inserting the event, the function now looks up the target's URL using its ID. This is needed so deliver_event knows where to send the POST request.

• Add the rest of the create_event function directly below the code you just pasted. This is the part that delivers the event and returns the updated status:

    # Deliver immediately after persisting
    await deliver_event(event_id, target["url"], event.payload)
    # Re-read the event to get the updated status
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        row = await (await db.execute("SELECT id, target_id, payload, status, created_at FROM events WHERE id = ?", (event_id,))).fetchone()
    return EventResponse(id=row["id"], target_id=row["target_id"], payload=json.loads(row["payload"]), status=row["status"], created_at=row["created_at"])

Why re-read the event after delivery?

The deliver_event function updates the event's status in the database ("delivered" or "failed"). By re-reading the event row afterward, the API response shows the actual delivery result instead of the stale "pending" status.

• Now add the list_events endpoint below create_event. This is the same function from Step 2, now with the correct aiosqlite import at the top of the file:

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, target_id, payload, status, created_at FROM events")
        rows = await cursor.fetchall()
        return [EventResponse(id=r["id"], target_id=r["target_id"], payload=json.loads(r["payload"]), status=r["status"], created_at=r["created_at"]) for r in rows]

• Save webhooks.py. Check the terminal to confirm the server reloads without errors.

Time to test. If you have leftover data from Step 2, delete the old database so the new delivery_attempts table is created fresh.

• Stop the dev server with Ctrl+C.
• Delete the old database and restart the server by running these commands:

rm -f hookshot.db
uv run uvicorn hookshot.main:app --reload

• Open the Swagger UI at http://localhost:8000/docs in your browser.
• Expand POST /api/targets, click Try it out, and register a target with this request body:

{"url": "https://httpbin.org/post"}

• Click Execute. Note the target id in the response (likely 1).
• Now expand POST /api/events, click Try it out, and send an event to that target:

{"target_id": 1, "payload": {"message": "hello from hookshot"}}

• Click Execute.

The response should show your event with "status": "delivered". Your webhook service just made its first delivery.

Seeing status "pending" or an error?

• If the status is still "pending", the deliver_event call is not being reached. Confirm that await deliver_event(event_id, target["url"], event.payload) appears in your create_event function after the db.commit() block.
• If you see a 500 Internal Server Error in Swagger, check the terminal for the full traceback. A common cause is a missing import aiosqlite at the top of webhooks.py.
• Make sure you deleted hookshot.db and restarted the server so the delivery_attempts table gets created.

My webhook delivery returns pending or 500. Help me debug.

Test with a broken target and watch delivery fail

Delivering to a healthy target worked. But production targets go down. Servers crash. Networks fail. What happens to your webhooks when the target cannot accept them?

• In the Swagger UI, expand POST /api/targets and register a target that always returns HTTP 500:

{"url": "https://httpbin.org/status/500"}

• Click Execute. Note the new target id (likely 2).

Before you send an event to this broken target, take a moment to predict. What do you think the event status will be? Will the service retry?

• Expand POST /api/events and send an event to the broken target:

{"target_id": 2, "payload": {"message": "this will fail"}}

• Click Execute.

The response shows "status": "failed". One attempt. No retry. The webhook is permanently lost.

Why this matters

This is the problem you just built. A single failed delivery means a lost webhook. In production, targets go down for seconds, minutes, or hours due to deployments, outages, or rate limiting. A naive fire-once approach permanently loses events during any temporary downtime.

This is exactly the gap the next step fills. Production webhook systems retry with exponential backoff, giving transient failures time to resolve.

Add delivery attempt history

You know the delivery failed, but right now you can only see the final status. To understand what actually happened during a delivery, you need an endpoint that shows the full attempt history.

• In routes/webhooks.py, add this endpoint below the list_events function:

@router.get("/events/{event_id}/attempts", response_model=list[DeliveryAttemptResponse])
async def list_attempts(event_id: int):
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, event_id, attempted_at, status_code, success, error_message FROM delivery_attempts WHERE event_id = ? ORDER BY attempted_at", (event_id,))
        rows = await cursor.fetchall()
        return [DeliveryAttemptResponse(id=r["id"], event_id=r["event_id"], attempted_at=r["attempted_at"], status_code=r["status_code"], success=bool(r["success"]), error_message=r["error_message"]) for r in rows]

What does this endpoint show?

This queries every delivery attempt for a specific event, ordered by timestamp. Each row shows the HTTP status code the target returned, whether it counted as a success, and any error message from a network failure. This is the audit trail for every delivery.

• Save webhooks.py.

Before you try the new endpoint, what do you expect to see for the failed event? How many attempts should be logged?

• In the Swagger UI, expand GET /api/events/{event_id}/attempts, click Try it out, enter 2 for the event ID (the failed event), and click Execute.

You should see a single delivery attempt with status_code: 500 and success: false. One attempt. No retry. The evidence is right there in the data.

Seeing an empty list?

• If you deleted hookshot.db after the failed delivery test, the attempt data was lost. Register the broken target again, send another event, and retry.
• Make sure you are querying the correct event ID. Check GET /api/events to see which event has status "failed" and use that ID.

Why am I getting an empty list for delivery attempts?

✔️ Awesome, I've got everything!

Double check you have saved all your files: database.py, delivery.py, models.py, routes/targets.py, and routes/webhooks.py.

ⓧ I'd like to double check the full code

Here are the files you changed in this step. Scroll through each one and compare against your code.

database.py (added the delivery_attempts table):

import aiosqlite

DB_PATH = "./hookshot.db"

async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS targets (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                created_at TEXT NOT NULL
            )
        """)
        await db.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                target_id INTEGER NOT NULL,
                payload TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'pending',
                created_at TEXT NOT NULL,
                FOREIGN KEY (target_id) REFERENCES targets(id)
            )
        """)

        await db.execute("""
            CREATE TABLE IF NOT EXISTS delivery_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id INTEGER NOT NULL,
                attempted_at TEXT NOT NULL,
                status_code INTEGER,
                success INTEGER NOT NULL,
                error_message TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id)
            )
        """)
        await db.commit()

def get_db():
    return aiosqlite.connect(DB_PATH)

delivery.py (new file):

import json
from datetime import datetime, timezone
import httpx
from hookshot.database import get_db

async def deliver_event(event_id: int, target_url: str, payload: dict):
    now = datetime.now(timezone.utc).isoformat()
    status_code = None
    success = False
    error_message = None
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(target_url, json=payload)
            status_code = response.status_code
            if response.is_success:
                success = True
    except Exception as e:
        error_message = str(e)

    async with get_db() as db:
        await db.execute(
            "INSERT INTO delivery_attempts (event_id, attempted_at, status_code, success, error_message) VALUES (?, ?, ?, ?, ?)",
            (event_id, now, status_code, int(success), error_message)
        )
        new_status = "delivered" if success else "failed"
        await db.execute(
            "UPDATE events SET status = ? WHERE id = ?",
            (new_status, event_id)
        )
        await db.commit()

models.py (added DeliveryAttemptResponse):

from pydantic import BaseModel
from typing import Any

class TargetCreate(BaseModel):
    url: str

class TargetResponse(BaseModel):
    id: int
    url: str
    created_at: str

class EventCreate(BaseModel):
    target_id: int
    payload: dict[str, Any]

class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: str

class DeliveryAttemptResponse(BaseModel):
    id: int
    event_id: int
    attempted_at: str
    status_code: int | None
    success: bool
    error_message: str | None

routes/targets.py (fixed import):

import aiosqlite
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import TargetCreate, TargetResponse

router = APIRouter()

@router.post("/targets", response_model=TargetResponse)
async def create_target(target: TargetCreate):
    now = datetime.now(timezone.utc).isoformat()
    async with get_db() as db:
        cursor = await db.execute(
            "INSERT INTO targets (url, created_at) VALUES (?, ?)",
            (target.url, now)
        )
        await db.commit()
        return TargetResponse(id=cursor.lastrowid, url=target.url, created_at=now)

@router.get("/targets", response_model=list[TargetResponse])
async def list_targets():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, url, created_at FROM targets")
        rows = await cursor.fetchall()
        return [TargetResponse(id=r["id"], url=r["url"], created_at=r["created_at"]) for r in rows]

routes/webhooks.py (updated imports, delivery call, and new endpoint):

import json
import asyncio
import aiosqlite
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import EventCreate, EventResponse, DeliveryAttemptResponse
from hookshot.delivery import deliver_event

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc).isoformat()
    payload_str = json.dumps(event.payload)
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        target_row = await db.execute("SELECT url FROM targets WHERE id = ?", (event.target_id,))
        target = await target_row.fetchone()
        cursor = await db.execute(
            "INSERT INTO events (target_id, payload, status, created_at) VALUES (?, ?, 'pending', ?)",
            (event.target_id, payload_str, now)
        )
        await db.commit()
        event_id = cursor.lastrowid

    await deliver_event(event_id, target["url"], event.payload)
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        row = await (await db.execute("SELECT id, target_id, payload, status, created_at FROM events WHERE id = ?", (event_id,))).fetchone()
    return EventResponse(id=row["id"], target_id=row["target_id"], payload=json.loads(row["payload"]), status=row["status"], created_at=row["created_at"])

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, target_id, payload, status, created_at FROM events")
        rows = await cursor.fetchall()
        return [EventResponse(id=r["id"], target_id=r["target_id"], payload=json.loads(r["payload"]), status=r["status"], created_at=r["created_at"]) for r in rows]

@router.get("/events/{event_id}/attempts", response_model=list[DeliveryAttemptResponse])
async def list_attempts(event_id: int):
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, event_id, attempted_at, status_code, success, error_message FROM delivery_attempts WHERE event_id = ? ORDER BY attempted_at", (event_id,))
        rows = await cursor.fetchall()
        return [DeliveryAttemptResponse(id=r["id"], event_id=r["event_id"], attempted_at=r["attempted_at"], status_code=r["status_code"], success=bool(r["success"]), error_message=r["error_message"]) for r in rows]

You now have a webhook delivery service that delivers events and logs every attempt. But it has a critical flaw you just proved: when a target is down, the webhook is permanently lost. In the next step, you'll build a retry engine with exponential backoff that gives failing targets time to recover.

Build the Retry Engine with Exponential Backoff

In the last step, you watched a webhook vanish. A single HTTP 500 from the target and your event was permanently marked "failed" with no second chance. That is not how production webhook systems work.

Production systems like Stripe and GitHub retry failed deliveries with exponential backoff. Instead of hammering a downed target every second, each retry waits longer than the last. This gives transient failures time to resolve. In this step, you will replace the naive fire-and-forget delivery with a proper retry engine.

In this step, get ready to:

• Build a retry engine that re-attempts failed deliveries with exponential backoff and jitter.
• Add delivery attempt tracking with attempt numbers and scheduled retry times.
• Test the retry engine end-to-end and watch attempts accumulate with increasing gaps.

Implement the retry engine with exponential backoff

The retry engine needs two things: new database columns to track attempt numbers and scheduled retry times, and a rewritten delivery function that spawns background retry tasks on failure.

• In database.py, find the delivery_attempts table creation. It currently looks like this:

        await db.execute("""
            CREATE TABLE IF NOT EXISTS delivery_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id INTEGER NOT NULL,
                attempted_at TEXT NOT NULL,
                status_code INTEGER,
                success INTEGER NOT NULL,
                error_message TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id)
            )
        """)

• Add two new columns before the FOREIGN KEY line: attempt_number and next_retry_at. The updated block should look like this:

        await db.execute("""
            CREATE TABLE IF NOT EXISTS delivery_attempts (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_id INTEGER NOT NULL,
                attempted_at TEXT NOT NULL,
                status_code INTEGER,
                success INTEGER NOT NULL,
                error_message TEXT,
                attempt_number INTEGER NOT NULL DEFAULT 0,
                next_retry_at TEXT,
                FOREIGN KEY (event_id) REFERENCES events(id)
            )
        """)

What do these columns do?

attempt_number tracks which retry this is (0 for the first try, 1 for the first retry, etc.). next_retry_at records when the next retry is scheduled, so you can query it later.

The existing hookshot.db file was created with the old schema. SQLite's CREATE TABLE IF NOT EXISTS will not add new columns to an existing table.

• Stop the dev server by pressing Ctrl+C in the terminal.
• Delete the old database file by running this command:

rm hookshot.db

Now for the core change. You will replace the entire contents of delivery.py with a retry-aware delivery engine.

• Open delivery.py, select all the existing code, and replace it with this:

import json
import asyncio
import random
from datetime import datetime, timezone, timedelta
import httpx
from hookshot.database import get_db

# Retry configuration
MAX_RETRIES = 3
BASE_DELAY = 2  # seconds

async def _attempt_delivery(event_id: int, target_url: str, payload: dict, attempt_number: int):
    now = datetime.now(timezone.utc).isoformat()
    status_code = None
    success = False
    error_message = None
    next_retry_at = None

    # Attempt the HTTP POST to the target URL
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(target_url, json=payload)
            status_code = response.status_code
            if response.is_success:
                success = True
    except Exception as e:
        error_message = str(e)

    # Calculate when the next retry should happen
    if not success and attempt_number < MAX_RETRIES:
        jitter = random.uniform(0, 1)
        delay = BASE_DELAY * (2 ** attempt_number) + jitter
        next_retry_at = (datetime.now(timezone.utc) + timedelta(seconds=delay)).isoformat()

What does this code do?

MAX_RETRIES = 3 and BASE_DELAY = 2 set the retry policy. The function _attempt_delivery replaces your old deliver_event and adds an attempt_number parameter to track which try this is.

The backoff formula BASE_DELAY * (2 ** attempt_number) + jitter produces increasing delays: roughly 2s, 4s, 8s. The random jitter prevents multiple failed events from retrying at the exact same instant.

• Directly below the code you just pasted (still inside delivery.py), add the database logging, retry spawning, and the two helper functions:

    # Log this attempt and update event status
    async with get_db() as db:
        await db.execute(
            "INSERT INTO delivery_attempts (event_id, attempted_at, status_code, success, error_message, attempt_number, next_retry_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (event_id, now, status_code, int(success), error_message, attempt_number, next_retry_at)
        )
        if success:
            new_status = "delivered"
        elif attempt_number < MAX_RETRIES:
            new_status = "retrying"
        else:
            new_status = "failed"
        await db.execute("UPDATE events SET status = ? WHERE id = ?", (new_status, event_id))
        await db.commit()

    # Spawn a background task to retry after the backoff delay
    if not success and attempt_number < MAX_RETRIES:
        jitter = random.uniform(0, 1)
        delay = BASE_DELAY * (2 ** attempt_number) + jitter
        asyncio.create_task(_retry_after(delay, event_id, target_url, payload, attempt_number + 1))

async def _retry_after(delay: float, event_id: int, target_url: str, payload: dict, attempt_number: int):
    await asyncio.sleep(delay)
    await _attempt_delivery(event_id, target_url, payload, attempt_number)

async def deliver_event(event_id: int, target_url: str, payload: dict):
    await _attempt_delivery(event_id, target_url, payload, 0)

What does this code do?

After each attempt, the function logs it to the database and sets the event status. If delivery failed and retries remain, the status becomes "retrying" instead of "failed". Only after all retries are exhausted does it become "failed".

asyncio.create_task(_retry_after(...)) spawns a background task that sleeps for the backoff duration, then re-attempts delivery. The deliver_event function is now a thin wrapper that kicks off the first attempt (attempt 0).

• Save database.py and delivery.py.
• Restart the dev server by running:

uv run uvicorn hookshot.main:app --reload

• Visit http://localhost:8000 in your browser to confirm the server is running and returns {"status": "ok"}.

Seeing an import error or missing column?

Make sure you deleted hookshot.db before restarting. The old database has the old schema without the new columns.

Double-check that the second code block in delivery.py is indented correctly. The async with get_db() block should be indented inside _attempt_delivery, at the same level as the try: block above it.

Help me debug my delivery.py retry engine

Add delivery attempt tracking and status queries

The retry engine writes new data (attempt numbers, retry times), but the API response models and query routes do not return it yet. You need to update the Pydantic models and the webhook routes to surface this information.

• In models.py, find the EventResponse class. It currently looks like this:

class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: str

• Add two new fields at the end of the class, so it looks like this:

class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: str
    attempt_count: int = 0
    next_retry_at: str | None = None

• Now find the DeliveryAttemptResponse class. It currently looks like this:

class DeliveryAttemptResponse(BaseModel):
    id: int
    event_id: int
    attempted_at: str
    status_code: int | None
    success: bool
    error_message: str | None

• Add two new fields at the end, so it looks like this:

class DeliveryAttemptResponse(BaseModel):
    id: int
    event_id: int
    attempted_at: str
    status_code: int | None
    success: bool
    error_message: str | None
    attempt_number: int
    next_retry_at: str | None

What do these new fields do?

attempt_count on EventResponse tells you how many delivery attempts have been made for that event. next_retry_at tells you when the next retry is scheduled (or null if no retry is pending).

The same two fields on DeliveryAttemptResponse let you see the full history: which attempt number each entry was and when the next one was scheduled.

• Save models.py.

Next, you need to update the webhook routes to query these new columns and return them in the responses.

• Open routes/webhooks.py, select all the existing code, and replace it with the updated imports and create_event function:

import json
import aiosqlite
from datetime import datetime, timezone
from fastapi import APIRouter
from hookshot.database import get_db
from hookshot.models import EventCreate, EventResponse, DeliveryAttemptResponse
from hookshot.delivery import deliver_event

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc).isoformat()
    payload_str = json.dumps(event.payload)
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        # Look up the target URL for the given target_id
        target_row = await (await db.execute("SELECT url FROM targets WHERE id = ?", (event.target_id,))).fetchone()
        cursor = await db.execute(
            "INSERT INTO events (target_id, payload, status, created_at) VALUES (?, ?, 'pending', ?)",
            (event.target_id, payload_str, now)
        )
        await db.commit()
        event_id = cursor.lastrowid
    # Deliver the event (retries happen in the background)
    await deliver_event(event_id, target_row["url"], event.payload)
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        row = await (await db.execute("SELECT id, target_id, payload, status, created_at FROM events WHERE id = ?", (event_id,))).fetchone()
        count_row = await (await db.execute("SELECT COUNT(*) as cnt FROM delivery_attempts WHERE event_id = ?", (event_id,))).fetchone()
        next_retry_row = await (await db.execute("SELECT next_retry_at FROM delivery_attempts WHERE event_id = ? ORDER BY attempt_number DESC LIMIT 1", (event_id,))).fetchone()
    return EventResponse(id=row["id"], target_id=row["target_id"], payload=json.loads(row["payload"]), status=row["status"], created_at=row["created_at"], attempt_count=count_row["cnt"], next_retry_at=next_retry_row["next_retry_at"] if next_retry_row else None)

What changed here?

The import asyncio line is gone because the retry logic now lives in delivery.py. After calling deliver_event, the response now queries the delivery_attempts table for the attempt count and the latest retry time.

• Directly below the code you just pasted (still in webhooks.py), add the updated list_events and list_attempts functions:

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute("SELECT id, target_id, payload, status, created_at FROM events")
        rows = await cursor.fetchall()
        result = []
        for r in rows:
            # Query attempt count and latest retry time for each event
            count_row = await (await db.execute("SELECT COUNT(*) as cnt FROM delivery_attempts WHERE event_id = ?", (r["id"],))).fetchone()
            next_retry_row = await (await db.execute("SELECT next_retry_at FROM delivery_attempts WHERE event_id = ? ORDER BY attempt_number DESC LIMIT 1", (r["id"],))).fetchone()
            result.append(EventResponse(id=r["id"], target_id=r["target_id"], payload=json.loads(r["payload"]), status=r["status"], created_at=r["created_at"], attempt_count=count_row["cnt"], next_retry_at=next_retry_row["next_retry_at"] if next_retry_row else None))
        return result

@router.get("/events/{event_id}/attempts", response_model=list[DeliveryAttemptResponse])
async def list_attempts(event_id: int):
    async with get_db() as db:
        db.row_factory = aiosqlite.Row
        # Select all columns including the new attempt_number and next_retry_at
        cursor = await db.execute("SELECT id, event_id, attempted_at, status_code, success, error_message, attempt_number, next_retry_at FROM delivery_attempts WHERE event_id = ? ORDER BY attempt_number", (event_id,))
        rows = await cursor.fetchall()
        return [DeliveryAttemptResponse(id=r["id"], event_id=r["event_id"], attempted_at=r["attempted_at"], status_code=r["status_code"], success=bool(r["success"]), error_message=r["error_message"], attempt_number=r["attempt_number"], next_retry_at=r["next_retry_at"]) for r in rows]

What does this code do?

list_events now loops through each event and queries the attempt count and latest retry time. This means every event in the list shows how many delivery attempts have been made and when the next one is scheduled.

list_attempts now selects attempt_number and next_retry_at from the database and orders by attempt_number so you see attempts in chronological order.

• Save models.py and webhooks.py. The dev server should auto-reload.
• Visit http://localhost:8000/docs to confirm the Swagger UI loads without errors.

Server not reloading or showing validation errors?

Check that DeliveryAttemptResponse in models.py has both attempt_number: int and next_retry_at: str | None. If the model and database columns do not match, FastAPI will throw a validation error when you query attempts.

Help me debug my webhooks.py route updates

✔️ Awesome, I've got everything!

Great. Make sure you have saved all four files: database.py, delivery.py, models.py, and routes/webhooks.py.

ⓧ I'd like to double check the full code

Scroll up and confirm each code block is present in the right file. Here is a summary of what each file should contain:

• database.py: The delivery_attempts table now includes attempt_number INTEGER NOT NULL DEFAULT 0 and next_retry_at TEXT before the FOREIGN KEY line.
• delivery.py: Two code blocks pasted sequentially. Imports and constants at the top, then _attempt_delivery (the backoff calculation, DB logging, and retry spawning), then _retry_after and deliver_event at the bottom.
• models.py: EventResponse has attempt_count: int = 0 and next_retry_at: str | None = None. DeliveryAttemptResponse has attempt_number: int and next_retry_at: str | None.
• routes/webhooks.py: Two code blocks pasted sequentially. The imports no longer include asyncio. All three route functions (create_event, list_events, list_attempts) now query and return attempt tracking fields.

Test the retry engine end-to-end

Time to see the retry engine in action. You deleted the old database, so you need to register a fresh target before sending events.

• Open the Swagger UI at http://localhost:8000/docs.
• Expand POST /api/targets, click Try it out, and register a target with URL https://httpbin.org/status/500. Click Execute.
• Note the id returned in the response (it should be 1 since you deleted the old database).

Before you send an event to this target, think about what you expect to happen. In the last step, a failed delivery was immediately marked "failed". What should happen now that you have a retry engine?

• Expand POST /api/events, click Try it out, and send an event with target_id set to 1 and any JSON payload. Click Execute.

The response should show "status": "retrying" instead of "failed". The first delivery attempt failed (the target returns HTTP 500), but the retry engine has already scheduled a background retry.

• Wait about 15 seconds to give the retries time to run (the backoff schedule is roughly 2s, 4s, 8s).
• Expand GET /api/events/{event_id}/attempts, click Try it out, enter 1 for the event ID, and click Execute.

You should see 4 delivery attempts (attempt 0 through attempt 3). Each one has a status_code of 500 and success: false. Look at the attempted_at timestamps. The gaps between them grow: roughly 2 seconds, then 4, then 8. That is exponential backoff in action.

• Expand GET /api/events, click Try it out, and click Execute to check the event's final status.

The event status should now be "failed" with attempt_count of 4. The retry engine tried 4 times (the initial attempt plus 3 retries) and gave up. This is a fundamentally different outcome from the naive delivery. The webhook was not silently lost after one failure. The system fought to deliver it.

Why 4 attempts and not 3?

MAX_RETRIES = 3 means 3 retries after the initial attempt. The initial delivery (attempt 0) is not counted as a retry. So you get attempt 0, retry 1, retry 2, retry 3, totaling 4 attempts.

Only seeing 1 attempt?

You may not have waited long enough. The retries need about 15 seconds total to complete (2 + 4 + 8 seconds plus jitter). Wait and query the attempts endpoint again.

If the attempt count stays at 1 after 30 seconds, check your terminal for errors. The background task may have crashed silently. Make sure asyncio.create_task is being called inside _attempt_delivery after the database write.

Help me debug why my retry engine is not spawning retries

Your webhook service now has real resilience. Failed deliveries get retried with exponential backoff, every attempt is logged, and the event status tracks the full lifecycle from "pending" through "retrying" to "delivered" or "failed". Next up, you will harden this service for production and containerize it with Docker.

Harden for Production and Containerize

Your webhook service works, but it is carrying several liabilities that would bite you in production. Configuration values like MAX_RETRIES and BASE_DELAY are hardcoded. There is no structured logging. There is no health check endpoint. And the database layer uses raw aiosqlite SQL with SQLite-specific syntax.

That last point is the critical one. Your raw SQL uses AUTOINCREMENT, ? placeholders, and INTEGER for booleans. None of that runs on PostgreSQL. Rather than writing your own SQL compatibility layer, you will adopt SQLAlchemy 2.0's async engine, which was designed to solve exactly this problem.

In this step, get ready to:

• Replace hardcoded config with environment-based settings and refactor the database layer to SQLAlchemy.
• Write tests with pytest and verify linting with Ruff.
• Containerize the service with a multi-stage Dockerfile and run it locally.

Add environment config and install SQLAlchemy

Before touching any application code, you need the new dependencies installed and a config layer that reads from environment variables.

• Add the SQLAlchemy and asyncpg dependencies by running:

uv add 'sqlalchemy[asyncio]' asyncpg

What do these packages do?

The sqlalchemy[asyncio] extra installs SQLAlchemy with the greenlet dependency needed for async support. asyncpg is the async PostgreSQL driver that SQLAlchemy uses when your DATABASE_URL starts with postgresql+asyncpg://.

• Open src/hookshot/config.py in your editor and replace the empty file with this Settings class:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Database connection string (swapped per environment)
    database_url: str = "sqlite+aiosqlite:///./hookshot.db"
    max_retries: int = 3
    base_delay: float = 2.0
    log_level: str = "INFO"
    environment: str = "development"

    class Config:
        env_file = ".env"

settings = Settings()

What does this code do?

The Settings class extends BaseSettings from pydantic-settings. Every field maps to an environment variable of the same name (case-insensitive). If no env var is set, the default value is used.

The class Config block tells pydantic-settings to also read values from a .env file in your project root. The module-level settings = Settings() creates a singleton that every other module can import.

• Create a .env file in the root of the hookshot project folder (the same level as pyproject.toml) with these development defaults:

DATABASE_URL=sqlite+aiosqlite:///./hookshot.db
MAX_RETRIES=3
BASE_DELAY=2
LOG_LEVEL=INFO
ENVIRONMENT=development

• Create a .gitignore file at the same level to keep secrets and generated files out of version control:

.env
.venv/
__pycache__/
*.pyc
hookshot.db

• Verify that the config loads correctly by running:

uv run python -c "from hookshot.config import settings; print(settings.database_url)"

You should see sqlite+aiosqlite:///./hookshot.db printed.

Seeing an import error?

Make sure pydantic-settings is in your pyproject.toml dependencies. It was added in Step 1. If missing, run uv add pydantic-settings.

Help me debug a pydantic-settings import error in my hookshot project.

Refactor the database layer to SQLAlchemy

This is the biggest change in the project. You are replacing every raw SQL statement with SQLAlchemy ORM models and queries that run identically on SQLite and PostgreSQL.

• Open src/hookshot/database.py and delete all the existing code. Replace it with the engine setup and the Target model:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, Float, Boolean, Text, ForeignKey, DateTime
from datetime import datetime, timezone
from hookshot.config import settings

# Create the async engine and session factory
engine = create_async_engine(settings.database_url, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# ORM model for webhook target URLs
class Target(Base):
    __tablename__ = "targets"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    url: Mapped[str] = mapped_column(String, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

What does this code do?

The create_async_engine function takes your DATABASE_URL and creates a connection pool. When the URL is sqlite+aiosqlite://, it speaks SQLite. When it is postgresql+asyncpg://, the same code speaks Postgres. Zero code changes.

The Target class uses Mapped type annotations and mapped_column() to define columns. SQLAlchemy translates these into the correct SQL dialect at runtime.

• Still in database.py, add the remaining models and helper functions directly below the Target class:

# ORM model for webhook events
class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    target_id: Mapped[int] = mapped_column(Integer, ForeignKey("targets.id"), nullable=False)
    payload: Mapped[str] = mapped_column(Text, nullable=False)
    status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)

# ORM model for delivery attempt history
class DeliveryAttempt(Base):
    __tablename__ = "delivery_attempts"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    event_id: Mapped[int] = mapped_column(Integer, ForeignKey("events.id"), nullable=False)
    attempted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    status_code: Mapped[int | None] = mapped_column(Integer, nullable=True)
    success: Mapped[bool] = mapped_column(Boolean, nullable=False)
    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
    attempt_number: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

def get_session() -> AsyncSession:
    return AsyncSessionLocal()

What does this code do?

The Event and DeliveryAttempt models mirror the tables you created earlier with raw SQL, but now SQLAlchemy handles type mapping across dialects. Boolean becomes INTEGER on SQLite and BOOLEAN on Postgres automatically.

The init_db() function uses Base.metadata.create_all to generate all tables from the ORM definitions. The get_session() function returns an async session for database operations.

• Delete the old SQLite database file since the schema source has changed. Run:

rm -f hookshot.db

• Verify the new database models load correctly by running:

uv run python -c "from hookshot.database import Target, Event, DeliveryAttempt; print('Models loaded')"

You should see Models loaded printed.

Now you need to update every file that used the old raw SQL. The app will not start until all of these files are refactored.

• Open src/hookshot/models.py and replace the entire file. The Pydantic response models now use datetime types instead of str for timestamp fields:

from pydantic import BaseModel
from typing import Any
from datetime import datetime

class TargetCreate(BaseModel):
    url: str

class TargetResponse(BaseModel):
    id: int
    url: str
    created_at: datetime

class EventCreate(BaseModel):
    target_id: int
    payload: dict[str, Any]

class EventResponse(BaseModel):
    id: int
    target_id: int
    payload: dict[str, Any]
    status: str
    created_at: datetime
    attempt_count: int = 0
    next_retry_at: datetime | None = None

class DeliveryAttemptResponse(BaseModel):
    id: int
    event_id: int
    attempted_at: datetime
    status_code: int | None
    success: bool
    error_message: str | None
    attempt_number: int
    next_retry_at: datetime | None

• Open src/hookshot/routes/targets.py and replace the entire file with SQLAlchemy-based queries:

from datetime import datetime, timezone
from fastapi import APIRouter
from sqlalchemy import select
from hookshot.database import get_session, Target
from hookshot.models import TargetCreate, TargetResponse

router = APIRouter()

@router.post("/targets", response_model=TargetResponse)
async def create_target(target: TargetCreate):
    now = datetime.now(timezone.utc)
    async with get_session() as session:
        obj = Target(url=target.url, created_at=now)
        session.add(obj)
        await session.commit()
        await session.refresh(obj)
        return TargetResponse(id=obj.id, url=obj.url, created_at=obj.created_at)

@router.get("/targets", response_model=list[TargetResponse])
async def list_targets():
    async with get_session() as session:
        result = await session.execute(select(Target))
        targets = result.scalars().all()
        return [TargetResponse(id=t.id, url=t.url, created_at=t.created_at) for t in targets]

What changed?

Instead of await db.execute("INSERT INTO targets ...") with ? placeholders, you now create a Target object and call session.add(). SQLAlchemy generates the correct INSERT statement for whatever database engine is configured.

• Verify the targets router loads with the new imports by running:

uv run python -c "from hookshot.routes.targets import router; print('Targets router loaded')"

You should see Targets router loaded printed.

• Open src/hookshot/delivery.py and replace the entire file. Start with the imports and the HTTP delivery logic:

import json
import asyncio
import random
import logging
from datetime import datetime, timezone, timedelta
import httpx
from sqlalchemy import select
from hookshot.database import get_session, Event, DeliveryAttempt
from hookshot.config import settings

logger = logging.getLogger(__name__)

async def _attempt_delivery(event_id: int, target_url: str, payload: dict, attempt_number: int):
    now = datetime.now(timezone.utc)
    status_code = None
    success = False
    error_message = None
    next_retry_at = None

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post(target_url, json=payload)
            status_code = response.status_code
            if response.is_success:
                success = True
                logger.info("Delivery succeeded", extra={"event_id": event_id, "attempt": attempt_number})
    except Exception as e:
        error_message = str(e)
        logger.warning("Delivery failed", extra={"event_id": event_id, "attempt": attempt_number, "error": error_message})

• Still in delivery.py, add the database update block directly below the except block you just pasted. This records the attempt and updates the event status:

    # Calculate backoff delay for failed attempts
    if not success and attempt_number < settings.max_retries:
        jitter = random.uniform(0, 1)
        delay = settings.base_delay * (2 ** attempt_number) + jitter
        next_retry_at = datetime.now(timezone.utc) + timedelta(seconds=delay)

    # Record the delivery attempt in the database
    async with get_session() as session:
        attempt = DeliveryAttempt(
            event_id=event_id,
            attempted_at=now,
            status_code=status_code,
            success=success,
            error_message=error_message,
            attempt_number=attempt_number,
            next_retry_at=next_retry_at,
        )
        session.add(attempt)
        if success:
            new_status = "delivered"
        elif attempt_number < settings.max_retries:
            new_status = "retrying"
        else:
            new_status = "failed"
            logger.error("Max retries reached", extra={"event_id": event_id})
        event_obj = await session.get(Event, event_id)
        event_obj.status = new_status
        await session.commit()

• Finally, add the retry scheduling and helper functions at the bottom of delivery.py:

    # Schedule a retry if the attempt failed and retries remain
    if not success and attempt_number < settings.max_retries:
        jitter = random.uniform(0, 1)
        delay = settings.base_delay * (2 ** attempt_number) + jitter
        asyncio.create_task(_retry_after(delay, event_id, target_url, payload, attempt_number + 1))

async def _retry_after(delay: float, event_id: int, target_url: str, payload: dict, attempt_number: int):
    await asyncio.sleep(delay)
    await _attempt_delivery(event_id, target_url, payload, attempt_number)

async def deliver_event(event_id: int, target_url: str, payload: dict):
    await _attempt_delivery(event_id, target_url, payload, 0)

What changed in the delivery module?

The hardcoded MAX_RETRIES and BASE_DELAY constants are replaced by settings.max_retries and settings.base_delay, pulled from environment variables.

Raw SQL like INSERT INTO delivery_attempts ... VALUES (?, ?, ...) is replaced by creating a DeliveryAttempt object and calling session.add(). The logging calls provide structured output for each delivery attempt.

• Verify the delivery module loads by running:

uv run python -c "from hookshot.delivery import deliver_event; print('Delivery module loaded')"

You should see Delivery module loaded printed.

• Open src/hookshot/routes/webhooks.py and replace the entire file. Start with the imports and the event creation endpoint:

import json
from datetime import datetime, timezone
from fastapi import APIRouter
from sqlalchemy import select, func
from hookshot.database import get_session, Event, Target, DeliveryAttempt
from hookshot.models import EventCreate, EventResponse, DeliveryAttemptResponse
from hookshot.delivery import deliver_event

router = APIRouter()

@router.post("/events", response_model=EventResponse)
async def create_event(event: EventCreate):
    now = datetime.now(timezone.utc)
    async with get_session() as session:
        target = await session.get(Target, event.target_id)
        obj = Event(target_id=event.target_id, payload=json.dumps(event.payload), status="pending", created_at=now)
        session.add(obj)
        await session.commit()
        await session.refresh(obj)
        event_id = obj.id
        target_url = target.url
    await deliver_event(event_id, target_url, event.payload)
    async with get_session() as session:
        obj = await session.get(Event, event_id)
        count = (await session.execute(select(func.count()).where(DeliveryAttempt.event_id == event_id))).scalar()
        last_attempt = (await session.execute(select(DeliveryAttempt).where(DeliveryAttempt.event_id == event_id).order_by(DeliveryAttempt.attempt_number.desc()).limit(1))).scalar_one_or_none()
        return EventResponse(id=obj.id, target_id=obj.target_id, payload=json.loads(obj.payload), status=obj.status, created_at=obj.created_at, attempt_count=count, next_retry_at=last_attempt.next_retry_at if last_attempt else None)

• Add the list endpoints directly below the create_event function in the same file:

@router.get("/events", response_model=list[EventResponse])
async def list_events():
    async with get_session() as session:
        events = (await session.execute(select(Event))).scalars().all()
        result = []
        for e in events:
            count = (await session.execute(select(func.count()).where(DeliveryAttempt.event_id == e.id))).scalar()
            last_attempt = (await session.execute(select(DeliveryAttempt).where(DeliveryAttempt.event_id == e.id).order_by(DeliveryAttempt.attempt_number.desc()).limit(1))).scalar_one_or_none()
            result.append(EventResponse(id=e.id, target_id=e.target_id, payload=json.loads(e.payload), status=e.status, created_at=e.created_at, attempt_count=count, next_retry_at=last_attempt.next_retry_at if last_attempt else None))
        return result

@router.get("/events/{event_id}/attempts", response_model=list[DeliveryAttemptResponse])
async def list_attempts(event_id: int):
    async with get_session() as session:
        attempts = (await session.execute(select(DeliveryAttempt).where(DeliveryAttempt.event_id == event_id).order_by(DeliveryAttempt.attempt_number))).scalars().all()
        return [DeliveryAttemptResponse(id=a.id, event_id=a.event_id, attempted_at=a.attempted_at, status_code=a.status_code, success=a.success, error_message=a.error_message, attempt_number=a.attempt_number, next_retry_at=a.next_retry_at) for a in attempts]

• Now open src/hookshot/main.py and replace the entire file. This adds structured logging, imports the new config, and wires up the health check:

import logging
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy import text
from hookshot.database import init_db, engine
from hookshot.config import settings
from hookshot.routes import targets, webhooks

# Configure logging format based on environment
if settings.environment == "production":
    logging.basicConfig(level=settings.log_level, format='{"time": "%(asctime)s", "level": "%(levelname)s", "message": "%(message)s"}')
else:
    logging.basicConfig(level=settings.log_level, format="%(asctime)s %(levelname)s %(name)s %(message)s")

logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    logger.info("Database initialized")
    yield

app = FastAPI(title="Hookshot", lifespan=lifespan)

app.include_router(targets.router, prefix="/api")
app.include_router(webhooks.router, prefix="/api")

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

• Still in main.py, add the health check endpoint at the bottom of the file:

@app.get("/health")
async def health():
    try:
        async with engine.connect() as conn:
            await conn.execute(text("select 1"))
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        return {"status": "unhealthy", "database": str(e)}

What does the health endpoint do?

The /health endpoint runs select 1 through the async engine to prove the database connection is alive. Deployment platforms like Render poll this endpoint to know if your service is ready to receive traffic.

• Save all files. Start the dev server by running:

uv run uvicorn hookshot.main:app --reload

Before you check the health endpoint, what do you expect it to return? Think about what the select 1 query proves.

• Open http://localhost:8000/health in your browser.

You should see {"status": "healthy", "database": "connected"}. The entire refactor works. SQLAlchemy created a fresh hookshot.db with the ORM-defined schema.

• Open http://localhost:8000/docs and test the full flow: register a target, send an event, and check GET /api/events.
• Stop the dev server with Ctrl+C.

Seeing a module import error on startup?

Every file that previously imported from hookshot.database import get_db must now import from hookshot.database import get_session. The function was renamed during the refactor. Check targets.py, webhooks.py, and delivery.py for any leftover get_db references.

Help me fix import errors after refactoring hookshot to SQLAlchemy.

Write tests and verify linting

With the refactor complete, you need automated tests to catch regressions before deploying.

• Create a tests/ directory in the root of your hookshot project folder (same level as src/) by running:

mkdir -p tests

• Create tests/test_api.py with tests for the health check, target creation, and event creation:

import pytest
from fastapi.testclient import TestClient
from hookshot.main import app

client = TestClient(app)

def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

def test_create_target():
    response = client.post("/api/targets", json={"url": "https://example.com/webhook"})
    assert response.status_code == 200
    data = response.json()
    assert data["url"] == "https://example.com/webhook"
    assert "id" in data

def test_create_event():
    target = client.post("/api/targets", json={"url": "https://httpbin.org/post"}).json()
    response = client.post("/api/events", json={"target_id": target["id"], "payload": {"hello": "world"}})
    assert response.status_code == 200
    data = response.json()
    assert data["target_id"] == target["id"]
    assert data["payload"] == {"hello": "world"}

What does this code do?

FastAPI's TestClient wraps httpx under the hood to make synchronous test requests against your async app. Each test function exercises one endpoint and asserts on the response status code and body.

• Run the test suite by running:

uv run pytest tests/ -v

You should see all three tests pass with green output.

• Verify your code has no linting issues by running:

uv run ruff check src/

Ruff should report no errors.

Tests failing with database errors?

Delete hookshot.db and run the tests again. The TestClient triggers the lifespan handler which calls init_db() to create a fresh database. A leftover database from a previous schema can cause column mismatch errors.

Help me debug failing pytest tests in my hookshot project.

Create the Dockerfile and containerize

The final piece is packaging the service into a Docker container so it can be deployed anywhere.

• Create a Dockerfile in the root of the hookshot project folder:

# Stage 1: build
FROM python:3.11-slim AS builder
WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

# Stage 2: runtime
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /app/.venv .venv
COPY src/ src/
ENV PATH="/app/.venv/bin:$PATH"
CMD ["uvicorn", "hookshot.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

This is a multi-stage build. The first stage installs uv and runs uv sync --frozen --no-dev to install only production dependencies into a virtual environment. The --frozen flag uses the lockfile without checking for updates. The --no-dev flag skips dev dependencies like pytest and ruff.

The second stage copies just the virtual environment and source code into a clean python:3.11-slim image. This keeps the final image small because build tools like uv are not included.

• Create a .dockerignore file to keep unnecessary files out of the build context:

.venv/
.git/
__pycache__/
.env
*.pyc
hookshot.db

• Build the Docker image by running:

docker build -t hookshot .

The build will take a minute as uv installs all production dependencies inside the container.

• Run the container by running:

docker run -p 8000:8000 hookshot

Before you check the health endpoint, what do you expect the database status to be? Remember, no .env file is in the container (excluded by .dockerignore), so the app uses the default DATABASE_URL from config.py.

• Open http://localhost:8000/health in your browser.

You should see {"status": "healthy", "database": "connected"}. The containerized service is running on SQLite locally, just like outside the container.

• Stop the container with Ctrl+C.

Docker build failing?

Make sure uv.lock exists in your project root. This file is generated when you run uv add or uv sync. The Dockerfile copies it into the builder stage. Without it, uv sync --frozen has nothing to lock from.

Help me debug a Docker build failure for my hookshot project.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files before moving on.

ⓧ I'd like to double check the full code

This step touched many files. Scroll back through the step and confirm each chunk is present in order. Here is a summary of the files you changed:

• config.py has the Settings class with BaseSettings and five fields.
• database.py has the engine, session factory, Base, three ORM models (Target, Event, DeliveryAttempt), init_db(), and get_session().
• models.py uses datetime types for all timestamp fields.
• routes/targets.py uses get_session() and select(Target).
• routes/webhooks.py uses get_session(), session.get(), and select() queries.
• delivery.py uses settings.max_retries, settings.base_delay, SQLAlchemy sessions, and logging.
• main.py has structured logging, the /health endpoint, and imports engine from database.
• tests/test_api.py has three tests using TestClient.
• Dockerfile is a multi-stage build with uv sync --frozen --no-dev.
• .dockerignore, .env, and .gitignore are all in the project root.

Your service is now production-hardened. It has environment-based config, a portable database layer, structured logging, passing tests, and a Docker image. Next up, you will deploy it to a real server with a real PostgreSQL database.

Deploy to Render with Postgres

Your webhook service is production-hardened, tested, and containerized. This step takes it from your laptop to a live public URL on Render.

Here is where SQLite hits its production limit. Render's free web services use an ephemeral filesystem, so every deploy or restart wipes the container's disk along with your SQLite database.

The SQLAlchemy refactor from the previous step makes the fix trivial. You will create a PostgreSQL instance on Render, point DATABASE_URL at it, and deploy. Zero code changes required.

In this step, get ready to:

• Push your code to a GitHub repository.
• Create a Render Postgres database and deploy the web service.
• Verify your data persists across redeploys.

Create a Render Postgres database and push to GitHub

Your service needs two things on Render: a Postgres database for persistent storage, and a web service to run your Docker container. You will create the database first so you have the connection URL ready when you configure the web service.

• Open your browser and go to dashboard.render.com/register to sign up for a free Render account.
• You can sign up with your GitHub account to speed things up.

What is Render?

Render is a cloud platform that hosts web services, databases, and static sites. It has a genuine free tier that includes web services (512 MB RAM, 0.1 CPU) and a Postgres database. No credit card required.

• Once signed in, click the + New button in the top-right corner of the Render Dashboard.
• Select Postgres from the menu.
• Enter hookshot-db in the Name field.
• Scroll down and select the Free instance type.
• Click Create Database.

The database's status updates to Available when it is ready. This takes about 30 seconds.

• On the database's page, click the Connect button in the top-right corner.
• Copy the Internal Database URL. Save it somewhere safe. You will need it in the next substep.

Your Postgres database expires in 30 days

Free Render Postgres databases expire 30 days after creation. This is fine for a learning project. Render will email you a reminder before it expires.

You can upgrade to a paid instance to keep your data, or delete and recreate for another 30 days.

Why change the URL scheme?

The Internal Database URL starts with postgres://. SQLAlchemy's async engine requires the postgresql+asyncpg:// scheme to know which driver to use.

When you set the DATABASE_URL environment variable on Render in the next substep, you will replace the postgres:// prefix with postgresql+asyncpg://.

Now push your project to GitHub so Render can pull the code.

• Make sure your terminal is in the hookshot folder.
• Initialize a git repository and create your first commit by running these commands:

git init
git add .
git commit -m "Initial commit"

You should see a summary of files committed, including Dockerfile, pyproject.toml, and the src/ directory.

• Go to github.com/new in your browser.
• Enter hookshot as the repository name.
• Leave all other fields as defaults (no README, no .gitignore). Your project already has these.
• Click Create repository.
• GitHub shows setup instructions. Copy the HTTPS URL for your new repository (it looks like https://github.com/your-username/hookshot.git).

• Connect your local repository to GitHub and push by running these commands:

git branch -M main
git remote add origin [[GITHUB_REPO_URL="https://github.com/your-username/hookshot.git"]]
git push -u origin main

You should see output ending with a success message showing your branch was pushed.

Push rejected or permission denied?

• Make sure you created the repository on GitHub first and that it is empty (no README).
• If you see a credentials prompt, GitHub requires a Personal Access Token instead of your password. Follow GitHub's token guide to create one.

Help me push my hookshot project to GitHub

• Visit your repository page on GitHub and confirm you see Dockerfile, pyproject.toml, uv.lock, src/, and tests/ listed.

Deploy the web service on Render

With your database running and your code on GitHub, you can now create the web service that pulls your Dockerfile and deploys it.

• In the Render Dashboard, click + New in the top-right corner.
• Select Web Service.
• Select Git Provider and connect your GitHub account if prompted.
• Find and select the hookshot repository from the list.
• Set Language to Docker. This tells Render to build from your Dockerfile instead of using a native runtime.
• Select the Free instance type (512 MB RAM / 0.1 CPU).

Now add the environment variables that configure your service for production.

• Scroll down to the Advanced section and expand it.
• Click Add Environment Variable.
• Add the first variable: set the key to ENVIRONMENT and the value to production.
• Add a second variable: set the key to LOG_LEVEL and the value to INFO.
• Add a third variable: set the key to DATABASE_URL.
• For the value, paste the Internal Database URL you copied earlier. Then change the postgres:// prefix at the start to postgresql+asyncpg://. The rest of the URL stays the same.

What do the environment variables do?

• ENVIRONMENT=production switches logging to JSON format, which is easier for log aggregation tools to parse.
• LOG_LEVEL=INFO keeps log output clean without debug noise.
• DATABASE_URL tells SQLAlchemy to connect to Postgres instead of SQLite. Your config.py reads this via pydantic-settings and passes it to create_async_engine.

• Click Create Web Service.

Render kicks off your service's first build. You can watch the build log on the Events page. The Docker build takes 2-5 minutes.

Free services spin down when idle

Render's free web services spin down after 15 minutes without traffic. The first request after a spin-down triggers a cold start that takes a few seconds. This is normal.

Verify the live service

Your service is deployed. Time to prove it works against a real Postgres database.

Before you open the URL, predict: will the health check pass, and which database backend will it be connected to?

• Find your service URL at the top of the service page in the Render Dashboard. It looks like https://hookshot-xxxx.onrender.com.
• Open your-service-url/health in your browser.

You should see:

{"status": "healthy", "database": "connected"}

The service is alive and connected to Postgres. The same /health endpoint that checked SQLite locally now checks Postgres in production. Zero code changes.

Seeing an unhealthy response or a timeout?

• If the page takes a long time to load, the free service may be cold-starting. Wait 10-15 seconds and refresh.
• If the database shows unhealthy, double-check that your DATABASE_URL environment variable starts with postgresql+asyncpg:// and not postgres://.
• Make sure your Postgres database and web service are in the same Render region.

Help me troubleshoot my Render deployment

Now test the full API through the live service.

• Open your-service-url/docs in your browser to access the Swagger UI on the live service.
• Use the POST /api/targets endpoint to register a target with URL https://httpbin.org/post.
• Use the POST /api/events endpoint to send an event to that target. Use target_id from the response above and any JSON payload.
• Use the GET /api/events endpoint to verify the event was persisted and the delivery status shows delivered.

Your webhook service is processing events through the live Postgres database.

The final test proves that Postgres survives what SQLite cannot: a container restart.

• In the Render Dashboard, click on your hookshot web service to open its page.
• On the Events page, open the Manual Deploy dropdown.
• Select Deploy latest commit. This tears down the running container and builds a fresh one.
• Wait for the new deployment to finish (watch the build log).

Before you check the data, predict: will your targets and events still be there after the container was replaced?

• Once the deploy finishes, go back to your-service-url/docs and hit GET /api/targets.
• Hit GET /api/events.

Your targets and events are still there. The container was completely replaced, but the data lives in Postgres, which is an independent service. This is exactly why you refactored from SQLite to SQLAlchemy with a configurable DATABASE_URL.

Secret mission

Add HMAC-SHA256 Webhook Signatures

Right now, any target receiving your webhooks has no way to verify they actually came from your service. An attacker could spoof payloads and your targets would never know. Can you add cryptographic signatures so targets can prove every event is authentic?

Clean Up Your Resources

Clean Up Your Resources

This project uses Render's free tier for both the web service and Postgres database, so there are no immediate costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Cost warning

Your free Render Postgres database expires 30 days after creation. Render sends an email reminder before expiry. If you want to keep your data beyond 30 days, you would need to upgrade to a paid instance.

Resources you used:

• Render Free Web Service (runs your hookshot API, auto-sleeps after 15 minutes of inactivity).
• Render Free PostgreSQL database (hookshot-db). Stores targets, events, and delivery attempts. Expires 30 days after creation.
• Local hookshot project folder on your Desktop.
• Local hookshot Docker image.
• GitHub repository containing the hookshot source code.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the hookshot service.

• Your Render web service costs nothing on the free tier and sleeps when idle. It auto-starts on the next incoming request.
• Your Render Postgres database is free for 30 days. Render sends an email reminder before it expires.
• Your local hookshot folder, Docker image, and GitHub repo remain on your machine and account.
• uv remains installed on your machine. You can reuse it for future Python projects.

Pause - I'll come back to this later

Shut down running processes to free up local resources while keeping your files and Render services intact.

• Stop the local dev server if it is still running by pressing Ctrl+C in the terminal.
• Your Render web service automatically sleeps after 15 minutes of inactivity. No action needed.
• Your Render Postgres database remains available until it expires (30 days after creation).
• Your local hookshot folder, Docker image, and GitHub repo stay untouched for when you return.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each action below.

• In the Render Dashboard, click your hookshot web service.
• Click Settings in the left sidebar.
• Scroll to the bottom and click Delete Web Service.

• In the Render Dashboard, click your hookshot-db Postgres database.
• Click Settings in the left sidebar.
• Scroll to the bottom and click Delete Database.

• On GitHub, open your hookshot repository.
• Click Settings.
• Scroll to the bottom, find the Danger Zone section, and click Delete this repository.

• Remove the local Docker image by running:

docker rmi hookshot

• Delete the local project folder.

macOS

• Run this command to remove the folder:

rm -rf ~/Desktop/hookshot

Windows

• Run this command to remove the folder:

rmdir /s /q %USERPROFILE%\Desktop\hookshot

Nice Work!

Nice Work!

Nice work! You just built and deployed a production webhook delivery service using FastAPI, SQLAlchemy, and Docker.

You've learned how to:

• Built a webhook delivery service with FastAPI that accepts events, persists them, and delivers them to target URLs with automatic retries and exponential backoff.
• Refactored raw SQL to SQLAlchemy 2.0 async, making the same codebase run on both SQLite locally and PostgreSQL in production with zero code changes.
• Containerized the service with a multi-stage Dockerfile and deployed it to Render with persistent Postgres storage that survives container restarts and redeploys.
• Completed an optional Secret Mission to push your skills even further.

Ready to quiz yourself?