Build a Webhook Delivery Service

Introduction

30 Second Summary

Every time you place an order online, your payment provider sends a notification to the store's server. That notification is a webhook. Behind the scenes, a delivery service decides which servers should receive which events and routes them accordingly.

In this project, you will build a webhook delivery service API from scratch. You'll create authenticated endpoints for registering URLs, subscribing them to event types, and ingesting events that get routed to the correct subscribers.

What You'll Build

A fully containerized REST API built with FastAPI, PostgreSQL, and Docker that manages webhook endpoint registration, event subscriptions, and event ingestion.

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

• A JWT-authenticated API with user registration, login, and protected routes for managing webhooks.
• A relational data model with endpoints, subscriptions, and events connected by foreign keys and isolated per user.
• A Docker Compose setup running the app and database together, with an automated test suite proving everything works.
• Secret Mission: Add HMAC signature verification so consumers can verify webhook payloads are authentic.

Are there any prerequisites?

You should be comfortable with Python (type hints, async/await, and decorators). You'll also need Docker, a code editor, and Git already installed on your machine.

Scaffold the Project and Docker Compose

Every production API needs a clear folder structure and a reproducible environment. Without these, onboarding new developers or deploying to a server becomes a guessing game.

You'll use Docker Compose to run two containers side by side: your FastAPI application and a PostgreSQL database. By the end of this step, your API skeleton will be live and responding to requests.

In this step, get ready to:

• Create the layered directory structure for the API.
• Write the Dockerfile, Docker Compose config, and dependency list.
• Build and run the containers to verify everything connects.

Create the project directory structure

Production Python APIs use a layered architecture. Each layer (routers, models, schemas, services) lives in its own folder so responsibilities stay separated and the codebase scales cleanly.

• Open your terminal.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create the project folder, move into it, and set up the directory structure by running these commands:

mkdir webhook-service
cd webhook-service
mkdir -p app/core app/models app/routers app/schemas app/services
touch app/__init__.py app/core/__init__.py app/models/__init__.py app/routers/__init__.py app/schemas/__init__.py app/services/__init__.py

• Confirm the directories were created by running:

ls app/

You should see __init__.py, core, models, routers, schemas, and services in the output.

Don't see all the folders?

Make sure you ran the mkdir -p command from inside the webhook-service folder. The -p flag creates parent directories as needed, so if it was omitted you may get an error.

Help me troubleshoot my directory structure

Why a layered architecture?

Each folder has a single job. Routers handle HTTP request/response logic. Models define database tables. Schemas define request/response shapes. Services hold business logic. Core holds shared config.

This separation makes each piece testable in isolation. It also matches what hiring managers expect to see in production Python codebases.

• Open your code editor.
• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type VS Code (or Cursor) and press Enter to open it.
• Click File in the top menu bar.
• Click Open Folder.
• Navigate to your Desktop, select the webhook-service folder, and click Open.

• Create a new file called app/main.py and add the following code:

from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI(title='Webhook Delivery Service')

# Basic health check endpoint to verify the app is running
@app.get('/health')
async def health():
    return {'status': 'ok'}

What does this code do?

• FastAPI(title='Webhook Delivery Service') creates the app instance. The title appears in the auto-generated API documentation.
• @app.get('/health') registers a GET endpoint at /health. This is a standard pattern for load balancers and monitoring tools to check if the service is alive.

• Save app/main.py.

You won't see any output yet. This file runs when Docker starts the container in a later substep.

• Create a new file called app/core/config.py and add the following code:

from pydantic_settings import BaseSettings

# Application settings loaded from environment variables
class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str

settings = Settings()

What does this code do?

• BaseSettings from pydantic-settings automatically reads values from environment variables. You never hardcode secrets in your source code.
• DATABASE_URL and SECRET_KEY are declared as required strings. If either is missing from the environment, the app refuses to start with a clear error.

• Save app/core/config.py.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

Here is app/main.py:

from fastapi import FastAPI

app = FastAPI(title='Webhook Delivery Service')

@app.get('/health')
async def health():
    return {'status': 'ok'}

Here is app/core/config.py:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str

settings = Settings()

Write the Dockerfile, Compose config, and requirements

Your app code is ready, but it needs an environment to run in. You will define three files: what packages to install, how to build the container image, and how to orchestrate the containers together.

• Create a new file called requirements.txt in the project root (the webhook-service/ folder) and add the following:

fastapi
uvicorn[standard]
sqlalchemy[asyncio]
asyncpg
alembic
pyjwt
pwdlib[argon2]
pytest
httpx
aiosqlite
python-multipart

What are these packages?

• fastapi is the web framework. uvicorn[standard] is the ASGI server that runs it.
• sqlalchemy[asyncio] and asyncpg provide async database access. alembic handles database migrations.
• pyjwt creates and verifies JSON Web Tokens. pwdlib[argon2] hashes passwords securely with the Argon2 algorithm.
• pytest and httpx are for automated testing. aiosqlite provides a lightweight async database for tests. python-multipart supports form data in login requests.

• Save requirements.txt.

• Create a new file called Dockerfile in the project root and add the following:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

What does this Dockerfile do?

• FROM python:3.12-slim starts from a lightweight Python image. The slim variant keeps the image small by excluding unnecessary system packages.
• Requirements are copied and installed BEFORE the app code. This means Docker caches the dependency layer and only re-installs packages when requirements.txt changes.
• --reload makes Uvicorn restart whenever you save a file. This is for development only.

• Save Dockerfile.

• Create a new file called docker-compose.yml in the project root and add the following:

version: '3.9'
services:
  api:
    build: .
    ports:
      - '8000:8000'
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/webhooks
      SECRET_KEY: supersecretkey
    depends_on:
      - db
    volumes:
      - .:/app
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: webhooks
    volumes:
      - pgdata:/var/lib/postgresql/data
volumes:
  pgdata:

What does this Compose file do?

• The api service builds from your Dockerfile and maps port 8000 to your machine. The volumes: .:/app mount means file changes on your machine instantly appear inside the container.
• The db service runs postgres:16-alpine, a lightweight PostgreSQL image. The pgdata volume persists database data even if the container restarts.
• Environment variables in the api service feed directly into your Settings class via pydantic-settings. No hardcoded values in your Python code.

• Save docker-compose.yml.

YAML indentation matters

YAML uses spaces for structure (never tabs). Each nested level must be indented by exactly 2 spaces. If Docker Compose throws a parsing error, check that your indentation is consistent throughout the file.

Help me fix my docker-compose.yml syntax

✔️ Awesome, I've got everything!

Great. Double check that all three files are saved in the project root (not inside the app/ folder).

ⓧ I'd like to double check the full code

Here is requirements.txt:

fastapi
uvicorn[standard]
sqlalchemy[asyncio]
asyncpg
alembic
pyjwt
pwdlib[argon2]
pytest
httpx
aiosqlite
python-multipart

Here is Dockerfile:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]

Here is docker-compose.yml:

version: '3.9'
services:
  api:
    build: .
    ports:
      - '8000:8000'
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/webhooks
      SECRET_KEY: supersecretkey
    depends_on:
      - db
    volumes:
      - .:/app
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: webhooks
    volumes:
      - pgdata:/var/lib/postgresql/data
volumes:
  pgdata:

Build and run with Docker Compose

All the pieces are in place. Time to build the container image and start both services.

• In your terminal, make sure you are inside the webhook-service folder.
• Build and start the containers by running this command:

docker compose up --build

What does this command do?

docker compose up starts all the services defined in your docker-compose.yml. The --build flag forces Docker to rebuild the image from scratch, which is important the first time or whenever you change your Dockerfile or requirements.

The first build downloads the Python and PostgreSQL images, so it may take 1-2 minutes depending on your internet speed. Subsequent builds are much faster because Docker caches layers.

You should see logs from both the api and db containers scrolling in your terminal. Look for a line containing:

Uvicorn running on http://0.0.0.0:8000

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

You should see the Swagger UI page with the title "Webhook Delivery Service" and one endpoint listed: GET /health.

• Click the GET /health row to expand it.
• Click Try it out.
• Click Execute.

The response body should show {"status": "ok"} with a 200 status code. Your API is running.

Page not loading at localhost:8000?

• Check your terminal for error messages. If you see a Python import error, make sure all files are saved and that your Dockerfile and requirements.txt are in the project root (not inside the app/ folder).
• If port 8000 is already in use, you will see an address-binding error. Stop whatever is using that port or change the port mapping in docker-compose.yml.
• Make sure Docker Desktop is running. The docker compose command requires the Docker daemon to be active.

Help me debug my Docker Compose setup

Your containers are running and the API skeleton is live. Next up, you will define the database models and run your first migration.

Define Database Models and Run Migrations

Your FastAPI app is running, but it has no way to store data. Every webhook that arrives, every user who registers, every subscription that gets created needs a permanent home in a database.

In this step, you will define SQLAlchemy models that map Python classes to PostgreSQL tables. Then you will configure Alembic to manage database migrations and run the first one to create your tables inside the Docker container.

In this step, get ready to:

• Define SQLAlchemy models for users, endpoints, subscriptions, and events.
• Configure the async database engine and session dependency.
• Initialize Alembic and run your first migration.

Define the SQLAlchemy models and database connection

SQLAlchemy 2.0 uses a declarative base class that all your models inherit from. Each model class maps to a table in PostgreSQL.

• Create a new file called base.py inside the app/models/ folder.
• Add the declarative base class by pasting this code:

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

• Create a new file called user.py inside app/models/ and paste this code:

from sqlalchemy import Column, Integer, String, DateTime, func
from app.models.base import Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

What does this code do?

The Base class is the foundation all your models inherit from. SQLAlchemy uses it to track which tables exist and what columns they have.

The User model maps to a users table. Each Column defines a database column with its type and constraints. The server_default=func.now() tells PostgreSQL to automatically set the timestamp when a row is inserted.

• Save both files.
• In your editor's sidebar, expand the app/models/ folder. You should see base.py and user.py listed alongside the existing __init__.py.

Now create the remaining three models. These introduce foreign keys that reference the users table you just defined.

• Create app/models/endpoint.py and paste this code:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
from app.models.base import Base

class Endpoint(Base):
    __tablename__ = 'endpoints'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    url = Column(String, nullable=False)
    description = Column(String)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

• Create app/models/subscription.py and paste this code:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
from app.models.base import Base

class Subscription(Base):
    __tablename__ = 'subscriptions'
    id = Column(Integer, primary_key=True, index=True)
    endpoint_id = Column(Integer, ForeignKey('endpoints.id'), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    event_type = Column(String, nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

• Create app/models/event.py and paste this code:

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, JSON, func
from app.models.base import Base

class Event(Base):
    __tablename__ = 'events'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    subscription_id = Column(Integer, ForeignKey('subscriptions.id'), nullable=True)
    event_type = Column(String, nullable=False, index=True)
    payload = Column(JSON, nullable=False)
    status = Column(String, default='pending', nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

What do foreign keys and JSON columns do?

The ForeignKey('users.id') constraint links each endpoint, subscription, and event back to its owning user. PostgreSQL enforces this relationship so you cannot create an endpoint for a user that does not exist.

The Event model uses a JSON column for the payload. This stores arbitrary webhook data (like an order object) without needing a fixed schema. The status field defaults to 'pending' so new events are automatically queued for delivery.

• Save all three files.
• Confirm your app/models/ folder now contains five files: __init__.py, base.py, user.py, endpoint.py, subscription.py, and event.py.

Not seeing the new files?

Make sure you created them inside app/models/ and not in the project root. Each file must import Base from app.models.base. If you accidentally placed them elsewhere, move them into the correct folder.

Help me debug my model file structure.

Next, create the database connection module that FastAPI will use to talk to PostgreSQL.

• Create a new file called database.py inside app/core/ and paste this code:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import settings

# Create the async engine using the DATABASE_URL from environment
engine = create_async_engine(settings.DATABASE_URL, echo=True)

# Session factory for creating database sessions
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    # Yield a session for each request, auto-close when done
    async with AsyncSessionLocal() as session:
        yield session

What does this code do?

The create_async_engine function creates an async connection pool to PostgreSQL using the asyncpg driver. The echo=True flag logs all SQL statements to your terminal, which helps with debugging during development.

The get_db function is a FastAPI dependency. Every route that needs database access will receive a fresh session through dependency injection. The async with block automatically closes the session when the request finishes.

• Save database.py.
• Check that app/core/ now contains __init__.py, config.py, and database.py.

✔️ Awesome, I've got everything!

Great. Make sure all your files are saved before moving on.

ⓧ I'd like to double check the full code

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

from sqlalchemy import Column, Integer, String, DateTime, func
from app.models.base import Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
from app.models.base import Base

class Endpoint(Base):
    __tablename__ = 'endpoints'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    url = Column(String, nullable=False)
    description = Column(String)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
from app.models.base import Base

class Subscription(Base):
    __tablename__ = 'subscriptions'
    id = Column(Integer, primary_key=True, index=True)
    endpoint_id = Column(Integer, ForeignKey('endpoints.id'), nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    event_type = Column(String, nullable=False, index=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, JSON, func
from app.models.base import Base

class Event(Base):
    __tablename__ = 'events'
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    subscription_id = Column(Integer, ForeignKey('subscriptions.id'), nullable=True)
    event_type = Column(String, nullable=False, index=True)
    payload = Column(JSON, nullable=False)
    status = Column(String, default='pending', nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Initialize Alembic and generate the migration

Alembic is a database migration tool that tracks changes to your models and applies them to the database as versioned scripts. Because this project uses an async database driver, you need the async template.

• Initialize Alembic with the async template by running this command inside your Docker container:

docker compose exec api alembic init -t async alembic

What does this command do?

The docker compose exec api part runs a command inside the running api container. The alembic init -t async alembic initializes a migration environment using the async template. It creates an alembic.ini config file and an alembic/ directory with env.py and a versions/ folder for migration scripts.

• Confirm the files were created by running:

ls alembic/

• You should see env.py, script.py.mako, and a versions directory in the output.

Now you need to configure env.py to use your application's database URL and model metadata. The generated template has placeholder values that you will replace.

• Open alembic/env.py in your editor and replace its entire contents with this code:

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

from app.core.config import settings
from app.models.base import Base

# Import all models so they register with Base.metadata
from app.models.user import User
from app.models.endpoint import Endpoint
from app.models.subscription import Subscription
from app.models.event import Event

# Alembic Config object for access to .ini values
config = context.config

# Set the SQLAlchemy URL from our app settings
config.set_main_option('sqlalchemy.url', settings.DATABASE_URL)

# Set up Python logging from the config file
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Point Alembic at our models' metadata for autogeneration
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """Run migrations in offline mode (generates SQL without connecting)."""
    url = config.get_main_option('sqlalchemy.url')
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={'paramstyle': 'named'},
    )
    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection) -> None:
    """Configure context and run migrations with a given connection."""
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """Create an async engine and run migrations."""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix='sqlalchemy.',
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in online mode using async engine."""
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

What is happening in this file?

This file tells Alembic how to connect to the database and which tables to track. The config.set_main_option('sqlalchemy.url', settings.DATABASE_URL) line pulls the database URL from your app's settings instead of hardcoding it in alembic.ini.

The model imports at the top register each table with Base.metadata. Without these imports, Alembic would not detect any tables to create.

The run_async_migrations function creates an async engine, opens a connection, and runs migrations through connection.run_sync. This bridges the async engine with Alembic's synchronous migration runner.

• Save alembic/env.py.

Now generate the initial migration. Alembic will compare your models against the empty database and produce a script that creates all four tables.

• Generate the migration by running:

docker compose exec api alembic revision --autogenerate -m "initial"

• You should see output mentioning Generating /app/alembic/versions/...initial.py with detected table operations for users, endpoints, subscriptions, and events.

Seeing a ModuleNotFoundError?

If you see ModuleNotFoundError: No module named 'app', make sure your volume mount in docker-compose.yml maps the current directory to /app inside the container. Also verify the container is running with docker compose ps.

If you see errors about the database URL or connection, confirm that both the api and db containers are up.

Help me fix my Alembic autogenerate error.

Run the migration and verify database connectivity

The migration script exists but has not been applied yet. Running it will create the actual tables in PostgreSQL.

• Apply the migration by running:

docker compose exec api alembic upgrade head

• You should see output ending with lines like Running upgrade -> ..., initial confirming the migration was applied.

Migration failed with a connection error?

If you see a connection refused error, the PostgreSQL container might not be ready. Run docker compose ps and check that the db service is healthy. You can also restart both containers with docker compose restart and try the upgrade command again.

Help me troubleshoot my Alembic migration connection error.

Now update the health endpoint to prove the app can actually query the database. This makes /health a real connectivity check instead of just returning a static response.

• Open app/main.py and replace its entire contents with this code:

from fastapi import FastAPI
from sqlalchemy import text
from app.core.database import get_db
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI(title='Webhook Delivery Service')

@app.get('/health')
async def health(db: AsyncSession = Depends(get_db)):
    # Verify the database connection is alive
    await db.execute(text('SELECT 1'))
    return {'status': 'ok'}

What changed?

The /health endpoint now injects a database session using Depends(get_db). It runs SELECT 1 against PostgreSQL before returning. If the database is unreachable, this endpoint will fail with a 500 error instead of falsely reporting everything is fine.

• Save app/main.py.
• Test the health endpoint by running:

curl http://localhost:8000/health

• You should see {"status":"ok"} in the response. This confirms your app is successfully querying PostgreSQL through the async engine.

Getting a 500 error on /health?

Check the container logs with docker compose logs api for the full traceback. Common causes include a typo in the import path for get_db or a missing database.py file. Make sure app/core/database.py exists and imports settings from app.core.config.

Help me fix the 500 error on my /health endpoint.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

from fastapi import FastAPI
from sqlalchemy import text
from app.core.database import get_db
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI(title='Webhook Delivery Service')

@app.get('/health')
async def health(db: AsyncSession = Depends(get_db)):
    await db.execute(text('SELECT 1'))
    return {'status': 'ok'}

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Your database layer is ready. You have models, migrations, and a proven connection. Next up, you will build the authentication system so users can register, log in, and receive JWT tokens to access protected endpoints.

Implement JWT Authentication

Your database tables are in place and the health endpoint proves the app can talk to PostgreSQL. Now it's time to lock the front door.

Every route you build from here on will need to know who is making the request. In this step, you'll add user registration, login, and a reusable authentication dependency that protects every future route with a JWT token.

In this step, get ready to:

• Build password hashing and token utilities with pwdlib and PyJWT.
• Create registration and login endpoints that return a JWT access token.
• Implement a reusable dependency that validates tokens on every protected request.

Implement user registration

Before you can register users, the app needs security utilities (password hashing, token creation) and Pydantic schemas that define what data flows in and out of the auth endpoints.

• In your editor, open app/core/config.py.
• Find these lines at the end of the Settings class:

    DATABASE_URL: str
    SECRET_KEY: str

• Add two new fields below SECRET_KEY, so the class now looks like this:

    DATABASE_URL: str
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ALGORITHM: str = 'HS256'

What do these settings control?

ACCESS_TOKEN_EXPIRE_MINUTES sets how long a login token stays valid (30 minutes).

ALGORITHM specifies the JWT signing algorithm. HS256 is a symmetric HMAC that uses your SECRET_KEY to both sign and verify tokens.

• Create a new file app/core/security.py and add the following code:

import jwt
from datetime import datetime, timedelta, timezone
from pwdlib import PasswordHash
from app.core.config import settings

# Initialize Argon2 password hasher (recommended by OWASP)
pwd_context = PasswordHash.recommended()

def hash_password(password: str) -> str:
    # One-way hash a plaintext password for safe storage
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    # Check a plaintext password against its stored hash
    return pwd_context.verify(plain, hashed)

def create_access_token(subject: int) -> str:
    # Build a JWT with the user's ID and an expiration timestamp
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {'sub': str(subject), 'exp': expire}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def decode_access_token(token: str) -> int:
    # Verify the token signature and extract the user ID
    payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
    return int(payload['sub'])

What does this code do?

PasswordHash.recommended() creates an Argon2 hasher. Argon2 is the current industry-standard algorithm for password hashing because it's resistant to GPU-based attacks.

hash_password and verify_password wrap the hasher so the rest of your code never touches the hashing library directly.

create_access_token encodes the user's database ID into a signed JWT. decode_access_token reverses the process, returning the user ID if the signature is valid.

• Save app/core/security.py.

You won't see a change yet. These utilities get used by the auth router you'll create next.

• Create a new file app/schemas/user.py and add the following code:

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    # Request body for registration
    email: EmailStr
    password: str

class UserRead(BaseModel):
    # Response body after registration (no password exposed)
    id: int
    email: str

    class Config:
        from_attributes = True

class Token(BaseModel):
    # Response body after login
    access_token: str
    token_type: str = 'bearer'

What are Pydantic schemas for?

Pydantic schemas define the shape of data entering and leaving your API. FastAPI uses them to validate request bodies automatically and to generate the Swagger documentation.

UserCreate enforces that registration requires a valid email and a password. UserRead strips the password from responses. Token structures the login response.

• Save app/schemas/user.py.

You won't see a change yet. The next chunk wires everything together into the register endpoint.

• Create a new file app/routers/auth.py and add the registration endpoint:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token
from app.models.user import User
from app.schemas.user import UserCreate, UserRead, Token

router = APIRouter()

@router.post('/register', response_model=UserRead, status_code=201)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):
    # Check if a user with this email already exists
    result = await db.execute(select(User).where(User.email == data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail='Email already registered')
    # Hash the password and create the user record
    user = User(email=data.email, hashed_password=hash_password(data.password))
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

What does the register endpoint do?

It accepts an email and password, checks the database for duplicates, hashes the password with Argon2, and stores the new user. The response returns only the id and email thanks to the UserRead schema.

The status_code=201 signals that a new resource was created, which is the HTTP convention for successful creation.

• Now register this router with the main app. Open app/main.py and find this line:

from sqlalchemy.ext.asyncio import AsyncSession

• Add an import for the auth router directly below it, so your imports look like this:

from sqlalchemy.ext.asyncio import AsyncSession
from app.routers import auth

• Then find this line:

app = FastAPI(title='Webhook Delivery Service')

• Add the router registration directly below it, so it looks like this:

app = FastAPI(title='Webhook Delivery Service')
app.include_router(auth.router, prefix='/auth', tags=['auth'])

• Save app/main.py and app/routers/auth.py.
• Switch to your browser and reload http://localhost:8000/docs. You should see a new auth section with a POST /auth/register endpoint listed.

Don't see the auth section in Swagger?

• Check the terminal running Docker for import errors. A typo in a filename or import path will prevent the app from reloading.
• Make sure app/routers/__init__.py exists (it can be empty). Python needs it to treat the folder as a package.
• Confirm app/schemas/__init__.py also exists.

Why isn't my auth router showing up in the Swagger docs?

Implement the login endpoint

Registration stores users. Now you need a login route that verifies credentials and returns a JWT the client can send with every future request.

• In app/routers/auth.py, add the login function below the register function:

@router.post('/login', response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    # Look up user by email (OAuth2 form uses 'username' field for the email)
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail='Invalid credentials')
    # Generate a signed JWT containing the user's ID
    token = create_access_token(user.id)
    return {'access_token': token, 'token_type': 'bearer'}

What does the login endpoint do?

OAuth2PasswordRequestForm is a FastAPI class that parses a standard OAuth2 login form (username and password fields). Your API uses the email as the username.

If the email exists and the password hash matches, the endpoint returns a signed JWT. The client stores this token and sends it in the Authorization: Bearer <token> header on future requests.

• Save app/routers/auth.py.
• Reload http://localhost:8000/docs. You should now see both POST /auth/register and POST /auth/login in the auth section.

Test the full flow by registering and logging in.

• In Swagger UI, expand POST /auth/register, click Try it out, enter a test email and password, then click Execute.
• You should see a 201 response with the user's id and email.
• Now expand POST /auth/login, click Try it out, enter the same email as username and your password, then click Execute.
• You should see a 200 response containing an access_token string and token_type set to bearer.

Getting a 422 or 401 error?

• A 422 on register means the email format is invalid. Use a full email like test@example.com.
• A 401 on login means the credentials don't match. Make sure you registered first and are using the exact same email and password.
• If the app isn't responding at all, check docker compose logs api for stack traces.

Help me debug my /auth/login 401 error.

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/routers/auth.py and confirmed both endpoints work in Swagger.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token
from app.models.user import User
from app.schemas.user import UserCreate, UserRead, Token

router = APIRouter()

@router.post('/register', response_model=UserRead, status_code=201)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail='Email already registered')
    user = User(email=data.email, hashed_password=hash_password(data.password))
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.post('/login', response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail='Invalid credentials')
    token = create_access_token(user.id)
    return {'access_token': token, 'token_type': 'bearer'}

Create the authentication dependency

You have registration and login working. The final piece is a reusable FastAPI dependency that reads the JWT from incoming requests and resolves it to the authenticated user.

• Create a new file app/core/deps.py and add the following code:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import decode_access_token
from app.models.user import User
import jwt

# Tells FastAPI to look for a Bearer token in the Authorization header
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/login')

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> User:
    credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials')
    try:
        # Decode the token and extract the user ID
        user_id = decode_access_token(token)
    except jwt.PyJWTError:
        raise credentials_exception
    # Look up the user in the database
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

How does get_current_user protect routes?

OAuth2PasswordBearer is a FastAPI class that automatically extracts the Bearer token from the Authorization header. If no token is present, FastAPI returns a 401 before your function even runs.

Any route that adds user: User = Depends(get_current_user) to its parameters becomes a protected route. Invalid or expired tokens raise a 401. The route handler receives the full User object, so it can filter data by user.id.

The tokenUrl='/auth/login' parameter also enables the Authorize button in Swagger UI. You'll use this to test protected endpoints in the next step.

• Save app/core/deps.py.

You won't see a visible change yet. The next step uses this dependency to protect the webhook CRUD routes.

✔️ Awesome, I've got everything!

Great. Confirm you've saved all files in this step.

ⓧ I'd like to double check the full code

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ALGORITHM: str = 'HS256'

settings = Settings()

import jwt
from datetime import datetime, timedelta, timezone
from pwdlib import PasswordHash
from app.core.config import settings

pwd_context = PasswordHash.recommended()

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(subject: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {'sub': str(subject), 'exp': expire}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)

def decode_access_token(token: str) -> int:
    payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
    return int(payload['sub'])

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import decode_access_token
from app.models.user import User
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/login')

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> User:
    credentials_exception = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials')
    try:
        user_id = decode_access_token(token)
    except jwt.PyJWTError:
        raise credentials_exception
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserRead(BaseModel):
    id: int
    email: str

    class Config:
        from_attributes = True

class Token(BaseModel):
    access_token: str
    token_type: str = 'bearer'

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.security import hash_password, verify_password, create_access_token
from app.models.user import User
from app.schemas.user import UserCreate, UserRead, Token

router = APIRouter()

@router.post('/register', response_model=UserRead, status_code=201)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == data.email))
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail='Email already registered')
    user = User(email=data.email, hashed_password=hash_password(data.password))
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.post('/login', response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.email == form_data.username))
    user = result.scalar_one_or_none()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail='Invalid credentials')
    token = create_access_token(user.id)
    return {'access_token': token, 'token_type': 'bearer'}

from fastapi import FastAPI
from sqlalchemy import text
from app.core.database import get_db
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.routers import auth

app = FastAPI(title='Webhook Delivery Service')
app.include_router(auth.router, prefix='/auth', tags=['auth'])

@app.get('/health')
async def health(db: AsyncSession = Depends(get_db)):
    await db.execute(text('SELECT 1'))
    return {'status': 'ok'}

Your API now has a complete authentication system. Users can register, log in, and receive tokens. In the next step, you'll put get_current_user to work by building the webhook CRUD endpoints that only authenticated users can access.

Build Webhook CRUD Routes

Your API can now register users and issue tokens. The next challenge is building the actual domain logic that makes this a webhook delivery service.

In this step, you will build the three core resource types: endpoints (URLs that receive webhooks), subscriptions (rules that link endpoints to event types), and the event ingestion pipeline that routes incoming events to the right subscribers.

In this step, get ready to:

• Build CRUD routes for webhook endpoints.
• Build CRUD routes for event subscriptions with filtering.
• Build the event ingestion route that stores events with a pending delivery status.

Create the endpoint schemas and router

An endpoint in this system represents a URL that will eventually receive webhook deliveries. Users register their endpoints first, then subscribe them to specific event types.

• Create a new file app/schemas/endpoint.py and add the following Pydantic schemas:

from pydantic import BaseModel
from typing import Optional

# Schema for creating a new endpoint
class EndpointCreate(BaseModel):
    url: str
    description: Optional[str] = None

# Schema for reading an endpoint from the database
class EndpointRead(BaseModel):
    id: int
    url: str
    description: Optional[str]
    user_id: int

    class Config:
        from_attributes = True

What does this code do?

EndpointCreate defines what data is required to register a new endpoint. Only a URL is mandatory.

EndpointRead defines the shape of endpoint data returned to the client. The from_attributes = True config lets Pydantic read directly from SQLAlchemy model instances.

• Save app/schemas/endpoint.py.

Now you will build the router that handles creating, listing, and deleting endpoints. Every route is protected by the get_current_user dependency so users can only access their own resources.

• Create a new file app/routers/endpoints.py and add the following code:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.endpoint import Endpoint
from app.models.user import User
from app.schemas.endpoint import EndpointCreate, EndpointRead
from typing import List

router = APIRouter()

# Create a new webhook endpoint for the authenticated user
@router.post('/', response_model=EndpointRead, status_code=201)
async def create_endpoint(data: EndpointCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    ep = Endpoint(url=data.url, description=data.description, user_id=current_user.id)
    db.add(ep)
    await db.commit()
    await db.refresh(ep)
    return ep

# List all endpoints belonging to the authenticated user
@router.get('/', response_model=List[EndpointRead])
async def list_endpoints(db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Endpoint).where(Endpoint.user_id == current_user.id))
    return result.scalars().all()

# Delete an endpoint only if it belongs to the authenticated user
@router.delete('/{endpoint_id}', status_code=204)
async def delete_endpoint(endpoint_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Endpoint).where(Endpoint.id == endpoint_id, Endpoint.user_id == current_user.id))
    ep = result.scalar_one_or_none()
    if not ep:
        raise HTTPException(status_code=404, detail='Endpoint not found')
    await db.delete(ep)
    await db.commit()

What does this code do?

• create_endpoint accepts a URL and optional description, attaches the current user's ID, and stores it in the database.
• list_endpoints queries only endpoints where user_id matches the authenticated user. This enforces user isolation.
• delete_endpoint checks both the endpoint ID and the user ID in its query. If User A tries to delete User B's endpoint, the query returns nothing and a 404 is raised.

• Save app/routers/endpoints.py.

Why filter by user_id in the query?

Adding Endpoint.user_id == current_user.id to every query is how you enforce user isolation at the database level. Even if someone guesses another user's endpoint ID, the query will return nothing because the user_id won't match.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.endpoint import Endpoint
from app.models.user import User
from app.schemas.endpoint import EndpointCreate, EndpointRead
from typing import List

router = APIRouter()

@router.post('/', response_model=EndpointRead, status_code=201)
async def create_endpoint(data: EndpointCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    ep = Endpoint(url=data.url, description=data.description, user_id=current_user.id)
    db.add(ep)
    await db.commit()
    await db.refresh(ep)
    return ep

@router.get('/', response_model=List[EndpointRead])
async def list_endpoints(db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Endpoint).where(Endpoint.user_id == current_user.id))
    return result.scalars().all()

@router.delete('/{endpoint_id}', status_code=204)
async def delete_endpoint(endpoint_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Endpoint).where(Endpoint.id == endpoint_id, Endpoint.user_id == current_user.id))
    ep = result.scalar_one_or_none()
    if not ep:
        raise HTTPException(status_code=404, detail='Endpoint not found')
    await db.delete(ep)
    await db.commit()

Create the subscription schemas and router

A subscription links an endpoint to a specific event type. For example, subscribing an endpoint to order.created means that endpoint will receive a webhook whenever an order.created event is ingested.

• Create a new file app/schemas/subscription.py and add the following schemas:

from pydantic import BaseModel

# Schema for creating a subscription (links an endpoint to an event type)
class SubscriptionCreate(BaseModel):
    endpoint_id: int
    event_type: str

# Schema for reading a subscription from the database
class SubscriptionRead(BaseModel):
    id: int
    endpoint_id: int
    event_type: str
    user_id: int

    class Config:
        from_attributes = True

What does this code do?

SubscriptionCreate requires an endpoint_id and an event_type string. The user is attached server-side from the JWT token, never from the request body.

• Save app/schemas/subscription.py.

The subscription router is similar to the endpoint router, but adds the ability to filter by event type when listing subscriptions.

• Create a new file app/routers/subscriptions.py and add the following code:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.subscription import SubscriptionCreate, SubscriptionRead
from typing import List, Optional

router = APIRouter()

# Create a subscription linking an endpoint to an event type
@router.post('/', response_model=SubscriptionRead, status_code=201)
async def create_subscription(data: SubscriptionCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    sub = Subscription(endpoint_id=data.endpoint_id, event_type=data.event_type, user_id=current_user.id)
    db.add(sub)
    await db.commit()
    await db.refresh(sub)
    return sub

# List subscriptions, optionally filtered by event_type
@router.get('/', response_model=List[SubscriptionRead])
async def list_subscriptions(event_type: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    query = select(Subscription).where(Subscription.user_id == current_user.id)
    if event_type:
        query = query.where(Subscription.event_type == event_type)
    result = await db.execute(query)
    return result.scalars().all()

# Delete a subscription only if it belongs to the authenticated user
@router.delete('/{subscription_id}', status_code=204)
async def delete_subscription(subscription_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Subscription).where(Subscription.id == subscription_id, Subscription.user_id == current_user.id))
    sub = result.scalar_one_or_none()
    if not sub:
        raise HTTPException(status_code=404, detail='Subscription not found')
    await db.delete(sub)
    await db.commit()

What does this code do?

• create_subscription links a user's endpoint to an event type string like order.created.
• list_subscriptions accepts an optional event_type query parameter. If provided, results are filtered to only that event type.
• delete_subscription follows the same user-isolation pattern as endpoint deletion. The query checks both the subscription ID and the user ID.

• Save app/routers/subscriptions.py.

✔️ Awesome, I've got everything!

Great. Make sure both subscription files are saved before continuing.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.subscription import SubscriptionCreate, SubscriptionRead
from typing import List, Optional

router = APIRouter()

@router.post('/', response_model=SubscriptionRead, status_code=201)
async def create_subscription(data: SubscriptionCreate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    sub = Subscription(endpoint_id=data.endpoint_id, event_type=data.event_type, user_id=current_user.id)
    db.add(sub)
    await db.commit()
    await db.refresh(sub)
    return sub

@router.get('/', response_model=List[SubscriptionRead])
async def list_subscriptions(event_type: Optional[str] = None, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    query = select(Subscription).where(Subscription.user_id == current_user.id)
    if event_type:
        query = query.where(Subscription.event_type == event_type)
    result = await db.execute(query)
    return result.scalars().all()

@router.delete('/{subscription_id}', status_code=204)
async def delete_subscription(subscription_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Subscription).where(Subscription.id == subscription_id, Subscription.user_id == current_user.id))
    sub = result.scalar_one_or_none()
    if not sub:
        raise HTTPException(status_code=404, detail='Subscription not found')
    await db.delete(sub)
    await db.commit()

Build the event ingestion and listing routes

The event ingestion route is where the webhook system comes together. When an event arrives, the API finds all subscriptions matching that event type and creates a pending event record for each one.

• Create a new file app/schemas/event.py and add the following schemas:

from pydantic import BaseModel
from typing import Any, Optional

# Schema for ingesting a new event
class EventIngest(BaseModel):
    event_type: str
    payload: Any

# Schema for reading an event from the database
class EventRead(BaseModel):
    id: int
    event_type: str
    payload: Any
    status: str
    subscription_id: Optional[int]
    user_id: int

    class Config:
        from_attributes = True

What does this code do?

EventIngest accepts an event_type string and an arbitrary JSON payload. The Any type hint means the payload can be any valid JSON structure.

EventRead includes the status field (which defaults to pending) and the subscription_id that links back to which subscription matched this event.

• Save app/schemas/event.py.

Now for the event router. This contains the ingestion logic and a route to list all events for the authenticated user.

• Create a new file app/routers/events.py and add the following code:

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.event import Event
from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.event import EventIngest, EventRead
from typing import List

router = APIRouter()

# Ingest an event: find matching subscriptions and store a pending event for each
@router.post('/ingest', response_model=List[EventRead], status_code=201)
async def ingest_event(data: EventIngest, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Subscription).where(Subscription.user_id == current_user.id, Subscription.event_type == data.event_type))
    subs = result.scalars().all()
    created = []
    for sub in subs:
        event = Event(user_id=current_user.id, event_type=data.event_type, payload=data.payload, status='pending', subscription_id=sub.id)
        db.add(event)
        created.append(event)
    await db.commit()
    for e in created:
        await db.refresh(e)
    return created

# List all events belonging to the authenticated user
@router.get('/', response_model=List[EventRead])
async def list_events(db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Event).where(Event.user_id == current_user.id))
    return result.scalars().all()

What does this code do?

• ingest_event queries all subscriptions matching the authenticated user AND the incoming event_type. For each matching subscription, it creates an Event record with status pending.
• The loop commits all events in a single transaction, then refreshes each to get their database-assigned IDs.
• list_events returns all events for the current user, regardless of status or subscription.

• Save app/routers/events.py.

✔️ Awesome, I've got everything!

Great. Make sure both event files are saved.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.core.deps import get_current_user
from app.models.event import Event
from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.event import EventIngest, EventRead
from typing import List

router = APIRouter()

@router.post('/ingest', response_model=List[EventRead], status_code=201)
async def ingest_event(data: EventIngest, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Subscription).where(Subscription.user_id == current_user.id, Subscription.event_type == data.event_type))
    subs = result.scalars().all()
    created = []
    for sub in subs:
        event = Event(user_id=current_user.id, event_type=data.event_type, payload=data.payload, status='pending', subscription_id=sub.id)
        db.add(event)
        created.append(event)
    await db.commit()
    for e in created:
        await db.refresh(e)
    return created

@router.get('/', response_model=List[EventRead])
async def list_events(db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user)):
    result = await db.execute(select(Event).where(Event.user_id == current_user.id))
    return result.scalars().all()

The final piece is registering all three new routers in your application entry point.

• In app/main.py, find this line:

from app.routers import auth

• Replace it with this updated import that includes all four routers:

from app.routers import auth, endpoints, subscriptions, events

• Find the line that says app.include_router(auth.router, prefix='/auth', tags=['auth']) and add these three lines directly below it:

app.include_router(endpoints.router, prefix='/endpoints', tags=['endpoints'])
app.include_router(subscriptions.router, prefix='/subscriptions', tags=['subscriptions'])
app.include_router(events.router, prefix='/events', tags=['events'])

What does this code do?

Each include_router call mounts a router at a URL prefix. The tags parameter groups routes together in the auto-generated OpenAPI documentation at /docs.

• Also update the existing imports at the top of app/main.py so they match this order:

from fastapi import FastAPI, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.routers import auth, endpoints, subscriptions, events

• Save app/main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved the file.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.routers import auth, endpoints, subscriptions, events

app = FastAPI(title='Webhook Delivery Service')
app.include_router(auth.router, prefix='/auth', tags=['auth'])
app.include_router(endpoints.router, prefix='/endpoints', tags=['endpoints'])
app.include_router(subscriptions.router, prefix='/subscriptions', tags=['subscriptions'])
app.include_router(events.router, prefix='/events', tags=['events'])

@app.get('/health')
async def health(db: AsyncSession = Depends(get_db)):
    await db.execute(text('SELECT 1'))
    return {'status': 'ok'}

• Rebuild the containers to pick up the new files by running:

docker compose up --build

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

You should see the Swagger UI with four tagged groups: auth, endpoints, subscriptions, and events. Each group shows the routes you just created.

Seeing an import error or the app won't start?

• Check that your file paths match exactly. The routers live at app/routers/endpoints.py, app/routers/subscriptions.py, and app/routers/events.py.
• Make sure your schema files are at app/schemas/endpoint.py, app/schemas/subscription.py, and app/schemas/event.py (singular, not plural).
• Check the Docker logs for the specific error line. The traceback will tell you which import failed.

Help me debug my FastAPI import error when registering routers.

Now test the full flow using the Swagger UI. You will register a user, get a token, create an endpoint, subscribe it, and ingest an event.

• In the Swagger UI, expand POST /auth/register and click Try it out.
• Register a user with any email and password, then click Execute.
• Expand POST /auth/login, enter the same credentials, and click Execute. Copy the access_token from the response.
• Click the Authorize button at the top of the page and paste your token into the value field.
• Expand POST /endpoints and create an endpoint with a URL like https://example.com/webhook. Note the returned id.
• Expand POST /subscriptions and create a subscription using that endpoint ID and the event type order.created.
• Expand POST /events/ingest and submit an event with event_type set to order.created and any JSON payload.
• Expand GET /events and click Execute.

You should see the event in the response with status set to pending and a subscription_id linking it back to your subscription.

Event ingestion returns an empty list?

• Make sure the event_type you pass to /events/ingest matches exactly what you used when creating the subscription (case-sensitive).
• Verify your subscription was created successfully by calling GET /subscriptions first.
• Confirm you are using the same authenticated user for all requests. The ingestion route only finds subscriptions belonging to the current user.

Help me debug why my event ingestion returns empty.

Your webhook delivery service now has a complete domain layer. Users can register endpoints, subscribe them to event types, and ingest events that get routed to the correct subscribers. Next up, you will write automated tests to prove all of this works correctly.

Write Automated Tests

Your webhook API is fully functional. You can register users, manage endpoints and subscriptions, and ingest events. But how do you know it actually works correctly after every change?

Automated tests give you a safety net. In this step, you will write a test suite using pytest and FastAPI's TestClient that proves your API handles happy paths and error cases correctly.

In this step, get ready to:

• Set up a test database using SQLite so tests run fast without touching PostgreSQL.
• Write auth flow tests covering registration, login, and token validation.
• Write webhook flow tests covering endpoint creation, subscriptions, event ingestion, and user isolation.

Set up the test infrastructure

Tests need their own database so they do not pollute your development data. You will use SQLite with the aiosqlite driver for speed.

FastAPI's dependency_overrides feature lets you swap the real PostgreSQL session for a lightweight SQLite session during tests.

• Create the test directory and an empty __init__.py to make it a Python package by running these commands:

mkdir -p tests
touch tests/__init__.py

• Confirm the files exist by running:

ls tests/

You should see __init__.py in the output.

• Create a new file called tests/conftest.py and add the following test infrastructure code:

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.main import app
from app.core.database import get_db
from app.models.base import Base
import asyncio

# Use SQLite for fast, isolated test runs
TEST_DATABASE_URL = 'sqlite+aiosqlite:///./test.db'

@pytest.fixture(scope='session')
def event_loop():
    # Provide a single event loop for all async fixtures in the session
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope='session')
async def test_engine():
    # Create all tables once at the start, drop them at the end
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
def client(test_engine):
    # Override the real database dependency with the test database
    TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)
    async def override_get_db():
        async with TestSessionLocal() as session:
            yield session
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def get_auth_headers(client, email='test@example.com', password='testpass123'):
    # Helper: register a user, log in, and return the auth header dict
    client.post('/auth/register', json={'email': email, 'password': password})
    resp = client.post('/auth/login', data={'username': email, 'password': password})
    token = resp.json()['access_token']
    return {'Authorization': f'Bearer {token}'}

What does this code do?

This file sets up three pytest fixtures and one helper function that all your test files will share.

• event_loop provides a single async event loop for session-scoped async fixtures.
• test_engine creates a SQLite database and builds all your tables from the SQLAlchemy models. At the end of the session it tears them down.
• client uses FastAPI's dependency_overrides to swap the real PostgreSQL session for the test SQLite session. Every test gets a fresh TestClient pointing at your app.
• get_auth_headers is a helper that registers a user, logs them in, and returns the Authorization header ready to use in protected route calls.

• Save tests/conftest.py.

Import errors when running tests later?

Make sure tests/__init__.py exists. Without it, Python cannot resolve imports from the tests package.

Also confirm that aiosqlite is listed in your requirements.txt. It was added in Step 1, but double check if you see a missing driver error.

Help me debug my import error

✔️ Awesome, I've got everything!

Great. Make sure you have saved tests/conftest.py before moving on.

ⓧ I'd like to double check the full code

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.main import app
from app.core.database import get_db
from app.models.base import Base
import asyncio

TEST_DATABASE_URL = 'sqlite+aiosqlite:///./test.db'

@pytest.fixture(scope='session')
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope='session')
async def test_engine():
    engine = create_async_engine(TEST_DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
def client(test_engine):
    TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)
    async def override_get_db():
        async with TestSessionLocal() as session:
            yield session
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def get_auth_headers(client, email='test@example.com', password='testpass123'):
    client.post('/auth/register', json={'email': email, 'password': password})
    resp = client.post('/auth/login', data={'username': email, 'password': password})
    token = resp.json()['access_token']
    return {'Authorization': f'Bearer {token}'}

Write auth flow tests

Auth is the gateway to every protected route. If registration or login breaks, nothing else works.

• Create a new file called tests/test_auth.py and add the following test functions:

from tests.conftest import get_auth_headers

def test_register(client):
    # Happy path: a new user can register
    resp = client.post('/auth/register', json={'email': 'a@test.com', 'password': 'pass'})
    assert resp.status_code == 201

def test_register_duplicate(client):
    # Error: registering with the same email twice returns 400
    client.post('/auth/register', json={'email': 'dup@test.com', 'password': 'pass'})
    resp = client.post('/auth/register', json={'email': 'dup@test.com', 'password': 'pass'})
    assert resp.status_code == 400

def test_login(client):
    # Happy path: a registered user can log in and receives a token
    client.post('/auth/register', json={'email': 'login@test.com', 'password': 'pass'})
    resp = client.post('/auth/login', data={'username': 'login@test.com', 'password': 'pass'})
    assert resp.status_code == 200
    assert 'access_token' in resp.json()

def test_login_invalid(client):
    # Error: wrong credentials return 401
    resp = client.post('/auth/login', data={'username': 'no@test.com', 'password': 'wrong'})
    assert resp.status_code == 401

def test_protected_no_token(client):
    # Error: accessing a protected route without a token returns 401
    resp = client.get('/endpoints/')
    assert resp.status_code == 401

What does this code do?

Each test function exercises one specific behavior of your auth system.

• test_register confirms a new user gets a 201 response.
• test_register_duplicate proves the duplicate email guard works (400 response).
• test_login verifies that a registered user receives a valid access_token on login.
• test_login_invalid confirms wrong credentials get rejected with 401.
• test_protected_no_token proves that unauthenticated requests to protected routes are blocked.

• Save tests/test_auth.py.

Tests failing with connection errors?

These tests do not use the PostgreSQL container. If you see connection errors mentioning asyncpg, make sure your conftest.py overrides get_db correctly. The override must point to the SQLite test engine.

Help me debug my database connection errors

✔️ Awesome, I've got everything!

Great. Make sure you have saved tests/test_auth.py before moving on.

ⓧ I'd like to double check the full code

from tests.conftest import get_auth_headers

def test_register(client):
    resp = client.post('/auth/register', json={'email': 'a@test.com', 'password': 'pass'})
    assert resp.status_code == 201

def test_register_duplicate(client):
    client.post('/auth/register', json={'email': 'dup@test.com', 'password': 'pass'})
    resp = client.post('/auth/register', json={'email': 'dup@test.com', 'password': 'pass'})
    assert resp.status_code == 400

def test_login(client):
    client.post('/auth/register', json={'email': 'login@test.com', 'password': 'pass'})
    resp = client.post('/auth/login', data={'username': 'login@test.com', 'password': 'pass'})
    assert resp.status_code == 200
    assert 'access_token' in resp.json()

def test_login_invalid(client):
    resp = client.post('/auth/login', data={'username': 'no@test.com', 'password': 'wrong'})
    assert resp.status_code == 401

def test_protected_no_token(client):
    resp = client.get('/endpoints/')
    assert resp.status_code == 401

Write webhook flow tests

These tests cover the core webhook lifecycle. They prove that endpoints, subscriptions, and events all work together correctly.

The final test verifies user isolation, which is critical for a multi-tenant API.

• Create a new file called tests/test_webhooks.py and add the following test functions:

from tests.conftest import get_auth_headers

def test_create_endpoint(client):
    # Happy path: an authenticated user can create a webhook endpoint
    headers = get_auth_headers(client, 'ep@test.com', 'pass')
    resp = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers)
    assert resp.status_code == 201

def test_create_subscription(client):
    # Happy path: subscribe an endpoint to an event type
    headers = get_auth_headers(client, 'sub@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers).json()
    resp = client.post('/subscriptions/', json={'endpoint_id': ep['id'], 'event_type': 'order.created'}, headers=headers)
    assert resp.status_code == 201

def test_ingest_event(client):
    # Full flow: create endpoint, subscribe, ingest event, verify pending status
    headers = get_auth_headers(client, 'ingest@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers).json()
    client.post('/subscriptions/', json={'endpoint_id': ep['id'], 'event_type': 'order.created'}, headers=headers)
    resp = client.post('/events/ingest', json={'event_type': 'order.created', 'payload': {'order_id': 1}}, headers=headers)
    assert resp.status_code == 201
    events = resp.json()
    assert len(events) == 1
    assert events[0]['status'] == 'pending'

def test_user_isolation(client):
    # Security: user B cannot delete user A's endpoint
    headers_a = get_auth_headers(client, 'userA@test.com', 'pass')
    headers_b = get_auth_headers(client, 'userB@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://a.com/hook'}, headers=headers_a).json()
    resp = client.delete(f'/endpoints/{ep["id"]}', headers=headers_b)
    assert resp.status_code == 404

What does this code do?

Each test builds on the previous one to exercise increasingly complex behavior.

• test_create_endpoint proves an authenticated user can register a webhook URL.
• test_create_subscription links an endpoint to the order.created event type.
• test_ingest_event runs the full lifecycle: create endpoint, subscribe, ingest, then asserts the event is stored with pending status.
• test_user_isolation creates an endpoint as user A, then attempts to delete it as user B. The API returns 404 because user B cannot even see user A's resources.

• Save tests/test_webhooks.py.

User isolation test returning 204 instead of 404?

This means user B can see user A's endpoint. Check your delete_endpoint route in app/routers/endpoints.py. The query must filter by both Endpoint.id == endpoint_id and Endpoint.user_id == current_user.id.

Help me fix my user isolation query

✔️ Awesome, I've got everything!

Great. Make sure you have saved tests/test_webhooks.py before moving on.

ⓧ I'd like to double check the full code

from tests.conftest import get_auth_headers

def test_create_endpoint(client):
    headers = get_auth_headers(client, 'ep@test.com', 'pass')
    resp = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers)
    assert resp.status_code == 201

def test_create_subscription(client):
    headers = get_auth_headers(client, 'sub@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers).json()
    resp = client.post('/subscriptions/', json={'endpoint_id': ep['id'], 'event_type': 'order.created'}, headers=headers)
    assert resp.status_code == 201

def test_ingest_event(client):
    headers = get_auth_headers(client, 'ingest@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://example.com/hook'}, headers=headers).json()
    client.post('/subscriptions/', json={'endpoint_id': ep['id'], 'event_type': 'order.created'}, headers=headers)
    resp = client.post('/events/ingest', json={'event_type': 'order.created', 'payload': {'order_id': 1}}, headers=headers)
    assert resp.status_code == 201
    events = resp.json()
    assert len(events) == 1
    assert events[0]['status'] == 'pending'

def test_user_isolation(client):
    headers_a = get_auth_headers(client, 'userA@test.com', 'pass')
    headers_b = get_auth_headers(client, 'userB@test.com', 'pass')
    ep = client.post('/endpoints/', json={'url': 'https://a.com/hook'}, headers=headers_a).json()
    resp = client.delete(f'/endpoints/{ep["id"]}', headers=headers_b)
    assert resp.status_code == 404

• Run the full test suite inside your Docker container by executing this command:

docker compose exec api pytest -v

What should I see?

You should see output listing each test function with a PASSED label next to it. All 9 tests (5 auth + 4 webhook) should pass. The output ends with a summary line showing the total number of tests that passed.

Seeing a 'no such service' error?

Make sure your containers are running. Run docker compose up -d first, then re-run the pytest command.

If tests fail with import errors, make sure the tests/ directory is inside your project root (the same folder that has your Dockerfile). The volume mount in docker-compose.yml maps your local project into the container at /app.

Help me troubleshoot my test setup

Your API now has a proven safety net. Next up, the Secret Mission challenges you to add HMAC signature verification so webhook consumers can verify payload authenticity.

Secret mission

Add HMAC Webhook Signatures

Real webhook systems like Stripe and GitHub sign every payload with a secret so consumers can verify authenticity. Your service delivers events, but how does a consumer know those events are genuine? In this secret mission, you will add HMAC-SHA256 signature verification to your webhook delivery service.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

• Docker containers (api and db services).
• Docker volume pgdata storing PostgreSQL data.
• Locally built Docker images.
• Project folder on your Desktop.

Keep everything running

No action needed. Choose this if you're still actively building or plan to continue with Project 2, which extends this service.

• If your containers are stopped, start them again in detached mode by running:

docker compose up -d

Your API stays available at http://localhost:8000/docs and all data in PostgreSQL persists.

Pause - I'll come back to this later

Shut down running containers to free up memory, but keep your data and files intact for later.

• Stop the containers by running:

docker compose stop

This halts the containers but preserves the PostgreSQL volume and all project files. When you're ready to resume, run docker compose up -d from your project folder.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes containers, volumes, and local files permanently.

• Remove containers and the database volume by running this from your project folder:

docker compose down -v

• Remove the locally built Docker images by running:

docker compose down --rmi local

• Delete the project folder. On macOS or Linux:

cd ~/Desktop
rm -rf webhook-delivery-service

On Windows:

cd ~\Desktop
rmdir /s /q webhook-delivery-service

Nice Work!

Nice Work!

You just built a production-structured webhook delivery API from scratch. This is the kind of backend infrastructure that powers notifications at companies like Stripe, GitHub, and Shopify.

You've learned how to:

• Build a REST API with FastAPI that handles webhook endpoint registration, event subscriptions, and event ingestion with JWT authentication and Argon2 password hashing.
• Design a relational data model with proper foreign keys and user isolation, backed by PostgreSQL with Alembic migrations.
• Containerize the entire stack with Docker Compose and write automated tests with pytest proving the API works correctly.
• Secret Mission: Added HMAC-SHA256 webhook signature verification so consumers can confirm payload authenticity.

Ready to quiz yourself?