Add User Authentication Part 3

Introduction

30 Second Summary

A booking service becomes risky when any visitor can change someone else's event. Each person needs a reliable way to prove who they are before the service trusts a request.

In this project, you will add secure user authentication to your existing FastAPI event-booking API. You will prove its ownership rules and admin boundaries with Postman requests plus automated pytest tests.

What You'll Build

You'll register two users, log in as each one, create an event, and watch the API block the second user from changing something they do not own.

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

• A secure account flow that stores Argon2 password hashes without exposing them. You'll log in to receive an expiring bearer token.
• A multi-user event API where event creators control their own events. Each booking is tied to the authenticated user who created it.
• A repeatable security proof that demonstrates successful requests. It also demonstrates missing authentication, denied ownership, duplicate bookings, and blocked admin access.
• Secret Mission: Add a regression test that creates an already-expired JWT and proves the API rejects it with 401 Unauthorized.

Are there any prerequisites?

This project continues directly from Project 2. You need that completed codebase plus its existing virtual environment.

You also need local PostgreSQL, Visual Studio Code, and Windows PowerShell. Step 1 adds the new pinned packages plus the clean databases used throughout the project.

Before We Start

This checkpoint helps you commit to the security model before changing your existing API. You will define who owns each resource and why identity still needs permission checks.

Prepare the Authentication Build

Your FastAPI API now has a clear authorization model. The build still needs its authentication libraries.

Password hashing introduces cryptography into the request flow. The login form requires dedicated parsing support.

Async API tests must start the application lifespan correctly. Exact package pins keep version drift from hiding security bugs.

In this step, get ready to:

• Install the complete pinned dependency set inside the existing virtual environment.
• Create clean PostgreSQL databases for development plus testing.
• Create the local .env configuration for authenticated API runs.

Pin and install the dependencies

A pinned requirements.txt gives every installation the same package set. The new packages complete the JWT authentication stack.

• Return to the Project 2 codebase in Visual Studio Code.
• Select requirements.txt in the Explorer sidebar.
• Replace the file contents with this complete pinned dependency set:

fastapi==0.162.0
tortoise-orm==1.1.9
asyncpg==0.32.0
pydantic==2.12.5
uvicorn==0.54.0
python-dotenv==1.2.4
PyJWT==2.15.1
pwdlib[argon2]==0.3.1
python-multipart==0.0.32
pytest==9.1.0
pytest-asyncio==1.3.0
httpx==0.28.1
asgi-lifespan==2.1.0

Why These Packages?

• PyJWT 2.15.1 signs access tokens and validates incoming tokens.
• pwdlib 0.3.1 supplies Argon2 password hashing through the selected extra.
• python-multipart 0.0.32 parses the OAuth2 password form used by the login route.
• asgi-lifespan 2.1.0 runs application startup plus shutdown during asynchronous tests.

• Save requirements.txt.

You should now see 13 pinned package lines in requirements.txt.

Does the Package List Look Different?

Remove any unpinned package lines left from the previous file. Keep the capitalization of PyJWT plus the [argon2] extra exactly as shown.

Ask for a line-by-line comparison: Help me compare my requirements.txt with the pinned authentication dependency set.

✔️ Awesome, I've got everything!

Your saved requirements.txt now contains the complete pinned dependency set.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt with this reference:

fastapi==0.162.0
tortoise-orm==1.1.9
asyncpg==0.32.0
pydantic==2.12.5
uvicorn==0.54.0
python-dotenv==1.2.4
PyJWT==2.15.1
pwdlib[argon2]==0.3.1
python-multipart==0.0.32
pytest==9.1.0
pytest-asyncio==1.3.0
httpx==0.28.1
asgi-lifespan==2.1.0

• Return to the PowerShell terminal in your Project 2 folder.
• Activate the existing virtual environment by running this command:

.\.venv\Scripts\Activate.ps1

What Does Activation Change?

PowerShell now uses the Python interpreter plus installed packages from .venv. This keeps the authentication dependencies inside the existing Project 2 environment.

Your PowerShell prompt should now show the virtual environment name.

Virtual Environment Did Not Activate?

Confirm that the terminal is inside the Project 2 folder containing .venv. Check that .venv\Scripts\Activate.ps1 exists before retrying.

Ask for help with the existing environment: Help me activate my existing Project 2 virtual environment in Windows PowerShell.

• Install every package from the saved file by running this command:

python -m pip install -r requirements.txt

What Does This Command Do?

• The python -m pip portion uses the package installer attached to the active Python interpreter.
• The -r requirements.txt portion installs every exact pin from the saved file.

PowerShell should finish the installation without dependency conflicts.

Dependency Installation Failed?

Confirm that the virtual environment is still active. Retry after restoring your network connection if the package download was interrupted.

Ask for help with the installation output: Help me diagnose my pinned dependency installation failure.

Create the clean databases

A new development database preserves your Project 2 data. A separate test database gives automated cleanup a safe target.

PostgreSQL may request the password assigned to your local postgres role. The role must have permission to create databases.

• Create both clean databases from the active PowerShell terminal by running these commands:

psql -U postgres -d postgres -c "CREATE DATABASE event_booking_auth;"
psql -U postgres -d postgres -c "CREATE DATABASE event_booking_auth_test;"

What Do These Commands Do?

• The -U postgres option connects through the local postgres role.
• The -d postgres option selects the existing administrative database.
• The -c option executes the supplied database statement.
• The event_booking_auth database holds local development data.
• The event_booking_auth_test database holds disposable test data.

That is the data safety boundary in place. Both authentication databases are ready without touching your Project 2 data.

Database Creation Failed?

Confirm that your local PostgreSQL service is running. Check that the postgres role can create databases.

A database with the same name may already exist. Help me diagnose the local database creation failure.

Configure the local environment

The application needs separate database URLs for normal runs plus tests. It also needs a private signing value for expiring access tokens.

• Select .env.example in the Visual Studio Code Explorer sidebar.
• Replace the file contents with this safe configuration template:

DATABASE_URL=postgres://postgres:your-postgres-password@localhost:5432/event_booking_auth
TEST_DATABASE_URL=postgres://postgres:your-postgres-password@localhost:5432/event_booking_auth_test
JWT_SECRET_KEY=your-generated-secret-here
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

What Does This Configuration Store?

• DATABASE_URL points normal application runs at the clean development database.
• TEST_DATABASE_URL points test runs at the isolated test database.
• JWT_SECRET_KEY holds the local signing secret.
• JWT_ALGORITHM fixes token signing to HS256.
• ACCESS_TOKEN_EXPIRE_MINUTES sets the access token lifetime to 30 minutes.

• Save .env.example.

You should see five settings in the saved template. Both database URLs should end with the new database names.

Template Has Missing Settings?

Compare the file with the full reference below. Keep each setting on its own line.

Ask for help comparing the template: Help me find missing or changed settings in my .env.example file.

✔️ Awesome, I've got everything!

Your saved .env.example contains the safe configuration template.

ⓧ I'd like to double check the full code

Compare your complete .env.example with this reference:

DATABASE_URL=postgres://postgres:your-postgres-password@localhost:5432/event_booking_auth
TEST_DATABASE_URL=postgres://postgres:your-postgres-password@localhost:5432/event_booking_auth_test
JWT_SECRET_KEY=your-generated-secret-here
JWT_ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Creating a signing secret can feel sensitive, so this value stays in your local .env file.

• Generate the local signing secret from the active PowerShell terminal by running this command:

python -c "import secrets; print(secrets.token_hex(32))"

What Does This Command Do?

Python generates 32 random bytes through its secrets module. Hexadecimal encoding prints them as one 64-character value.

You should see one random 64-character hexadecimal value in PowerShell.

• Copy the generated value without adding spaces or quotation marks.
• Create the local .env file from the safe template by running this command:

Copy-Item -Path ".env.example" -Destination ".env"

What Does This Command Do?

PowerShell copies .env.example to a local file named .env. The safe template remains available for source control.

You should now see .env beside .env.example in the Visual Studio Code Explorer.

• Select .env in the Explorer sidebar.

You should see the same five settings from the safe template.

• Replace both your-postgres-password entries with the password for your local postgres role.
• Save .env.

Both database URLs should now contain your local PostgreSQL password.

• Replace your-generated-secret-here with the copied 64-character value.
• Save .env.

The JWT_SECRET_KEY line should now contain 64 hexadecimal characters. Keep the actual value private.

• Leave .env uncommitted.

Keep the Signing Secret Local

Your local .env now contains a database password plus a token-signing secret. Share .env.example without sharing .env.

Before you run the final check, which four package names plus versions do you expect PowerShell to report?

• Verify the new authentication packages inside the active virtual environment by running this command:

python -m pip show PyJWT pwdlib python-multipart asgi-lifespan

What Should You See?

The command should report PyJWT 2.15.1 plus pwdlib 0.3.1.

It should also report python-multipart 0.0.32 plus asgi-lifespan 2.1.0.

Package Details Missing?

Confirm that the existing virtual environment is active. Rerun the pinned installation command if any package is missing.

Ask for help reading the package report: Help me diagnose a missing or incorrect authentication package version.

Your authentication environment is ready across dependencies, databases, plus local settings. Next, you'll rebuild the schema for the first registered user.

Rebuild the Schema and Register a User

Your dependencies are pinned. Your two PostgreSQL databases are ready for a clean authentication schema.

Registration needs a migration-managed user table before it can persist an identity. This step stores each password as an Argon2 hash while keeping password data out of every response.

In this step, get ready to:

• Configure database selection for development and test runs.
• Model users plus owned events and bookings.
• Add a registration endpoint that returns public user data.

Configure the database schema

A database schema defines which records exist. It also defines how those records relate to one another.

Why use a clean schema?

The clean databases protect your Project 2 data from migration changes. They also give development runs and automated tests separate storage.

The TESTING environment variable selects the test database. Normal runs continue to use the development database.

• Switch back to Visual Studio Code from earlier.
• Create settings.py in the same folder as your existing main.py file.
• Build the database configuration by pasting this code into settings.py:

import os

from dotenv import load_dotenv

load_dotenv()

TESTING = os.getenv("TESTING") == "1"
DATABASE_URL = (
    os.environ["TEST_DATABASE_URL"] if TESTING else os.environ["DATABASE_URL"]
)
JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES = int(
    os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30")
)

TORTOISE_ORM = {
    "connections": {"default": DATABASE_URL},
    "apps": {
        "models": {
            "models": ["models"],
            "default_connection": "default",
            "migrations": "migrations",
        }
    },
    "use_tz": True,
    "timezone": "UTC",
}

What does this configuration do?

• The TESTING check selects TEST_DATABASE_URL only when its value is 1.
• The token settings load the signing key plus the fixed algorithm and expiration values from .env.
• The TORTOISE_ORM dictionary connects the models app to the selected database.
• The migrations setting tells Tortoise ORM where to store generated migration files.

• Save settings.py.
• Confirm that settings.py is listed beside main.py in the Explorer sidebar.

Your application now has one configuration path for both local databases.

• Select the existing models.py file in the Explorer sidebar.
• Replace its current contents with the imports and User model below:

from tortoise import fields
from tortoise.migrations.constraints import UniqueConstraint
from tortoise.models import Model


class User(Model):
    id = fields.IntField(primary_key=True)
    email = fields.CharField(max_length=255, unique=True)
    hashed_password = fields.CharField(max_length=255)
    is_admin = fields.BooleanField(default=False)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"

What does the user model store?

Each User has a unique email address. The database stores hashed_password instead of the submitted password.

The is_admin flag supports role checks later. The explicit table name keeps the generated migration predictable.

• Add the owned Event model below the User model:

class Event(Model):
    id = fields.IntField(primary_key=True)
    title = fields.CharField(max_length=200)
    description = fields.TextField(null=True)
    start_time = fields.DatetimeField()
    capacity = fields.IntField()
    creator = fields.ForeignKeyField("models.User", related_name="events")

    class Meta:
        table = "events"

How does event ownership work?

The creator foreign key links each event to one user. Tortoise also exposes the related creator_id value for ownership checks.

• Save models.py.
• Open the View menu in VS Code.
• Select Problems.

You should see no Python syntax errors for the user or event models.

• Add the owned Booking model below the Event model:

class Booking(Model):
    id = fields.IntField(primary_key=True)
    attendee_name = fields.CharField(max_length=200)
    attendee_email = fields.CharField(max_length=255)
    created_at = fields.DatetimeField(auto_now_add=True)
    event = fields.ForeignKeyField("models.Event", related_name="bookings")
    user = fields.ForeignKeyField("models.User", related_name="bookings")

    class Meta:
        table = "bookings"
        constraints = [
            UniqueConstraint(
                fields=("event", "user"),
                name="uid_booking_event_user",
            )
        ]

How does the booking constraint help?

The user foreign key records who owns each booking. The named uid_booking_event_user constraint allows one booking for each user and event pair.

• Save models.py.
• Return to the Problems panel.

You should see no Python syntax errors. Your three database models are ready for migration generation.

Seeing model errors?

Check that User, Event, and Booking each begin at the left edge of models.py.

Confirm that UniqueConstraint is imported from the migration constraints module.

Help me fix my Tortoise model definitions.

Apply the schema migrations

The database models control persistence. Pydantic schemas control the data accepted by your API.

• Select the existing schemas.py file in the Explorer sidebar.
• Replace its contents with the registration plus event and booking schemas below:

from datetime import datetime

from pydantic import BaseModel


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


class Token(BaseModel):
    access_token: str
    token_type: str


class EventCreate(BaseModel):
    title: str
    description: str | None = None
    start_time: datetime
    capacity: int


class EventUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    start_time: datetime | None = None
    capacity: int | None = None


class BookingCreate(BaseModel):
    attendee_name: str
    attendee_email: str

What do these schemas control?

• The UserCreate schema accepts the registration email and password.
• The Token schema defines the future access-token response.
• The event schemas separate required creation values from optional update values.
• The BookingCreate schema accepts attendee details without accepting a client-supplied user ID.

• Save schemas.py.
• Switch back to the activated Windows PowerShell terminal from earlier.
• Initialize the migration package and apply the development schema by running these commands:

tortoise -c settings.TORTOISE_ORM init
tortoise -c settings.TORTOISE_ORM makemigrations --name initial_auth_schema
tortoise -c settings.TORTOISE_ORM migrate

What do these migration commands do?

• The first command initializes the configured migrations package.
• The second command compares the models with migration history and creates the named schema change.
• The third command applies pending migrations to event_booking_auth.

• Confirm that the migrations package now appears in the VS Code Explorer sidebar.

The terminal should report that the development migration completed. The generated filename can differ because Tortoise creates it for you.

Migration command failed?

Confirm that the virtual environment remains active. Check that settings.py and models.py are in the same folder.

A database connection failure usually means one of the PostgreSQL values in your local .env file does not match your installation.

Help me troubleshoot my Tortoise migration.

The same migration history must also exist in the isolated test database. A temporary environment value switches the configuration for this migration run.

• Apply the pending migration to the test database by running these commands:

$env:TESTING="1"
tortoise -c settings.TORTOISE_ORM migrate
Remove-Item Env:TESTING

How does the test migration stay isolated?

Setting TESTING to 1 makes settings.py select TEST_DATABASE_URL.

The migration then targets event_booking_auth_test. The final command removes the temporary value from this PowerShell session.

Both clean databases now share the same migration-managed users, events, and bookings structure. Your original Project 2 database remains untouched.

Add registration and test the boundary

A FastAPI registration route accepts an email and password. The route normalizes the email before creating the user.

The password must cross the API boundary once. Password hashing turns it into a one-way value before the database stores it.

• Select the existing main.py file in the VS Code Explorer sidebar.
• Update the database-related imports and application setup to match this code:

from pwdlib import PasswordHash
from tortoise.contrib.fastapi import register_tortoise

from models import Booking, Event, User
from schemas import BookingCreate, EventCreate, EventUpdate, UserCreate
from settings import TORTOISE_ORM

app = FastAPI(title="Event Booking API with Authentication")
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False)

What does this setup change?

The imports expose the new user model and registration schema. The application now reads its database connection from TORTOISE_ORM.

Setting generate_schemas=False keeps schema changes under migration control.

• Add the password hasher below the application setup:

password_hash = PasswordHash.recommended()


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

How is the password protected?

The recommended password hasher uses the installed Argon2 support. The hash_password() helper returns the stored hash.

The submitted password never becomes a field on the User model.

• Add the public response helpers below the password helper:

def public_user(user: User) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "is_admin": user.is_admin,
        "created_at": user.created_at,
    }


def public_event(event: Event) -> dict:
    return {
        "id": event.id,
        "title": event.title,
        "description": event.description,
        "start_time": event.start_time,
        "capacity": event.capacity,
        "creator_id": event.creator_id,
    }


def public_booking(booking: Booking) -> dict:
    return {
        "id": booking.id,
        "event_id": booking.event_id,
        "user_id": booking.user_id,
        "attendee_name": booking.attendee_name,
        "attendee_email": booking.attendee_email,
        "created_at": booking.created_at,
    }

Why use public response helpers?

Each helper selects the fields that may leave the API. The user response excludes hashed_password completely.

The event and booking helpers expose owner IDs for later authorization checks. They do not accept those IDs from the client.

• Save main.py.
• Return to the VS Code Problems panel.

You should see no Python syntax errors in the application setup or response helpers.

• Add the registration route below the public response helpers:

@app.post("/auth/register")
async def register(payload: UserCreate) -> dict:
    email = payload.email.strip().lower()
    if "@" not in email:
        raise HTTPException(status_code=422, detail="Enter a valid email")
    if len(payload.password) < 8:
        raise HTTPException(
            status_code=422,
            detail="Password must be at least 8 characters",
        )
    if await User.exists(email=email):
        raise HTTPException(status_code=409, detail="Email already registered")

    user = await User.create(
        email=email,
        hashed_password=hash_password(payload.password),
    )
    return public_user(user)

What does registration enforce?

• The route strips surrounding spaces from the email.
• The route converts the email to lowercase before its uniqueness check.
• Invalid email and password values produce 422 Unprocessable Content.
• An existing normalized email produces 409 Conflict.
• A successful response passes through public_user() so the password hash stays private.

• Save main.py.

Seeing import or route errors?

Confirm that PasswordHash, User, and UserCreate are imported before the application setup.

Check that your existing FastAPI import still includes FastAPI and HTTPException.

Help me debug my registration route.

✔️ Awesome, I've got everything!

Great. Double check that settings.py, models.py, schemas.py, and main.py are saved.

Confirm that the generated migrations package remains in your project.

ⓧ I'd like to double check the full code

Check settings.py

import os

from dotenv import load_dotenv

load_dotenv()

TESTING = os.getenv("TESTING") == "1"
DATABASE_URL = (
    os.environ["TEST_DATABASE_URL"] if TESTING else os.environ["DATABASE_URL"]
)
JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES = int(
    os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30")
)

TORTOISE_ORM = {
    "connections": {"default": DATABASE_URL},
    "apps": {
        "models": {
            "models": ["models"],
            "default_connection": "default",
            "migrations": "migrations",
        }
    },
    "use_tz": True,
    "timezone": "UTC",
}

Check models.py

from tortoise import fields
from tortoise.migrations.constraints import UniqueConstraint
from tortoise.models import Model


class User(Model):
    id = fields.IntField(primary_key=True)
    email = fields.CharField(max_length=255, unique=True)
    hashed_password = fields.CharField(max_length=255)
    is_admin = fields.BooleanField(default=False)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"


class Event(Model):
    id = fields.IntField(primary_key=True)
    title = fields.CharField(max_length=200)
    description = fields.TextField(null=True)
    start_time = fields.DatetimeField()
    capacity = fields.IntField()
    creator = fields.ForeignKeyField("models.User", related_name="events")

    class Meta:
        table = "events"


class Booking(Model):
    id = fields.IntField(primary_key=True)
    attendee_name = fields.CharField(max_length=200)
    attendee_email = fields.CharField(max_length=255)
    created_at = fields.DatetimeField(auto_now_add=True)
    event = fields.ForeignKeyField("models.Event", related_name="bookings")
    user = fields.ForeignKeyField("models.User", related_name="bookings")

    class Meta:
        table = "bookings"
        constraints = [
            UniqueConstraint(
                fields=("event", "user"),
                name="uid_booking_event_user",
            )
        ]

Check schemas.py

from datetime import datetime

from pydantic import BaseModel


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


class Token(BaseModel):
    access_token: str
    token_type: str


class EventCreate(BaseModel):
    title: str
    description: str | None = None
    start_time: datetime
    capacity: int


class EventUpdate(BaseModel):
    title: str | None = None
    description: str | None = None
    start_time: datetime | None = None
    capacity: int | None = None


class BookingCreate(BaseModel):
    attendee_name: str
    attendee_email: str

Check the registration setup in main.py

Compare the new application setup in main.py with this reference:

from pwdlib import PasswordHash
from tortoise.contrib.fastapi import register_tortoise

from models import Booking, Event, User
from schemas import BookingCreate, EventCreate, EventUpdate, UserCreate
from settings import TORTOISE_ORM

app = FastAPI(title="Event Booking API with Authentication")
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False)

password_hash = PasswordHash.recommended()


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

def public_user(user: User) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "is_admin": user.is_admin,
        "created_at": user.created_at,
    }


def public_event(event: Event) -> dict:
    return {
        "id": event.id,
        "title": event.title,
        "description": event.description,
        "start_time": event.start_time,
        "capacity": event.capacity,
        "creator_id": event.creator_id,
    }


def public_booking(booking: Booking) -> dict:
    return {
        "id": booking.id,
        "event_id": booking.event_id,
        "user_id": booking.user_id,
        "attendee_name": booking.attendee_name,
        "attendee_email": booking.attendee_email,
        "created_at": booking.created_at,
    }

@app.post("/auth/register")
async def register(payload: UserCreate) -> dict:
    email = payload.email.strip().lower()
    if "@" not in email:
        raise HTTPException(status_code=422, detail="Enter a valid email")
    if len(payload.password) < 8:
        raise HTTPException(
            status_code=422,
            detail="Password must be at least 8 characters",
        )
    if await User.exists(email=email):
        raise HTTPException(status_code=409, detail="Email already registered")

    user = await User.create(
        email=email,
        hashed_password=hash_password(payload.password),
    )
    return public_user(user)

The API is ready for a live registration check. The development server keeps this terminal busy until you stop it.

• Start the API from the activated PowerShell terminal by running this command:

uvicorn main:app --reload

What does this command do?

Uvicorn loads the app object from main.py. Reload mode restarts the development server after saved code changes.

Leave this terminal running during the request checks. Press Ctrl+C when you need to stop the server.

Server did not start?

A startup error naming an environment variable means the local .env file is missing that setting. A database connection error means the selected PostgreSQL URL could not be reached.

Help me fix my FastAPI startup error.

• Switch back to Postman from earlier.
• Create a request in your existing Project 2 collection.
• Set the request method to POST.
• Enter the local API URL from your existing collection with /auth/register appended.
• Select the raw request body editor.
• Select JSON as the body format.
• Enter {"email":" User@Example.com ","password":"password123"} as the request body.

Before you send the request, which submitted field should never appear in the response?

• Click Send.

You should receive 200 OK with id, email, is_admin, and created_at.

The returned email should be user@example.com. Neither the password nor hashed_password should appear.

• Click Send again without changing the request.

You should receive 409 Conflict because the normalized email already exists.

The account now exists in the database. The final check tests whether that stored account already provides reusable proof of identity.

• Create another request in the same Postman collection.
• Set the request method to POST.
• Enter the local API URL from your existing collection with /auth/login appended.

Before you send this request, do you think storing the account also created a login endpoint?

• Click Send.

You should see 404 Not Found. This shortfall is intentional because registration persists an account without issuing reusable proof of identity.

You now have migration-managed identities and safe password storage. Next up, you will issue expiring bearer tokens and turn each token into the current database user.

Issue JWTs and Protect Requests

Your registration route now stores identities safely. The 404 Not Found response from /auth/login exposed the missing piece. Users still cannot prove who they are on later requests.

A JWT access token provides that reusable proof. FastAPI dependencies turn each valid bearer token into the matching database user.

In this step, get ready to:

• Build the reusable authentication module.
• Add OAuth2 password login that returns an expiring bearer token.
• Require bearer authentication for event operations and booking operations.

Build the JWT security layer

The new auth.py module centralizes password handling. It also owns token creation and token validation.

• Return to your Project 2 files in Visual Studio Code.
• Create auth.py beside main.py from the Explorer sidebar.
• Add the authentication imports and shared security objects by pasting this first chunk:

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash

from models import User
from settings import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    JWT_ALGORITHM,
    JWT_SECRET_KEY,
)

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("not-a-real-password")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

What does this code do?

• The PasswordHash.recommended() object applies the recommended Argon2 password-hashing configuration.
• The DUMMY_HASH value lets a failed login perform password verification even when the email does not exist.
• The OAuth2PasswordBearer dependency reads bearer tokens from protected requests.

• Save auth.py.
• Check the Problems panel for unresolved imports.

Your installed authentication packages should resolve every import in this chunk.

Seeing an unresolved import?

Confirm that VS Code is using the interpreter from the existing .venv environment. The authentication packages were installed there in Step 1.

Ask for help with the active interpreter if the warning remains. Help me check why the authentication imports in auth.py are unresolved in my existing virtual environment.

• Add the password helpers and token-signing function below oauth2_scheme by pasting this chunk:

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


def verify_password(password: str, hashed_password: str) -> bool:
    return password_hash.verify(password, hashed_password)


def create_access_token(
    subject: str,
    expires_delta: timedelta | None = None,
) -> str:
    now = datetime.now(timezone.utc)
    expires = expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": subject, "iat": now, "exp": now + expires}
    return jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)

How are tokens created?

• The hash_password() helper converts a plaintext password into a stored hash.
• The verify_password() helper checks a login password against that stored hash.
• The create_access_token() function signs a token with the configured secret.
• The sub claim identifies the user. The exp claim limits the token lifetime.

• Save auth.py.
• Confirm the Problems panel shows no syntax errors in the new helpers.

Seeing a syntax warning?

Check the indentation inside create_access_token(). Confirm that the payload remains inside the function.

Use this prompt if the warning remains. Help me compare my create_access_token function with the intended indentation and imports.

• Add the current-user dependency below create_access_token() by pasting this chunk:

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    credentials_error = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token,
            JWT_SECRET_KEY,
            algorithms=[JWT_ALGORITHM],
            options={"require": ["exp", "sub"]},
        )
        user_id = int(payload["sub"])
    except (InvalidTokenError, KeyError, TypeError, ValueError):
        raise credentials_error

    user = await User.get_or_none(id=user_id)
    if user is None:
        raise credentials_error
    return user

How does a token become a user?

• The dependency receives the bearer token through oauth2_scheme.
• The jwt.decode() call accepts only the configured algorithm.
• The required claim list rejects tokens without exp or sub.
• The integer subject loads the corresponding User record.
• Every invalid path produces 401 Unauthorized with a bearer challenge.

• Save auth.py.
• Confirm the Problems panel shows no errors in get_current_user().

Seeing a type or indentation error?

Confirm that return user stays inside get_current_user(). Check that the database lookup follows the token-decoding block.

Use this prompt for a line-by-line check. Help me debug the get_current_user dependency in auth.py.

• Add the reusable role dependency at the end of auth.py by pasting this chunk:

async def require_admin(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user

Why define this dependency now?

The require_admin() dependency builds on the authenticated user. The next step connects it to the admin route.

• Save auth.py.
• Confirm the complete file has no syntax diagnostics.

• In main.py, replace the import section with this authentication-aware import group:

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from tortoise.contrib.fastapi import register_tortoise

from auth import (
    DUMMY_HASH,
    create_access_token,
    get_current_user,
    hash_password,
    verify_password,
)
from models import Booking, Event, User
from schemas import BookingCreate, EventCreate, EventUpdate, UserCreate
from settings import TORTOISE_ORM

What changes in the imports?

The application now uses one authentication module for registration and login. OAuth2PasswordRequestForm parses the login form required by the OAuth2 password flow.

• Inside register(), replace the existing hashed_password assignment with this final line:

        hashed_password=hash_password(payload.password),

Why change registration?

Registration now uses the shared hash_password() helper. This keeps password hashing consistent across the application.

• Save main.py.
• Return to the PowerShell window from Step 2.
• Start the API with this command if the previous Uvicorn process is no longer running:

uvicorn main:app --reload

What does this command do?

Uvicorn loads app from main.py. Reload mode watches the Python files for saved changes.

You should see the API complete its startup without an import error. Leave this PowerShell window running for the request checks.

Does the API fail during startup?

Check that auth.py sits beside main.py. Confirm that every imported name uses the same spelling in both files.

Use this prompt if startup still fails. Help me trace an import or startup error after connecting auth.py to main.py.

Add the login route

The login route receives form-encoded credentials. It verifies the stored Argon2 hash before issuing a signed access token.

• In main.py, add this route directly below register().

@app.post("/auth/login")
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> dict:
    email = form_data.username.strip().lower()
    user = await User.get_or_none(email=email)
    if user is None:
        verify_password(form_data.password, DUMMY_HASH)
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {
        "access_token": create_access_token(str(user.id)),
        "token_type": "bearer",
    }

How does login stay secure?

• The route treats the OAuth2 username field as the user's email address.
• The normalized email matches registration behavior.
• The dummy hash keeps nonexistent-user checks on the password-verification path.
• A successful response contains only access_token and token_type.

• Save main.py.

Uvicorn should reload the application successfully. The former missing route is now available.

Does the reload fail?

Check that OAuth2PasswordRequestForm appears in the FastAPI security import. Confirm that DUMMY_HASH appears in the auth.py import group.

Use this prompt if the route still fails to load. Help me debug the imports and indentation for my login route.

The login form uses the field name username because that is the OAuth2 password form contract. Your API interprets its value as an email address.

• Return to Postman.
• Return to the POST /auth/login request from Step 2.
• Select the Body tab.
• Select x-www-form-urlencoded.
• Enter username in the first Key field.
• Enter the email you registered in the matching Value field.
• Enter password in the second Key field.
• Enter the same password you used during registration in the matching Value field.

Before you send the request, do you expect the previous 404 Not Found response to remain?

• Send the login request.

You should receive 200 OK. The response body contains access_token and token_type with the value bearer.

That is the missing proof of identity working. Your registered account can now authenticate without exposing password data.

Does login return 401?

Confirm that the email matches your registered account after whitespace is removed. Re-enter the same password used during registration.

Use this prompt if valid credentials still fail. Help me debug why my registered user cannot log in through the OAuth2 form.

Protect event and booking operations

The token becomes useful when protected routes depend on get_current_user(). Public event reads remain open while every event write and booking operation requires identity.

• In main.py, replace the existing create_event() function with this authenticated version:

@app.post("/events")
async def create_event(
    payload: EventCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    if payload.capacity < 1:
        raise HTTPException(status_code=422, detail="Capacity must be positive")
    event = await Event.create(
        title=payload.title,
        description=payload.description,
        start_time=payload.start_time,
        capacity=payload.capacity,
        creator=current_user,
    )
    return public_event(event)

How is the event owner recorded?

The dependency supplies the authenticated User instance. The creator=current_user argument records that server-verified identity on the new event.

• Save main.py.

Uvicorn should reload successfully with the protected event-creation route.

Does event creation fail to load?

Check the comma after the current_user parameter. Confirm that Annotated and Depends are imported.

Use this prompt for focused help. Help me debug the get_current_user dependency on create_event.

• Replace the existing update_event() function with this authenticated version:

@app.put("/events/{event_id}")
async def update_event(
    event_id: int,
    payload: EventUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    if payload.title is not None:
        event.title = payload.title
    if payload.description is not None:
        event.description = payload.description
    if payload.start_time is not None:
        event.start_time = payload.start_time
    if payload.capacity is not None:
        if payload.capacity < 1:
            raise HTTPException(status_code=422, detail="Capacity must be positive")
        event.capacity = payload.capacity
    await event.save()
    return public_event(event)

What changes on event updates?

The update route now requires a valid current user before it reads or changes the event. Ownership permission is the next security boundary you will add.

• Save main.py.

Uvicorn should reload with the authenticated update signature.

Does the update route fail to reload?

Confirm that the existing payload update blocks remain inside update_event(). Check the indentation of await event.save().

Use this prompt if the function still fails. Help me debug the authenticated update_event function.

• Replace the existing delete_event() function with this authenticated version:

@app.delete("/events/{event_id}")
async def delete_event(
    event_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    await event.delete()
    return {"message": "Event deleted"}

What changes on event deletion?

The delete route now resolves a valid current user before deleting an event. A missing or invalid bearer token stops the request first.

• Save main.py.

Uvicorn should reload with the protected event-deletion route.

Does the delete route fail to load?

Confirm that the dependency parameter appears inside delete_event(). Check that the return statement follows the delete call.

Use this prompt for help. Help me debug the authenticated delete_event route.

• Replace the existing create_booking() function with this authenticated version:

@app.post("/events/{event_id}/bookings")
async def create_booking(
    event_id: int,
    payload: BookingCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    if await Booking.exists(event_id=event_id, user_id=current_user.id):
        raise HTTPException(status_code=409, detail="You already booked this event")
    if await Booking.filter(event_id=event_id).count() >= event.capacity:
        raise HTTPException(status_code=409, detail="Event is full")

    booking = await Booking.create(
        event=event,
        user=current_user,
        attendee_name=payload.attendee_name,
        attendee_email=payload.attendee_email,
    )
    return public_booking(booking)

How is booking identity enforced?

The duplicate check now uses the authenticated user's ID. The new booking stores that same user through user=current_user.

• Save main.py.

Uvicorn should reload with user-owned booking creation.

Does booking creation fail to load?

Check the spellings of current_user.id and user=current_user. Confirm that both lines remain inside create_booking().

Use this prompt if the route still fails. Help me debug authenticated booking creation and user ownership.

• Replace the existing booking-list route with this current-user version:

@app.get("/bookings/me")
async def list_my_bookings(
    current_user: Annotated[User, Depends(get_current_user)],
) -> list[dict]:
    bookings = await Booking.filter(user_id=current_user.id).order_by("id")
    return [public_booking(booking) for booking in bookings]

Why filter by the current user?

The route derives the booking owner from the validated token. The client cannot supply a different user ID to read another account's bookings.

• Save main.py.

Uvicorn should reload with the authenticated /bookings/me route.

Does the booking list fail to load?

Confirm that the return type remains list[dict]. Check that the filter uses current_user.id.

Use this prompt if the function still fails. Help me debug the authenticated bookings list.

• Replace the existing delete_booking() function with this authenticated version:

@app.delete("/bookings/{booking_id}")
async def delete_booking(
    booking_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    booking = await Booking.get_or_none(id=booking_id)
    if booking is None:
        raise HTTPException(status_code=404, detail="Booking not found")
    await booking.delete()
    return {"message": "Booking deleted"}

What changes on booking deletion?

The route now requires a validated user before deleting a booking. The next step adds the ownership comparison that decides whether this user may delete the selected record.

• Save main.py.

Uvicorn should complete one more successful reload. All event writes and booking operations now require authentication.

Does the final reload fail?

Read the last traceback entry in PowerShell. Match its function name to the route you most recently replaced.

Use this prompt for targeted help. Help me diagnose the final Uvicorn reload failure after protecting my routes.

✔️ Awesome, I've got everything!

Great. Double check that auth.py and main.py are saved. Confirm that Uvicorn is running without startup errors.

ⓧ I'd like to double check the full code

Compare your complete auth.py file with this reference:

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash

from models import User
from settings import (
    ACCESS_TOKEN_EXPIRE_MINUTES,
    JWT_ALGORITHM,
    JWT_SECRET_KEY,
)

password_hash = PasswordHash.recommended()
DUMMY_HASH = password_hash.hash("not-a-real-password")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")


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


def verify_password(password: str, hashed_password: str) -> bool:
    return password_hash.verify(password, hashed_password)


def create_access_token(
    subject: str,
    expires_delta: timedelta | None = None,
) -> str:
    now = datetime.now(timezone.utc)
    expires = expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": subject, "iat": now, "exp": now + expires}
    return jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> User:
    credentials_error = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token,
            JWT_SECRET_KEY,
            algorithms=[JWT_ALGORITHM],
            options={"require": ["exp", "sub"]},
        )
        user_id = int(payload["sub"])
    except (InvalidTokenError, KeyError, TypeError, ValueError):
        raise credentials_error

    user = await User.get_or_none(id=user_id)
    if user is None:
        raise credentials_error
    return user


async def require_admin(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user

What should this file contain?

The complete module contains password helpers and token signing. It also contains current-user validation plus the role dependency used in the next step.

Compare your cumulative main.py file with this reference:

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from tortoise.contrib.fastapi import register_tortoise

from auth import (
    DUMMY_HASH,
    create_access_token,
    get_current_user,
    hash_password,
    verify_password,
)
from models import Booking, Event, User
from schemas import BookingCreate, EventCreate, EventUpdate, UserCreate
from settings import TORTOISE_ORM

app = FastAPI(title="Event Booking API with Authentication")
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False)


def public_user(user: User) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "is_admin": user.is_admin,
        "created_at": user.created_at,
    }


def public_event(event: Event) -> dict:
    return {
        "id": event.id,
        "title": event.title,
        "description": event.description,
        "start_time": event.start_time,
        "capacity": event.capacity,
        "creator_id": event.creator_id,
    }


def public_booking(booking: Booking) -> dict:
    return {
        "id": booking.id,
        "event_id": booking.event_id,
        "user_id": booking.user_id,
        "attendee_name": booking.attendee_name,
        "attendee_email": booking.attendee_email,
        "created_at": booking.created_at,
    }


@app.post("/auth/register")
async def register(payload: UserCreate) -> dict:
    email = payload.email.strip().lower()
    if "@" not in email:
        raise HTTPException(status_code=422, detail="Enter a valid email")
    if len(payload.password) < 8:
        raise HTTPException(
            status_code=422,
            detail="Password must be at least 8 characters",
        )
    if await User.exists(email=email):
        raise HTTPException(status_code=409, detail="Email already registered")

    user = await User.create(
        email=email,
        hashed_password=hash_password(payload.password),
    )
    return public_user(user)


@app.post("/auth/login")
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> dict:
    email = form_data.username.strip().lower()
    user = await User.get_or_none(email=email)
    if user is None:
        verify_password(form_data.password, DUMMY_HASH)
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {
        "access_token": create_access_token(str(user.id)),
        "token_type": "bearer",
    }


@app.get("/events")
async def list_events() -> list[dict]:
    return [public_event(event) for event in await Event.all().order_by("id")]


@app.get("/events/{event_id}")
async def get_event(event_id: int) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    return public_event(event)


@app.post("/events")
async def create_event(
    payload: EventCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    if payload.capacity < 1:
        raise HTTPException(status_code=422, detail="Capacity must be positive")
    event = await Event.create(
        title=payload.title,
        description=payload.description,
        start_time=payload.start_time,
        capacity=payload.capacity,
        creator=current_user,
    )
    return public_event(event)


@app.put("/events/{event_id}")
async def update_event(
    event_id: int,
    payload: EventUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    if payload.title is not None:
        event.title = payload.title
    if payload.description is not None:
        event.description = payload.description
    if payload.start_time is not None:
        event.start_time = payload.start_time
    if payload.capacity is not None:
        if payload.capacity < 1:
            raise HTTPException(status_code=422, detail="Capacity must be positive")
        event.capacity = payload.capacity
    await event.save()
    return public_event(event)


@app.delete("/events/{event_id}")
async def delete_event(
    event_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    await event.delete()
    return {"message": "Event deleted"}


@app.post("/events/{event_id}/bookings")
async def create_booking(
    event_id: int,
    payload: BookingCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    if await Booking.exists(event_id=event_id, user_id=current_user.id):
        raise HTTPException(status_code=409, detail="You already booked this event")
    if await Booking.filter(event_id=event_id).count() >= event.capacity:
        raise HTTPException(status_code=409, detail="Event is full")

    booking = await Booking.create(
        event=event,
        user=current_user,
        attendee_name=payload.attendee_name,
        attendee_email=payload.attendee_email,
    )
    return public_booking(booking)


@app.get("/bookings/me")
async def list_my_bookings(
    current_user: Annotated[User, Depends(get_current_user)],
) -> list[dict]:
    bookings = await Booking.filter(user_id=current_user.id).order_by("id")
    return [public_booking(booking) for booking in bookings]


@app.delete("/bookings/{booking_id}")
async def delete_booking(
    booking_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    booking = await Booking.get_or_none(id=booking_id)
    if booking is None:
        raise HTTPException(status_code=404, detail="Booking not found")
    await booking.delete()
    return {"message": "Booking deleted"}

What should this file prove?

The cumulative application keeps event reads public. Registration and login are available while event writes and booking operations depend on the authenticated user.

Your final check proves both sides of authentication. A valid token creates an owned event while a missing token receives the correct challenge.

• Copy the access_token value from the successful login response.
• Create a new POST /events request in Postman.
• Select the request's Authorization tab.
• Select Bearer Token from the Auth Type list.
• Paste the copied access token into the Token field.
• Select the Body tab.
• Select raw.
• Select JSON as the body format.
• Paste this event body:

{
    "title": "Security Workshop",
    "description": "A protected event",
    "start_time": "2026-12-01T10:00:00Z",
    "capacity": 20
}

What does this request create?

The body supplies the event fields. The API supplies creator_id from the bearer token after the request is authenticated.

Before you send this request, do you expect the API to trust the event body or the bearer token for the creator identity?

• Send the authenticated event request.

You should receive 200 OK. The response includes a creator_id that matches the user represented by your token.

• Return to the same POST /events request.
• Select the Authorization tab.
• Select No Auth from the Auth Type list.

Before you send the request again, which response do you predict once the bearer token is removed?

• Send the event request without authentication.

You should receive 401 Unauthorized. This proves the dependency blocks the request before event creation runs.

That closes the authentication loop. Your API now registers users, issues expiring tokens, resolves current users, and blocks unauthenticated writes.

Does the request succeed without a token?

Confirm that create_event() includes the Depends(get_current_user) parameter. Save main.py before sending the request again.

Use this prompt if the route still allows anonymous access. Help me trace why POST /events is not enforcing bearer authentication.

Identity now follows every protected request into your database operations. Next, you will decide what each authenticated user is allowed to change.

Enforce Ownership and Admin Access

Your FastAPI API now turns a JWT bearer token into a database user. That completes authentication for protected requests.

Permission is the next boundary. Authorization must compare that user with each resource owner or required role before a write proceeds. A valid token proves identity but does not grant every action.

In this step, get ready to:

• Restrict event updates or deletions to the event creator.
• Prevent duplicate bookings or deletions by another user.
• Protect the users list with the existing admin dependency.

Enforce event ownership

The authenticated user is available as current_user. Each event also stores its owner through creator_id.

Comparing those IDs keeps event writes under the creator's control. Missing events still return 404 Not Found before the ownership check runs.

• In the Explorer sidebar in Visual Studio Code, select main.py.
• Find the missing-event check inside update_event().
• Add this ownership check immediately below the missing-event check:

    if event.creator_id != current_user.id:
        raise HTTPException(status_code=403, detail="Only the creator can edit this event")

How does the update check work?

• The comparison checks whether the authenticated user's ID matches the event's stored creator ID.
• A mismatch raises 403 Forbidden because the request has a valid identity without ownership permission.

• Find the missing-event check inside delete_event().
• Add this ownership check immediately below the missing-event check:

    if event.creator_id != current_user.id:
        raise HTTPException(status_code=403, detail="Only the creator can delete this event")

How does the delete check work?

• The check protects deletion with the same creator comparison used for updates.
• The API reaches event.delete() only after the ownership check passes.

• Save main.py.
• Start the development server in your project PowerShell terminal if the earlier server process is stopped by running:

uvicorn main:app --reload

What does this command do?

Uvicorn runs the app object from main.py. The reload option applies saved Python changes during local development.

The ownership boundary needs two identities to prove it works. You will use Postman with owner@example.com as User A and other@example.com as User B.

• Return to your existing Postman collection.
• Register User A through your existing POST /auth/register request using owner@example.com with password password123.
• Register User B through the same request using other@example.com with password password123.
• Log in User A through your existing POST /auth/login request.
• Copy User A's access_token from the response.
• Paste User A's token after Bearer in the Authorization header of your existing POST /events request.
• Send the event creation request.
• Record the returned id here: your event ID.

You should see 200 OK with a creator_id for User A. The server has recorded ownership from the token.

• Set your existing PUT /events/{event_id} request to use your event ID.
• Keep User A's bearer token in the request.
• Set title to Creator Updated Event.
• Send the update request.

You should see 200 OK with the updated title. User A can change the event because User A owns it.

• Log in User B through your existing POST /auth/login request.
• Copy User B's access_token from the response.
• Replace the bearer token in the event update request with User B's token.
• Set title to Other User Update.

Before you send the request, predict whether User B's valid token is enough to edit User A's event.

• Send the event update request as User B.

You should see 403 Forbidden. The token identifies User B successfully while the ownership check refuses the update.

Did User B update the event?

Confirm that the event was created with User A's token. Check that the failed update used User B's token.

Make sure the creator_id comparison sits inside update_event() before any fields change.

Help me debug my event ownership check.

Protect booking boundaries

Booking authorization has two parts. The API limits one booking per authenticated user for each event.

The booking owner also controls deletion. Capacity remains a separate conflict check against the event's limit.

• In main.py, find the missing-event check inside create_booking().
• Add these conflict checks immediately below the missing-event check:

    if await Booking.exists(event_id=event_id, user_id=current_user.id):
        raise HTTPException(status_code=409, detail="You already booked this event")
    if await Booking.filter(event_id=event_id).count() >= event.capacity:
        raise HTTPException(status_code=409, detail="Event is full")

How are booking conflicts handled?

• The first query checks whether the current user already owns a booking for the event.
• The second query compares the current booking count with the event capacity.
• Either conflict returns 409 Conflict before a new booking is stored.

• Find the missing-booking check inside delete_booking().
• Add this ownership check immediately below the missing-booking check:

    if booking.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="You do not own this booking")

How is booking deletion protected?

• The booking's stored user_id identifies its owner.
• A different authenticated user receives 403 Forbidden before deletion runs.

• Save main.py.
• Set your existing POST /events/{event_id}/bookings request to use your event ID.
• Use User B's bearer token in the booking request.
• Send the booking request once.
• Record the returned booking id here: your booking ID.

You should see 200 OK with User B's user_id in the booking response.

Before you repeat the request, predict how the API handles a second booking from the same authenticated user.

• Send the same booking request again with User B's token.

You should see 409 Conflict. The first booking remains unchanged.

• Set your existing DELETE /bookings/{booking_id} request to use your booking ID.
• Replace User B's token with User A's token.

Before you send the deletion request, predict whether the event creator also owns User B's booking.

• Send the booking deletion request as User A.

You should see 403 Forbidden. Event ownership does not transfer ownership of another user's booking.

Did the duplicate booking succeed?

Confirm that both booking requests use the same event ID. Confirm that both requests use User B's token.

Check that the duplicate query filters with both event_id and current_user.id.

Help me debug my booking conflict checks.

Prove admin access rules

The existing require_admin() dependency builds on get_current_user(). It first resolves the authenticated user before checking is_admin.

This layered dependency keeps role policy reusable. The admin route can focus on returning public user fields.

• Find the import group from auth at the top of main.py.
• Add this dependency to the import group:

    require_admin,

Why import this dependency?

The route passes require_admin to FastAPI's dependency system. The dependency returns the user only after authentication and the role check succeed.

• Add the admin users route at the bottom of main.py by copying this code:

@app.get("/admin/users")
async def list_users(
    admin: Annotated[User, Depends(require_admin)],
) -> list[dict]:
    return [public_user(user) for user in await User.all().order_by("id")]

How does the admin route stay safe?

• The dependency blocks normal authenticated users before the database query runs.
• The route passes every user through public_user() so password hashes stay out of the response.

• Save main.py.
• Create a GET /admin/users request in Postman.
• Use User A's bearer token in the request.

Before you send the request, predict whether a valid token from a normal user satisfies the admin role check.

• Send the admin users request as User A.

You should see 403 Forbidden. That confirms the API distinguishes an authenticated user from an authorized administrator.

Can a normal user see the users list?

Confirm that the route depends on require_admin instead of get_current_user directly. Check that User A still has the default is_admin value.

Help me debug my admin dependency.

✔️ Awesome, I've got everything!

Your API now enforces event ownership, booking ownership, duplicate-booking conflicts, capacity limits, and admin-only access. Save main.py before moving on.

ⓧ I'd like to double check the full code

Compare your main.py file with this complete version.

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from tortoise.contrib.fastapi import register_tortoise

from auth import (
    DUMMY_HASH,
    create_access_token,
    get_current_user,
    hash_password,
    require_admin,
    verify_password,
)
from models import Booking, Event, User
from schemas import BookingCreate, EventCreate, EventUpdate, UserCreate
from settings import TORTOISE_ORM

app = FastAPI(title="Event Booking API with Authentication")
register_tortoise(app, config=TORTOISE_ORM, generate_schemas=False)


def public_user(user: User) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "is_admin": user.is_admin,
        "created_at": user.created_at,
    }


def public_event(event: Event) -> dict:
    return {
        "id": event.id,
        "title": event.title,
        "description": event.description,
        "start_time": event.start_time,
        "capacity": event.capacity,
        "creator_id": event.creator_id,
    }


def public_booking(booking: Booking) -> dict:
    return {
        "id": booking.id,
        "event_id": booking.event_id,
        "user_id": booking.user_id,
        "attendee_name": booking.attendee_name,
        "attendee_email": booking.attendee_email,
        "created_at": booking.created_at,
    }


@app.post("/auth/register")
async def register(payload: UserCreate) -> dict:
    email = payload.email.strip().lower()
    if "@" not in email:
        raise HTTPException(status_code=422, detail="Enter a valid email")
    if len(payload.password) < 8:
        raise HTTPException(
            status_code=422,
            detail="Password must be at least 8 characters",
        )
    if await User.exists(email=email):
        raise HTTPException(status_code=409, detail="Email already registered")

    user = await User.create(
        email=email,
        hashed_password=hash_password(payload.password),
    )
    return public_user(user)


@app.post("/auth/login")
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> dict:
    email = form_data.username.strip().lower()
    user = await User.get_or_none(email=email)
    if user is None:
        verify_password(form_data.password, DUMMY_HASH)
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {
        "access_token": create_access_token(str(user.id)),
        "token_type": "bearer",
    }


@app.get("/events")
async def list_events() -> list[dict]:
    return [public_event(event) for event in await Event.all().order_by("id")]


@app.get("/events/{event_id}")
async def get_event(event_id: int) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    return public_event(event)


@app.post("/events")
async def create_event(
    payload: EventCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    if payload.capacity < 1:
        raise HTTPException(status_code=422, detail="Capacity must be positive")
    event = await Event.create(
        title=payload.title,
        description=payload.description,
        start_time=payload.start_time,
        capacity=payload.capacity,
        creator=current_user,
    )
    return public_event(event)


@app.put("/events/{event_id}")
async def update_event(
    event_id: int,
    payload: EventUpdate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    if event.creator_id != current_user.id:
        raise HTTPException(status_code=403, detail="Only the creator can edit this event")

    if payload.title is not None:
        event.title = payload.title
    if payload.description is not None:
        event.description = payload.description
    if payload.start_time is not None:
        event.start_time = payload.start_time
    if payload.capacity is not None:
        if payload.capacity < 1:
            raise HTTPException(status_code=422, detail="Capacity must be positive")
        event.capacity = payload.capacity
    await event.save()
    return public_event(event)


@app.delete("/events/{event_id}")
async def delete_event(
    event_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    if event.creator_id != current_user.id:
        raise HTTPException(status_code=403, detail="Only the creator can delete this event")
    await event.delete()
    return {"message": "Event deleted"}


@app.post("/events/{event_id}/bookings")
async def create_booking(
    event_id: int,
    payload: BookingCreate,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    event = await Event.get_or_none(id=event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    if await Booking.exists(event_id=event_id, user_id=current_user.id):
        raise HTTPException(status_code=409, detail="You already booked this event")
    if await Booking.filter(event_id=event_id).count() >= event.capacity:
        raise HTTPException(status_code=409, detail="Event is full")

    booking = await Booking.create(
        event=event,
        user=current_user,
        attendee_name=payload.attendee_name,
        attendee_email=payload.attendee_email,
    )
    return public_booking(booking)


@app.get("/bookings/me")
async def list_my_bookings(
    current_user: Annotated[User, Depends(get_current_user)],
) -> list[dict]:
    bookings = await Booking.filter(user_id=current_user.id).order_by("id")
    return [public_booking(booking) for booking in bookings]


@app.delete("/bookings/{booking_id}")
async def delete_booking(
    booking_id: int,
    current_user: Annotated[User, Depends(get_current_user)],
) -> dict:
    booking = await Booking.get_or_none(id=booking_id)
    if booking is None:
        raise HTTPException(status_code=404, detail="Booking not found")
    if booking.user_id != current_user.id:
        raise HTTPException(status_code=403, detail="You do not own this booking")
    await booking.delete()
    return {"message": "Booking deleted"}


@app.get("/admin/users")
async def list_users(
    admin: Annotated[User, Depends(require_admin)],
) -> list[dict]:
    return [public_user(user) for user in await User.all().order_by("id")]

How to use this reference

• Compare the ownership checks, booking conflict checks, auth imports, and admin route with your saved file.

Your API now separates identity from permission across events, bookings, and admin access. Next, you will turn these manual security checks into repeatable automated tests.

Prove the Security Boundaries

Your Postman flow proved that the API accepts an event owner while rejecting another authenticated user. Manual checks provide useful evidence during development.

This step turns those checks into a repeatable pytest suite. The suite protects authentication behavior from regressions while keeping every test inside the separate test database.

In this step, get ready to:

• Configure an async test harness with application lifespan support.
• Test registration, login, missing authentication, and ownership rejection.
• Override the current-user dependency to prove admin access.

Configure the async test harness

Async API tests need the same startup and shutdown behavior as the running application. A lifespan-aware client starts the FastAPI integration before each test uses the database.

The cleanup fixture deletes bookings before events. It deletes events before users because those tables contain foreign-key relationships.

• Switch back to the Project 2 folder in Visual Studio Code.
• Create pytest.ini inside the Project 2 folder with the new-file control in the Explorer sidebar.
• Add the test discovery configuration by pasting this code into pytest.ini:

[pytest]
asyncio_mode = auto
testpaths = tests

What does this configuration do?

• The asyncio_mode = auto setting lets pytest run asynchronous test functions without decorating each one.
• The testpaths = tests setting keeps test discovery focused on the tests folder.

• Save pytest.ini.

You should see pytest.ini listed beside the existing application files in the Explorer sidebar.

• Create a tests folder inside the Project 2 folder with the new-folder control in the Explorer sidebar.
• Create conftest.py inside the new tests folder.
• Add the test environment and async client fixture by pasting this code into tests/conftest.py:

import os

import httpx
import pytest_asyncio
from asgi_lifespan import LifespanManager

os.environ["TESTING"] = "1"
os.environ.setdefault(
    "JWT_SECRET_KEY",
    "test-secret-key-that-is-not-used-outside-tests",
)

from main import app
from models import Booking, Event, User


@pytest_asyncio.fixture
async def client():
    async with LifespanManager(app) as manager:
        transport = httpx.ASGITransport(app=manager.app)
        async with httpx.AsyncClient(
            transport=transport,
            base_url="http://testserver",
        ) as async_client:
            yield async_client

How does the test client work?

• The test environment is selected before importing main. This makes settings.py choose TEST_DATABASE_URL.
• The test-only signing value gives token code a predictable local secret without exposing the secret from .env.
• LifespanManager starts the application lifecycle so the database integration is ready.
• HTTPX sends requests directly to the application through ASGITransport.

• Save tests/conftest.py.

You should see conftest.py inside the tests folder in the Explorer sidebar.

• Add the cleanup fixture below the client fixture in tests/conftest.py by pasting this code:

@pytest_asyncio.fixture(autouse=True)
async def clean_database(client):
    await Booking.all().delete()
    await Event.all().delete()
    await User.all().delete()
    app.dependency_overrides = {}
    yield
    app.dependency_overrides = {}
    await Booking.all().delete()
    await Event.all().delete()
    await User.all().delete()

Why clean before and after each test?

The fixture starts every test with empty booking, event, and user tables. It repeats the cleanup afterward so a failed test cannot leave records for the next one.

Resetting app.dependency_overrides keeps an overridden identity inside the test that requested it. Each test must prove its own security state.

• Save tests/conftest.py.

The completed file now contains the client fixture followed by the automatic clean_database fixture.

Does the test harness fail to load?

Check that TESTING is set before the import from main. Importing the application first can select the development database.

Confirm that tests/conftest.py is inside the tests folder. Pytest discovers shared fixtures from that location.

Help me diagnose my async pytest fixture setup.

Test authentication and ownership failures

The first test group covers identity entry points and rejected event access. Each assertion records an HTTP boundary that future changes must preserve.

• Switch back to the PowerShell prompt for the Project 2 folder.
• Select the test database for this PowerShell session by running:

$env:TESTING="1"

What does this setting do?

The environment flag makes the application select TEST_DATABASE_URL from settings.py. This keeps test cleanup away from the development database.

• Return to Visual Studio Code.
• Create test_auth.py inside the existing tests folder.
• Add the request helpers by pasting this code into tests/test_auth.py:

from auth import get_current_user, hash_password
from main import app
from models import User


async def register(client, email: str, password: str = "password123"):
    return await client.post(
        "/auth/register",
        json={"email": email, "password": password},
    )


async def login(client, email: str, password: str = "password123"):
    return await client.post(
        "/auth/login",
        data={"username": email, "password": password},
    )


def bearer(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}

What do these helpers provide?

• The register helper sends the account payload used across the suite.
• The login helper puts the email into the OAuth2 username form field.
• The bearer helper creates the authorization header used by protected requests.

• Save tests/test_auth.py.

You should see test_auth.py beside conftest.py in the tests folder.

• Add the registration test below the helper functions by pasting this code:

async def test_registration_and_duplicate_email(client):
    first = await register(client, "user@example.com")
    duplicate = await register(client, "user@example.com")

    assert first.status_code == 200
    assert first.json()["email"] == "user@example.com"
    assert "hashed_password" not in first.json()
    assert duplicate.status_code == 409

What does this test prove?

The first request proves registration returns the normalized public email. The password hash stays outside the response.

The repeated email must produce a conflict response. This protects the unique identity boundary.

• Save tests/test_auth.py.
• Run the current test suite from the Project 2 folder by running:

pytest -q

What should you see?

You should see the registration test pass. That result proves the application writes to the test database and hides the stored password hash.

Does the registration test fail?

Confirm that the clean test database already contains the migrated schema. The suite expects the users table to exist before it starts.

Check the deletion order in clean_database. Bookings must be deleted before events and users.

Help me debug the registration test failure.

• Add the login test below the registration test by pasting this code:

async def test_login_failure_and_success(client):
    await register(client, "user@example.com")

    failed = await login(client, "user@example.com", "wrong-password")
    successful = await login(client, "user@example.com")

    assert failed.status_code == 401
    assert successful.status_code == 200
    assert successful.json()["token_type"] == "bearer"
    assert successful.json()["access_token"]

What does the login test prove?

The failed request proves an incorrect password cannot produce a token. The successful request proves the same account can receive a bearer access token with the correct password.

• Save tests/test_auth.py.
• Run the expanded suite by running:

pytest -q

What should you see now?

You should see both tests pass. Login now has repeatable evidence for its rejected and successful paths.

Does the login test fail?

Check that the form payload uses username for the email. The login route reads that OAuth2 field before normalizing it.

Confirm that the password values match the helper defaults. The successful request depends on the password stored during registration.

Help me debug the login test failure.

• Add the missing-token test below the login test by pasting this code:

async def test_missing_token_is_unauthorized(client):
    response = await client.post(
        "/events",
        json={
            "title": "Security Workshop",
            "description": "A protected event",
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )

    assert response.status_code == 401

What does this test prove?

The request contains a valid event payload without an authorization header. The current-user dependency must reject it before an event is created.

• Save tests/test_auth.py.
• Run the expanded suite by running:

pytest -q

What should you see now?

You should see all current tests pass. The protected event route now has an automated check for missing authentication.

Does the missing-token test fail?

Check that create_event still depends on get_current_user. Removing that dependency would allow the request to reach the route body.

Help me trace the missing-token response.

• Add the non-owner test below the missing-token test by pasting this code:

async def test_non_owner_cannot_edit_event(client):
    await register(client, "owner@example.com")
    await register(client, "other@example.com")
    owner_token = (await login(client, "owner@example.com")).json()["access_token"]
    other_token = (await login(client, "other@example.com")).json()["access_token"]

    created = await client.post(
        "/events",
        headers=bearer(owner_token),
        json={
            "title": "Owner Event",
            "description": None,
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )
    response = await client.put(
        f"/events/{created.json()['id']}",
        headers=bearer(other_token),
        json={"title": "Stolen Event"},
    )

    assert response.status_code == 403

What does the ownership test prove?

The owner creates the event with one valid token. A second valid token identifies a different user.

The rejected update proves authentication alone does not grant permission. The creator check enforces ownership after identity is known.

• Save tests/test_auth.py.
• Run the expanded suite by running:

pytest -q

What should you see now?

You should see every current test pass. The suite now distinguishes missing authentication from an authenticated user without ownership.

Does the ownership test fail?

Confirm that the event creation request uses owner_token. The update request must use other_token.

Check that the created response contains an event ID before the update URL is built.

Help me debug the non-owner event test.

Prove admin access and run the suite

A dependency override lets a test supply a controlled identity without changing production authentication code. This test uses that seam to provide an admin user directly to get_current_user.

The same test also proves that an ordinary authenticated user remains blocked. The fixture resets the override afterward so later tests cannot inherit admin access.

• Add the owner and admin test below the non-owner test by pasting this first part:

async def test_owner_success_and_admin_dependency_override(client):
    await register(client, "owner@example.com")
    owner_token = (await login(client, "owner@example.com")).json()["access_token"]

    created = await client.post(
        "/events",
        headers=bearer(owner_token),
        json={
            "title": "Owner Event",
            "description": None,
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )
    updated = await client.put(
        f"/events/{created.json()['id']}",
        headers=bearer(owner_token),
        json={"title": "Updated Owner Event"},
    )
    denied = await client.get("/admin/users", headers=bearer(owner_token))

What does this part set up?

The owner creates an event before updating the same resource with the same token. The normal-user request to /admin/users captures the role-based rejection for the final assertions.

• Append the admin override and assertions after the denied response by pasting this code:

    admin = await User.create(
        email="admin@example.com",
        hashed_password=hash_password("password123"),
        is_admin=True,
    )

    async def override_current_user():
        return admin

    app.dependency_overrides[get_current_user] = override_current_user
    allowed = await client.get("/admin/users")

    assert created.status_code == 200
    assert updated.status_code == 200
    assert updated.json()["title"] == "Updated Owner Event"
    assert denied.status_code == 403
    assert allowed.status_code == 200
    assert any(user["email"] == "admin@example.com" for user in allowed.json())

How does the override prove admin access?

• The test creates an admin user with a valid password hash.
• The override returns that user whenever the application requests the current identity.
• The admin request succeeds without weakening require_admin in production code.
• The final assertion confirms the public admin response includes the admin email.

• Save tests/test_auth.py.

✔️ Awesome, I've got everything!

Great. Double check that pytest.ini, tests/conftest.py, and tests/test_auth.py are saved.

ⓧ I'd like to double check the full code

Compare your complete pytest.ini with this file:

[pytest]
asyncio_mode = auto
testpaths = tests

This file enables automatic async test handling and limits discovery to the tests folder.

Compare your complete tests/conftest.py with this file:

import os

import httpx
import pytest_asyncio
from asgi_lifespan import LifespanManager

os.environ["TESTING"] = "1"
os.environ.setdefault(
    "JWT_SECRET_KEY",
    "test-secret-key-that-is-not-used-outside-tests",
)

from main import app
from models import Booking, Event, User


@pytest_asyncio.fixture
async def client():
    async with LifespanManager(app) as manager:
        transport = httpx.ASGITransport(app=manager.app)
        async with httpx.AsyncClient(
            transport=transport,
            base_url="http://testserver",
        ) as async_client:
            yield async_client


@pytest_asyncio.fixture(autouse=True)
async def clean_database(client):
    await Booking.all().delete()
    await Event.all().delete()
    await User.all().delete()
    app.dependency_overrides = {}
    yield
    app.dependency_overrides = {}
    await Booking.all().delete()
    await Event.all().delete()
    await User.all().delete()

This file selects the test environment before importing the application. It also provides the async client and automatic cleanup fixtures.

Compare your complete tests/test_auth.py with this file:

from auth import get_current_user, hash_password
from main import app
from models import User


async def register(client, email: str, password: str = "password123"):
    return await client.post(
        "/auth/register",
        json={"email": email, "password": password},
    )


async def login(client, email: str, password: str = "password123"):
    return await client.post(
        "/auth/login",
        data={"username": email, "password": password},
    )


def bearer(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


async def test_registration_and_duplicate_email(client):
    first = await register(client, "user@example.com")
    duplicate = await register(client, "user@example.com")

    assert first.status_code == 200
    assert first.json()["email"] == "user@example.com"
    assert "hashed_password" not in first.json()
    assert duplicate.status_code == 409


async def test_login_failure_and_success(client):
    await register(client, "user@example.com")

    failed = await login(client, "user@example.com", "wrong-password")
    successful = await login(client, "user@example.com")

    assert failed.status_code == 401
    assert successful.status_code == 200
    assert successful.json()["token_type"] == "bearer"
    assert successful.json()["access_token"]


async def test_missing_token_is_unauthorized(client):
    response = await client.post(
        "/events",
        json={
            "title": "Security Workshop",
            "description": "A protected event",
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )

    assert response.status_code == 401


async def test_non_owner_cannot_edit_event(client):
    await register(client, "owner@example.com")
    await register(client, "other@example.com")
    owner_token = (await login(client, "owner@example.com")).json()["access_token"]
    other_token = (await login(client, "other@example.com")).json()["access_token"]

    created = await client.post(
        "/events",
        headers=bearer(owner_token),
        json={
            "title": "Owner Event",
            "description": None,
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )
    response = await client.put(
        f"/events/{created.json()['id']}",
        headers=bearer(other_token),
        json={"title": "Stolen Event"},
    )

    assert response.status_code == 403


async def test_owner_success_and_admin_dependency_override(client):
    await register(client, "owner@example.com")
    owner_token = (await login(client, "owner@example.com")).json()["access_token"]

    created = await client.post(
        "/events",
        headers=bearer(owner_token),
        json={
            "title": "Owner Event",
            "description": None,
            "start_time": "2026-12-01T10:00:00Z",
            "capacity": 20,
        },
    )
    updated = await client.put(
        f"/events/{created.json()['id']}",
        headers=bearer(owner_token),
        json={"title": "Updated Owner Event"},
    )
    denied = await client.get("/admin/users", headers=bearer(owner_token))

    admin = await User.create(
        email="admin@example.com",
        hashed_password=hash_password("password123"),
        is_admin=True,
    )

    async def override_current_user():
        return admin

    app.dependency_overrides[get_current_user] = override_current_user
    allowed = await client.get("/admin/users")

    assert created.status_code == 200
    assert updated.status_code == 200
    assert updated.json()["title"] == "Updated Owner Event"
    assert denied.status_code == 403
    assert allowed.status_code == 200
    assert any(user["email"] == "admin@example.com" for user in allowed.json())

This file contains the shared request helpers and every authentication or authorization test added in this step.

Before you run the completed suite, which boundaries do you expect to pass without any manual Postman setup?

• Run the completed security suite from the Project 2 folder by running:

pytest -q

What should you see?

You should see every collected test pass. The result confirms registration, login, missing authentication, ownership rejection, owner success, role rejection, cleanup, and the admin dependency override.

Does the complete suite fail?

Read the first failed assertion because later failures can be side effects of the same setup problem. Confirm that PowerShell is inside the Project 2 folder before rerunning the suite.

Check that the test database has the migrated tables. The fixtures clean existing tables but do not create the schema.

Help me diagnose the failing security suite.

• Remove the temporary PowerShell test flag by running:

Remove-Item Env:TESTING

Why remove the flag?

Removing the session flag returns later commands to the normal environment selection. The test fixture still sets TESTING before importing the application whenever pytest runs.

That is the full security loop complete. Your API now has repeatable proof that identity, ownership, admin roles, database isolation, and dependency wiring work together.

Secret mission

Reject an Expired Token

A signed JWT can still be unsafe after its lifetime ends. Add a regression test that creates an already-expired token and proves the protected bookings route rejects it with 401 Unauthorized.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep, pause, or delete the local resources from this authentication build. This project uses no paid services, so none of these choices creates an ongoing charge.

Resources you used:

• A local Uvicorn development server process.
• The PostgreSQL development database named event_booking_auth.
• The PostgreSQL test database named event_booking_auth_test.
• The local .env file containing JWT_SECRET_KEY.
• Saved authentication proof requests in Postman.

Keep everything running

No action needed. Choose this if you are still developing the authenticated API or repeating its security checks.

• Keep Uvicorn running while you test requests.
• Keep event_booking_auth available for development data.
• Keep event_booking_auth_test available for isolated test runs.
• Keep .env outside source control.
• Retain your Postman requests for future authentication checks.
• Keep .env.example in source control.
• Keep the generated migrations package in source control.

Pause - I'll come back to this later

Shut down local processes to free memory. Your databases and project artifacts stay ready for your return.

• Stop the running Uvicorn process with the interrupt control in its PowerShell terminal.
• Stop your local PostgreSQL service through your normal administration workflow.
• Leave event_booking_auth in place.
• Leave event_booking_auth_test in place.
• Leave .env in the existing Project 2 codebase.
• Keep your saved Postman requests.

Delete - I don't want to use this again

Remove the runtime state created for this authentication build. Keep the reusable non-secret source artifacts in version control.

• Stop the running Uvicorn process with the interrupt control in its PowerShell terminal.
• Delete event_booking_auth through your normal PostgreSQL administration workflow.
• Delete event_booking_auth_test through your normal PostgreSQL administration workflow.
• Refresh the database list in the same administration workflow.

You should no longer see event_booking_auth or event_booking_auth_test in the database list.

• Return to the Postman workspace from earlier.
• Remove each saved authentication request with its deletion control.

Your Postman workspace should no longer show the saved requests from this authentication project.

• Return to the project files from earlier.
• Delete .env with your editor's file deletion control.

You should still see .env.example. The secret-bearing .env file should be gone.

• Retain .env.example in source control.
• Retain the generated migrations package in source control.
• Retain tests/test_auth.py in source control.
• Retain tests/test_expired_token.py in source control.

Nice Work!

Nice Work!

You did it! You've turned the Project 2 event-booking API into a multi-user FastAPI service with expiring JWT bearer tokens. Every protected operation now links a verified identity to a tested permission rule.

You've learned how to:

• Build persistent user identities with Tortoise ORM migrations against PostgreSQL. Each account stores an Argon2 password hash. Events and bookings retain their authenticated owners.
• Implement an OAuth2 login flow that accepts an email through the username form field. The flow returns an expiring bearer token. The reusable get_current_user dependency validates each token before protected operations run.
• Enforce creator ownership for event changes. Enforce booking ownership for booking deletion. Add admin-only access through a reusable dependency. Prove each boundary with Postman plus async pytest tests.
• Complete the Secret Mission by adding an expired-token regression test in tests/test_expired_token.py. The test creates an already-expired token through create_access_token(). A request to /bookings/me proves the API rejects that token with 401 Unauthorized.

Ready to quiz yourself?