Build an Async Event API with PostgreSQL

Build an async event API with PostgreSQL, migrations, CRUD, and rollback tests.

Introduction

30 Second Summary

Event details rarely stay fixed. When locations or capacities change, scattered notes make it hard to know which version is current.

In this project, you will build an async Event API with FastAPI backed by PostgreSQL. Version-controlled Alembic migrations will make its schema repeatable across development and test databases.

What You'll Build

Your finished service lets you create an event in Swagger UI while pgAdmin 4 shows the matching row in PostgreSQL.

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

  • A live database health check where opening /health displays {"database": "connected"} after a real PostgreSQL query.
  • A full event lifecycle you can drive through Swagger UI. Each updated row stays visible in pgAdmin 4.
  • Version-controlled schemas for separate development and test databases. A rollback-isolated async test that returns the test database to its original row count.
  • Secret Mission: Protect an event from deletion when a booking refers to it.

Do I need any experience before starting?

You need a Windows device with PostgreSQL server and pgAdmin 4 already installed. Step 1 verifies Python 3.11 or later plus Visual Studio Code, with setup branches if either tool is missing.

Before We Start

This step marks your commitment to an async Event API that manages event records through HTTP requests backed by PostgreSQL. Its separate test database must remove test rows after each run so integration tests remain independent.

Set Up the Toolchain and Databases

Your API needs a known Python runtime before it can make a reliable database connection. It also needs two fixed PostgreSQL targets.

An isolated environment keeps this project's packages separate inside Visual Studio Code. This step prepares one database for development plus one database for rollback-safe tests.

In this step, get ready to:
  • Verify the Windows toolchain from PowerShell.
  • Create an isolated workspace with pinned dependencies.
  • Create separate development and test databases.
Verify Python and Visual Studio Code

The pinned SQLAlchemy release requires Python 3.11 or later. PowerShell gives you a quick way to check the runtime available to this project.

  • Press the Windows key to open Windows search.
  • Type PowerShell into the search bar.
  • Press Enter to open PowerShell.
  • Check your Python version by running this command:
python --version

What does this command check?

The command asks the Python runtime on your Windows path to report its version. This project needs version 3.11 or later.

✔️ I see version 3.11 or higher

Your Python runtime meets the project's minimum version. Keep this PowerShell window available for the workspace commands.

ⓧ I see an older version

The installed runtime cannot use the pinned SQLAlchemy release. Install a current CPython runtime before continuing.

  • Open Python's official Windows setup guide.
  • Install a current CPython runtime with the Python install manager.
  • Close the existing PowerShell window after installation.
  • Start a new PowerShell window from Windows search.
  • Return to the version check above.

ⓧ Command not found

Windows cannot find a Python runtime on your path. The Python install manager provides the global command this project uses.

  • Open Python's official Windows setup guide.
  • Install a current CPython runtime with the Python install manager.
  • Close the existing PowerShell window after installation.
  • Start a new PowerShell window from Windows search.
  • Return to the version check above.

The editor check confirms that Windows can launch Visual Studio Code from the same PowerShell session. This command becomes useful when you open the project folder.

  • Check the Visual Studio Code command by running:
code --version

What does this command check?

The command asks Visual Studio Code to print its installed version. A version number confirms that the editor command is available from PowerShell.

✔️ I see a version number

Visual Studio Code is installed. Its command is ready to open the project workspace.

ⓧ Command not found

The editor command is unavailable in this PowerShell session. The Visual Studio Code User setup for Windows adds it to your path.

  • Open the official Visual Studio Code Windows setup guide.
  • Install the Visual Studio Code User setup for Windows.
  • Close the existing PowerShell window after installation.
  • Start a new PowerShell window from Windows search.
  • Return to the version check above.

Still having trouble with the tool checks?

A newly installed command may remain unavailable in a PowerShell window that was already open. Close that window before starting a fresh one.

If either check still fails, help me troubleshoot my Windows toolchain.

Create and configure the project environment

A virtual environment gives this project its own package installation. The event-api folder keeps that environment beside the source files it supports.

  • Create the event-api folder on your Desktop by running these PowerShell commands:
Set-Location -Path "$env:USERPROFILE\Desktop"
New-Item -Path "." -Name "event-api" -ItemType "Directory"
Set-Location -Path ".\event-api" -PassThru

What do these commands do?

  • The first command moves PowerShell to your Desktop.
  • The second command creates the event-api folder.
  • The final command enters that folder. It also prints the new location for confirmation.

You should see a path ending in Desktop\event-api. Your workspace now has a predictable location.

Folder already exists?

PowerShell may report that the folder already exists if you created it during an earlier attempt. Enter the existing event-api folder before continuing.

If the path still looks wrong, help me locate my event-api folder.

  • Create the .venv environment by running these commands:
python -m venv .venv
.venv\Scripts\Activate.ps1

What do these commands do?

The first command creates an isolated Python environment inside .venv. The second command activates that environment for the current PowerShell session.

Every package you install next goes into this project environment. Other Python projects on your machine stay untouched.

You should see (.venv) at the start of the PowerShell prompt. That prefix confirms the environment is active.

Activation blocked?

PowerShell can block local activation scripts when its execution policy is restrictive. Check your organization's policy before changing any security setting.

For help choosing a safe fix, help me activate this virtual environment.

  • Open the current event-api folder in Visual Studio Code by running:
code .

What does this command do?

The period refers to the current event-api folder. Visual Studio Code opens that folder as the project workspace.

You should see event-api at the top of the Explorer sidebar. The .venv folder should also be visible.

  • Create requirements.txt inside the event-api folder from the Visual Studio Code Explorer sidebar.
  • Add the pinned package list by pasting this content:
fastapi==0.162.0
uvicorn==0.54.0
SQLAlchemy[asyncio]==2.1.6
asyncpg==0.32.0
alembic==1.20.1
pydantic==2.14.1
pytest==9.2.0
pytest-asyncio==1.4.1
httpx==0.28.1

Why pin these packages?

  • FastAPI provides the HTTP application. Uvicorn runs it locally.
  • SQLAlchemy maps Python objects to database rows. asyncpg performs asynchronous PostgreSQL I/O.
  • Alembic manages schema versions. Pydantic validates API data.
  • pytest runs the integration test. pytest-asyncio supports its coroutine. HTTPX sends requests to the app during that test.
  • Save requirements.txt.

The first package installation may take a minute while PowerShell downloads the pinned releases. A stream of package names during this step is normal.

  • Switch back to the PowerShell window with (.venv) in its prompt.
  • Install the project dependencies by running:
python -m pip install -r requirements.txt

What does this command do?

Python runs pip from the active virtual environment. The installer reads every exact version from requirements.txt.

The SQLAlchemy[asyncio] extra includes the dependency required by SQLAlchemy's asyncio extension.

The installation succeeds when PowerShell returns to the (.venv) prompt without reporting a package error. Your isolated environment now contains the complete project toolchain.

Dependency installation failed?

Confirm that (.venv) still appears in the prompt. Confirm that PowerShell is inside the event-api folder containing requirements.txt.

If a package reports an incompatible Python release, repeat the earlier version check. For another installation failure, help me diagnose the pinned dependency installation.

  • Create an app folder inside event-api from the Visual Studio Code Explorer sidebar.
  • Create __init__.py inside the new app folder.
  • Add the package marker by pasting this content:
"""Event API package."""

What does this file do?

The file marks app as the package that holds the Event API code. Its docstring records the package's purpose.

  • Save app/__init__.py.

The Explorer sidebar should now show __init__.py inside app.

Database URLs need a driver plus the local server details. URL.create() accepts the raw password without forcing you to hand-escape special characters.

  • Create config.py inside the app folder.
  • Add the database configuration by pasting this content:
from sqlalchemy import URL


def build_database_url(database: str) -> URL:
    return URL.create(
        drivername="postgresql+asyncpg",
        username="postgres",
        password="your-password-here",
        host="localhost",
        port=5432,
        database=database,
    )


DATABASE_URL = build_database_url("event_api")
TEST_DATABASE_URL = build_database_url("event_api_test")

What does this configuration do?

  • The build_database_url() function creates a SQLAlchemy URL for one named database.
  • The postgresql+asyncpg driver connects SQLAlchemy's asynchronous engine to PostgreSQL.
  • DATABASE_URL targets the development database. TEST_DATABASE_URL targets the isolated test database.

The file now contains the required structure. Its password is still the plain placeholder from the project template.

Protect your local password

Typing a password into source code is sensitive. Keep this project local during the walkthrough.

  • Replace your-password-here with your local PostgreSQL password.
  • Save app/config.py.

The unsaved marker on the editor tab should disappear. Your password is now stored in the local project file.

  • Replace postgres only if your local PostgreSQL username is different.
  • Save app/config.py again if you changed the username.

Configuration file not matching?

Check that both database names use underscores. Confirm that event_api belongs to DATABASE_URL.

Confirm that event_api_test belongs to TEST_DATABASE_URL. If the structure still differs, help me compare my database configuration.

✔️ Awesome, I've got everything

  • Confirm that requirements.txt is saved.
  • Confirm that app/__init__.py is saved.
  • Confirm that app/config.py is saved with your local credentials.

ⓧ I'd like to double check the full code

Compare each file with the reference below. Keep your real local password in place while comparing app/config.py.

fastapi==0.162.0
uvicorn==0.54.0
SQLAlchemy[asyncio]==2.1.6
asyncpg==0.32.0
alembic==1.20.1
pydantic==2.14.1
pytest==9.2.0
pytest-asyncio==1.4.1
httpx==0.28.1
"""Event API package."""
from sqlalchemy import URL


def build_database_url(database: str) -> URL:
    return URL.create(
        drivername="postgresql+asyncpg",
        username="postgres",
        password="your-password-here",
        host="localhost",
        port=5432,
        database=database,
    )


DATABASE_URL = build_database_url("event_api")
TEST_DATABASE_URL = build_database_url("event_api_test")
Create the development and test databases

The local PostgreSQL server starts with system databases. pgAdmin 4 lets you create the two project databases without depending on psql.

The event_api database holds development rows. The event_api_test database gives future integration tests a separate target.

  • Press the Windows key to open Windows search.
  • Type pgAdmin 4 into the search bar.
  • Press Enter to open pgAdmin 4.
  • Connect to your registered local PostgreSQL server in Object Explorer.
  • Enter your local PostgreSQL password if pgAdmin 4 requests it.
  • Select the postgres database in Object Explorer.
  • Open Tools from the top menu.
  • Select Query Tool.
  • Enable Auto commit in the Query Tool.

Why enable Auto commit?

PostgreSQL does not allow CREATE DATABASE inside a transaction block. Auto commit sends each database creation as its own committed statement.

  • Create the development database by running this statement:
CREATE DATABASE event_api;

What does this statement do?

The statement asks PostgreSQL to create an empty database named event_api. The API uses this database for development records.

The Query Tool should report a successful command on its Messages tab. That result confirms the development database exists.

  • Select the existing statement in the Query Tool.
  • Replace it with the test database statement below:
CREATE DATABASE event_api_test;

What does this statement do?

The statement creates an empty database named event_api_test. Later tests can use this target without mixing temporary rows into development data.

The Messages tab should report another successful command. Both PostgreSQL targets now exist.

Before you refresh Object Explorer, which two database names do you expect to appear?

  • Refresh Object Explorer in pgAdmin 4.
  • Expand the Databases node.

You should see event_api plus event_api_test in the database list. This proves both persistent targets are ready.

Database creation denied?

The connected PostgreSQL role must be a superuser or have the CREATEDB privilege. Reconnect with a role that can create databases.

Confirm that Auto commit remains enabled before retrying each statement separately. If creation still fails, help me troubleshoot PostgreSQL database creation.

That is the setup done: your Windows toolchain is ready. Both PostgreSQL targets are visible.

Your project now has an active environment, pinned packages, local credentials, plus two empty databases. Next, you will connect the API to PostgreSQL and prove the connection over HTTP.

Connect the API and Expose the Missing Table

Your Python workspace is ready. PostgreSQL now has two empty databases waiting for the Event API.

A reachable database does not guarantee that application tables exist.

You'll first prove the connection through an async FastAPI service. You'll then test the first event write against the empty database.

In this step, get ready to:
  • Build the asynchronous database connection layer.
  • Define the event mapping and API schemas.
  • Verify the connection and attempt the first event write.
Build the async connection layer

SQLAlchemy manages database sessions for the API. Its async engine uses asyncpg to communicate with PostgreSQL without blocking other requests.

A yielded dependency gives each request its own session. The surrounding transaction commits a successful request or rolls back a failed request.

  • Switch back to the Explorer sidebar in Visual Studio Code.
  • Select the app folder.
  • Create app/database.py with the new-file control.
  • Add the async engine and session dependency by pasting this code:
from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import DATABASE_URL

engine = create_async_engine(DATABASE_URL)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionFactory.begin() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session, scope="function")]

What does this connection layer do?

  • The engine holds the asynchronous connection interface for DATABASE_URL.
  • The SessionFactory creates sessions that keep loaded attributes available after a commit.
  • The get_session() dependency wraps each request in a managed transaction.
  • The SessionDep annotation lets FastAPI inject that session into a route.
  • Save app/database.py.
  • Confirm database.py appears inside the app folder in the Explorer sidebar.

Seeing import warnings in database.py?

  • Check that Visual Studio Code is using the interpreter inside .venv.
  • Confirm the dependency installation completed inside the activated virtual environment.
  • Still stuck? Help me diagnose the imports in my async database module.

Uvicorn needs a small entry point that exposes the FastAPI application object. The entry point keeps the server command separate from the application module.

  • Select the event-api folder at the top of the Explorer sidebar.
  • Create example.py with the new-file control.
  • Expose the application object by pasting this import:
from app.main import app

What does the entry point expose?

The import makes the app object from app/main.py available through example.py. Uvicorn uses that object to serve the API.

  • Save example.py.
  • Confirm example.py appears beside the app folder in the Explorer sidebar.

Is example.py in the wrong folder?

Move example.py directly inside event-api. It must sit beside the app folder so the import path resolves.

Still stuck? Help me place the Uvicorn entry point correctly.

Define the event resource

An ORM mapping connects a Python class to a database table. Each typed attribute on Event describes one column PostgreSQL eventually needs.

  • Select the app folder in the Explorer sidebar.
  • Create app/models.py with the new-file control.
  • Define the base mapping and event table shape by pasting this code:
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    location: Mapped[str] = mapped_column(String(120))
    capacity: Mapped[int]

How does the mapping describe an event?

  • The Base class collects table metadata for the application's mappings.
  • The __tablename__ value connects Event to a table named events.
  • The id attribute becomes the primary key.
  • The remaining attributes describe the event name, location, and capacity.
  • Save app/models.py.
  • Confirm models.py appears inside the app folder.

Seeing type or mapping warnings?

  • Check that every mapped attribute uses the matching Mapped type.
  • Check that __tablename__ contains the exact value events.
  • Still stuck? Help me compare my Event mapping with the required SQLAlchemy structure.

Pydantic schemas define the data accepted and returned by the API. The input schema carries event fields. The response schema adds the generated ID.

  • Create app/schemas.py inside the selected app folder.
  • Define the input and response shapes by pasting this code:
from pydantic import BaseModel, ConfigDict


class EventData(BaseModel):
    name: str
    location: str
    capacity: int


class EventRead(EventData):
    id: int
    model_config = ConfigDict(from_attributes=True)

How do the schemas shape API data?

  • The EventData schema validates the three fields supplied when an event is created.
  • The EventRead schema adds the database-generated id.
  • The from_attributes setting lets the response schema read values from an ORM object.
  • Save app/schemas.py.
  • Confirm schemas.py appears beside models.py inside the app folder.

Are schema imports unresolved?

  • Confirm pydantic==2.14.1 remains listed in requirements.txt.
  • Check that Visual Studio Code still points to the interpreter inside .venv.
  • Still stuck? Help me resolve the Pydantic schema imports.
Run the API and test both routes

The application needs one route that checks the database connection. It also needs one route that attempts a real event write.

The health route runs a simple query through the injected session. The event route builds an Event from validated request data.

  • Create app/main.py inside the app folder.
  • Add the health and event routes by pasting this code:
from fastapi import FastAPI, status
from sqlalchemy import select

from app.database import SessionDep
from app.models import Event
from app.schemas import EventData, EventRead

app = FastAPI(title="Async Event API")


@app.get("/health")
async def health(session: SessionDep):
    await session.execute(select(1))
    return {"database": "connected"}


@app.post(
    "/events",
    response_model=EventRead,
    status_code=status.HTTP_201_CREATED,
)
async def create_event(payload: EventData, session: SessionDep):
    event = Event(**payload.model_dump())
    session.add(event)
    await session.flush()
    await session.refresh(event)
    return event

What do these routes prove?

  • The /health route executes select(1) through the real PostgreSQL connection.
  • The /events route validates request data with EventData.
  • The session.add() call places the new object in the session.
  • The flush() and refresh() calls send the write to PostgreSQL and load the generated values.
  • Save app/main.py.
  • Confirm main.py appears inside the app folder.

Seeing errors in the route definitions?

  • Check that the imported names match SessionDep, Event, EventData, and EventRead exactly.
  • Check that every route decorator begins with the same lowercase app object.
  • Still stuck? Help me find the mismatch in my FastAPI route module.

✔️ Awesome, I've got everything!

Your connection layer, resource mapping, schemas, entry point, and initial routes are saved.

ⓧ I'd like to double check the full code

  • Compare each project file with the cumulative versions below.
"""Event API package."""

What belongs in the package file?

The package docstring identifies app as the Event API package.

from sqlalchemy import URL


def build_database_url(database: str) -> URL:
    return URL.create(
        drivername="postgresql+asyncpg",
        username="postgres",
        password="your-password-here",
        host="localhost",
        port=5432,
        database=database,
    )


DATABASE_URL = build_database_url("event_api")
TEST_DATABASE_URL = build_database_url("event_api_test")

What belongs in the configuration file?

This file retains the development and test database URLs configured in the previous step.

from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import DATABASE_URL

engine = create_async_engine(DATABASE_URL)
SessionFactory = async_sessionmaker(engine, expire_on_commit=False)


async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionFactory.begin() as session:
        yield session


SessionDep = Annotated[AsyncSession, Depends(get_session, scope="function")]

What belongs in the database file?

This file contains the async engine, session factory, yielded session dependency, and dependency annotation.

from fastapi import FastAPI, status
from sqlalchemy import select

from app.database import SessionDep
from app.models import Event
from app.schemas import EventData, EventRead

app = FastAPI(title="Async Event API")


@app.get("/health")
async def health(session: SessionDep):
    await session.execute(select(1))
    return {"database": "connected"}


@app.post(
    "/events",
    response_model=EventRead,
    status_code=status.HTTP_201_CREATED,
)
async def create_event(payload: EventData, session: SessionDep):
    event = Event(**payload.model_dump())
    session.add(event)
    await session.flush()
    await session.refresh(event)
    return event

What belongs in the application file?

This file contains the FastAPI application plus the initial health and create routes.

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    location: Mapped[str] = mapped_column(String(120))
    capacity: Mapped[int]

What belongs in the models file?

This file contains the declarative base and typed Event mapping.

from pydantic import BaseModel, ConfigDict


class EventData(BaseModel):
    name: str
    location: str
    capacity: int


class EventRead(EventData):
    id: int
    model_config = ConfigDict(from_attributes=True)

What belongs in the schemas file?

This file contains the event input schema and ORM-compatible response schema.

from app.main import app

What belongs in the entry point?

This file exposes the FastAPI application object to Uvicorn.

fastapi==0.162.0
uvicorn==0.54.0
SQLAlchemy[asyncio]==2.1.6
asyncpg==0.32.0
alembic==1.20.1
pydantic==2.14.1
pytest==9.2.0
pytest-asyncio==1.4.1
httpx==0.28.1

What belongs in the requirements file?

This file retains the pinned dependencies installed in the previous step.

Before you start the server, do you expect the application imports to complete successfully?

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

What does this command start?

Uvicorn imports app from example.py. It then serves the FastAPI application on its local default address.

The terminal stays occupied while the server runs. Seeing startup information without an import traceback confirms that every module loads successfully.

Does the server stop during startup?

  • Check the final traceback line for the file and import that stopped startup.
  • Confirm the terminal prompt shows the activated .venv environment.
  • Still stuck? Help me troubleshoot why Uvicorn cannot start example:app.

The first request checks only whether the API can reach PostgreSQL. Before you send it, what result do you expect from a simple database query?

  • Open http://127.0.0.1:8000/docs in your browser.
  • Expand the GET /health route in Swagger UI.
  • Send the health request with the route's interactive controls.

You'll see a successful response containing {"database": "connected"}. This proves the engine, credentials, async driver, and development database can work together.

Good work. Your API has now completed a real query against PostgreSQL.

Does the health request fail?

  • Confirm the local PostgreSQL server is running.
  • Check the username and password stored in app/config.py.
  • Still stuck? Help me diagnose my failed async PostgreSQL health check.

The next request asks PostgreSQL to store an Event. Before you send it, do you think the write will succeed or fail?

  • Expand the POST /events route in Swagger UI.
  • Activate the route's interactive request controls.
  • Set name to Async Summit in the request body.
  • Set location to Berlin in the request body.
  • Set capacity to 200 in the request body.
  • Send the event request.

You'll see an HTTP 500 response. The Uvicorn terminal shows a database error that mentions the missing events table.

The missing-table wall is expected

The health result already proved that the engine can reach PostgreSQL. The event write reaches the empty database next, where the absent table stops it.

Keep this failure visible. It proves that application models do not create database tables by themselves.

You now have a running async API with a verified PostgreSQL connection and a visible schema gap. Next, you'll replace that missing schema with a version-controlled migration.

Migrate the Event Schema

Your API already reaches PostgreSQL. The failed event write proved that a working connection still needs a matching database schema.

This step uses Alembic to turn the existing Event mapping into a version-controlled migration. You will apply the same revision to the development database and the test database.

In this step, get ready to:
  • Initialize the async migration environment.
  • Create a reversible revision for the events table.
  • Apply the revision to both databases.
Bootstrap the migration environment

A migration environment stores schema changes as ordered revisions. The async template prepares Alembic to connect through the same async database driver as your API.

  • Stop the running Uvicorn process by pressing Ctrl+C in the activated PowerShell terminal.
  • Create the async Alembic environment inside the existing `event-api` folder by running this command:
alembic init -t async alembic

What does this command create?

  • The async template creates the `alembic` directory with an async-ready `env.py` file.
  • The command also creates `alembic.ini` beside your existing application files.
  • The `versions` directory holds each ordered schema revision.
  • Confirm the VS Code Explorer sidebar now shows `alembic.ini` and the `alembic` folder.
  • Leave the generated `README` file unchanged.
  • Leave the generated `script.py.mako` file unchanged.

Alembic environment missing?

  • Check that the terminal is inside the `event-api` folder before running the initialization command.
  • Confirm the activated terminal prompt still includes `.venv`.
  • Still stuck? Help me troubleshoot Alembic initialization in my event-api project.

The two configuration files distinguish your development database from your test database. Each file sets a short selector that `env.py` uses to choose the correct URL.

  • Replace the generated contents of `alembic.ini` with this configuration:
[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .
path_separator = os
database.name = main

How does the main configuration work?

  • `script_location` points Alembic to the generated migration directory.
  • `prepend_sys_path` makes the application package available to the migration environment.
  • `database.name = main` selects `DATABASE_URL` for the development database.
  • Save `alembic.ini`.
  • Confirm its final line reads `database.name = main`.

Main configuration not saving?

  • Check that you edited the `alembic.ini` file beside `example.py`.
  • Confirm the section header is `[alembic]`.
  • Still stuck? Help me check my main Alembic configuration.

✔️ Awesome, I've got everything!

Your main Alembic configuration now selects the development database.

ⓧ I'd like to double check the full code

Compare your saved `alembic.ini` file with this complete reference.

[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .
path_separator = os
database.name = main

What should match?

The file contains one `[alembic]` section. Its final value selects the main database.

  • Create `mycustom.ini` beside `alembic.ini` in the `event-api` folder with this configuration:
[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .
path_separator = os
database.name = test

How does the test configuration differ?

This file points to the same revision history. Its `database.name = test` selector makes the migration environment use `TEST_DATABASE_URL`.

  • Save `mycustom.ini`.
  • Confirm the VS Code Explorer sidebar lists `mycustom.ini` beside `alembic.ini`.

Test configuration not appearing?

  • Check that the filename is exactly `mycustom.ini`.
  • Confirm the file sits directly inside `event-api`.
  • Still stuck? Help me create the alternate Alembic configuration.

✔️ Awesome, I've got everything!

Your alternate configuration now selects the test database.

ⓧ I'd like to double check the full code

Compare your saved `mycustom.ini` file with this complete reference.

[alembic]
script_location = %(here)s/alembic
prepend_sys_path = .
path_separator = os
database.name = test

What should match?

The structure matches the main configuration. The final value selects the test database.

Define the async migration and revision

The generated environment needs access to both project URLs and the existing SQLAlchemy metadata. It can then choose a database before running each revision.

  • Replace the top of `alembic/env.py` with this configuration and metadata setup:
import asyncio

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from app.config import DATABASE_URL, TEST_DATABASE_URL
from app.models import Base

config = context.config

database_name = config.get_main_option("database.name")
selected_url = TEST_DATABASE_URL if database_name == "test" else DATABASE_URL
config.set_main_option(
    "sqlalchemy.url",
    selected_url.render_as_string(hide_password=False).replace("%", "%%"),
)

target_metadata = Base.metadata

What does this setup do?

  • `database_name` reads the selector from the active configuration file.
  • `selected_url` chooses between `DATABASE_URL` and `TEST_DATABASE_URL`.
  • `target_metadata` gives Alembic the table definitions registered on `Base`.
  • Save `alembic/env.py`.
  • Confirm the current file ends with `target_metadata = Base.metadata`.

Imports showing as unresolved?

  • Confirm `alembic/env.py` imports both URLs from `app.config`.
  • Check that `Base` is imported from `app.models`.
  • Still stuck? Help me fix the imports in my async Alembic environment.
  • Add the offline migration function and synchronous migration bridge below `target_metadata = Base.metadata`:
def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()

Why are there two migration paths?

`run_migrations_offline()` configures migrations from a URL without opening an application session. `do_run_migrations()` receives a synchronous connection bridge from the async engine.

  • Save `alembic/env.py`.
  • Confirm the new section defines both `run_migrations_offline()` and `do_run_migrations()`.

Migration functions indented incorrectly?

  • Align both function definitions with the left edge of the file.
  • Keep each `context.run_migrations()` call inside its matching transaction block.
  • Still stuck? Help me correct the migration function structure.
  • Add the async engine runner and mode selection below `do_run_migrations()`:
async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


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

How do async migrations run?

  • `async_engine_from_config()` creates an async engine from the selected URL.
  • `connection.run_sync()` passes the connection into Alembic's synchronous migration context.
  • The final condition selects the offline path or the online path.
  • Save `alembic/env.py`.
  • Confirm the final line calls `run_migrations_online()`.

Async runner incomplete?

  • Check that `await connection.run_sync(do_run_migrations)` remains inside the connection block.
  • Confirm `await connectable.dispose()` aligns with the connection block.
  • Still stuck? Help me check the async Alembic runner.

✔️ Awesome, I've got everything!

Your migration environment now selects a URL and runs revisions through an async engine.

ⓧ I'd like to double check the full code

Compare your complete `alembic/env.py` file with this reference.

import asyncio

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from app.config import DATABASE_URL, TEST_DATABASE_URL
from app.models import Base

config = context.config

database_name = config.get_main_option("database.name")
selected_url = TEST_DATABASE_URL if database_name == "test" else DATABASE_URL
config.set_main_option(
    "sqlalchemy.url",
    selected_url.render_as_string(hide_password=False).replace("%", "%%"),
)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


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

What should match?

The complete file selects the database URL before defining the offline and online migration paths.

A revision records one reversible schema change. This first revision creates columns that match the existing `Event` mapping exactly.

  • Create `0001_create_events_table.py` inside `alembic/versions` with this revision:
from alembic import op
import sqlalchemy as sa

revision = "0001"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "events",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(length=120), nullable=False),
        sa.Column("location", sa.String(length=120), nullable=False),
        sa.Column("capacity", sa.Integer(), nullable=False),
    )


def downgrade() -> None:
    op.drop_table("events")

What does revision 0001 change?

  • `revision = "0001"` gives this schema change a stable identifier.
  • `upgrade()` creates the `events` table with the four mapped columns.
  • `downgrade()` reverses the revision by dropping the table.
  • Save `alembic/versions/0001_create_events_table.py`.
  • Confirm its columns are `id`, `name`, `location`, and `capacity`.

Revision file not detected?

  • Check that the file sits inside `alembic/versions`.
  • Confirm the filename is exactly `0001_create_events_table.py`.
  • Still stuck? Help me verify my first Alembic revision.

✔️ Awesome, I've got everything!

Revision `0001` now describes a reversible `events` table migration.

ⓧ I'd like to double check the full code

Compare your complete revision file with this reference.

from alembic import op
import sqlalchemy as sa

revision = "0001"
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "events",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(length=120), nullable=False),
        sa.Column("location", sa.String(length=120), nullable=False),
        sa.Column("capacity", sa.Integer(), nullable=False),
    )


def downgrade() -> None:
    op.drop_table("events")

What should match?

The upgrade creates four columns. The downgrade removes the same table.

Apply the schema to both databases

Both databases need the same revision history. Applying the default configuration migrates `event_api` first.

Before you run the migration, do you expect the earlier missing-table failure to remain after revision `0001` reaches the database?

  • Apply revision `0001` to `event_api` by running this command:
alembic upgrade head

What does this command do?

Alembic reads `alembic.ini`, selects the main database URL, and applies every pending upgrade through revision `0001`.

The command completes without a migration error. The development database now has the application table and Alembic's revision table.

Main migration failed?

  • Confirm the password in `app/config.py` still matches your local PostgreSQL server.
  • Check that `event_api` remains available in pgAdmin 4.
  • Still stuck? Help me troubleshoot the main Alembic migration.

The alternate configuration runs the same revision history against `event_api_test`. This keeps integration tests separate from development data.

  • Apply revision `0001` to `event_api_test` by running this command:
alembic --config mycustom.ini upgrade head

How is the test database selected?

The `--config mycustom.ini` option loads the alternate selector. The migration environment responds by choosing `TEST_DATABASE_URL`.

The second migration completes without an error. Both databases now share revision `0001`.

Test migration failed?

  • Confirm `mycustom.ini` contains `database.name = test`.
  • Check that `event_api_test` remains available in pgAdmin 4.
  • Still stuck? Help me troubleshoot the test database migration.

The database tree provides a direct view of the schema Alembic created. You can now verify the same two tables in each database without using `psql`.

  • Return to pgAdmin 4 from earlier.
  • Refresh `event_api` in Object Explorer.
  • Expand `Schemas` under `event_api`.
  • Expand `public`.
  • Expand `Tables`.

You should see `events` and `alembic_version` under the development database. This proves revision `0001` reached `event_api`.

  • Refresh `event_api_test` in Object Explorer.
  • Expand `Schemas` under `event_api_test`.
  • Expand `public`.
  • Expand `Tables`.

You should see `events` and `alembic_version` under the test database. Both database targets now have the same schema version.

Before you retry the failed request, do you think `POST /events` can now store Async Summit successfully?

  • Restart the API from the activated PowerShell terminal by running this command:
uvicorn example:app

What is running now?

Uvicorn loads the existing FastAPI application from `example.py`. The application connects to the newly migrated development database.

  • Return to `http://127.0.0.1:8000/docs` in your browser.
  • Expand the `POST /events` operation.
  • Enable editing for the request body.
  • Set the event name to `Async Summit`.
  • Set the event location to `Berlin`.
  • Set the event capacity to `200`.
  • Send the request.

You should receive status `201` with the new event in the response body. That is the hard part resolved: the same write that previously failed now reaches a migrated table.

Still receiving a 500 response?

  • Confirm `events` appears under `event_api` in pgAdmin 4.
  • Check that the API uses `DATABASE_URL` from `app/config.py`.
  • Still stuck? Help me diagnose the failed event write after migration.

Your Event API now has a version-controlled schema in both PostgreSQL databases. Next up, you will complete the event lifecycle with read, update, and delete operations.

Complete the Event CRUD API

Your migration fixed the missing-table failure. Your FastAPI service can now store events in PostgreSQL through HTTP.

A create-only endpoint cannot manage a complete resource lifecycle. In this step, you will add the remaining CRUD operations before verifying every route through Swagger UI.

In this step, get ready to:
  • Add list and single-event read operations.
  • Add update and delete operations with missing-event handling.
  • Verify the complete event lifecycle in Swagger UI and pgAdmin 4.
Add the read endpoints

Read endpoints let clients inspect the rows already stored in the database. The collection route returns events in ID order. The detail route looks up one specific event.

  • Return to app/main.py in Visual Studio Code.
  • Find this import line:
from fastapi import FastAPI, status

What does this import provide?

This existing import provides the application class and the named response status used by create_event().

  • Replace the import with this updated line:
from fastapi import FastAPI, HTTPException, status

Why add HTTPException?

The HTTPException import lets a route stop with a specific HTTP status. You will use it to return 404 when an event ID has no matching row.

  • Add the collection route below create_event() by pasting this code:
@app.get("/events", response_model=list[EventRead])
async def list_events(session: SessionDep):
    result = await session.scalars(select(Event).order_by(Event.id))
    return result.all()

What does the list route do?

  • The response model tells FastAPI to serialize a list of event records.
  • The SQLAlchemy query selects every Event row.
  • The order_by(Event.id) clause gives the response a predictable order.
  • The result.all() call returns every selected event.
  • Save app/main.py.
  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the current server process.
  • Restart the API with the updated route by running this command:
uvicorn example:app

What does this command start?

The Uvicorn server imports app from example.py. It keeps this terminal occupied while the API handles requests.

  • Refresh Swagger UI at http://127.0.0.1:8000/docs.
  • Expand GET /events.
  • Click Try it out.
  • Click Execute.

You will see a 200 response containing the existing Async Summit event. The response is a JSON list because this route can return multiple events.

  • Record the returned id here: your event ID.

Don't see the event list?

  • Confirm the terminal still shows the running Uvicorn process.
  • Check that list_events() sits below create_event() without extra indentation.
  • Check that the route decorator uses /events.
  • Still stuck? Help me troubleshoot my GET /events route.
  • Add the single-event route below list_events() by pasting this code:
@app.get("/events/{event_id}", response_model=EventRead)
async def get_event(event_id: int, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    return event

How does the detail route work?

  • The event_id path parameter identifies the requested row.
  • The session.get() call looks up an Event by its primary key.
  • A missing row produces an HTTPException with status 404.
  • A matching row is serialized through EventRead.
  • Save app/main.py.
  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the current server process.
  • Restart the API with the detail route by running this command:
uvicorn example:app

Why restart the server?

Restarting Uvicorn imports the saved version of app/main.py. Swagger UI can then generate documentation for the new route.

  • Refresh Swagger UI.
  • Expand GET /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Click Execute.

You will see a 200 response containing the Async Summit event as one JSON object. This confirms the route can retrieve a row by its primary key.

Getting a 404 for your event?

  • Run GET /events again to confirm the event still exists.
  • Copy the id value from the list response without quotation marks.
  • Check that get_event() passes Event and event_id to session.get().
  • Still stuck? Help me troubleshoot my single-event route.
Add the update and delete endpoints

An update operation changes the stored fields for an existing event. A delete operation removes the matching row. Both routes protect clients from acting on an ID that does not exist.

  • Add the update route below get_event() by pasting this code:
@app.put("/events/{event_id}", response_model=EventRead)
async def update_event(event_id: int, payload: EventData, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    event.name = payload.name
    event.location = payload.location
    event.capacity = payload.capacity
    await session.flush()
    await session.refresh(event)
    return event

What does the update route do?

  • The route retrieves the event before changing any fields.
  • The three assignments copy the validated request values onto the mapped row.
  • The flush() call sends the pending update through the current database transaction.
  • The refresh() call reloads the updated row before FastAPI builds the response.
  • Save app/main.py.
  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the current server process.
  • Restart the API with the update route by running this command:
uvicorn example:app

What changes after this restart?

The restarted application includes update_event(). Swagger UI will now show a PUT /events/{event_id} operation.

  • Refresh Swagger UI.
  • Expand PUT /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Set the request body's name value to Async Summit.
  • Set the request body's location value to London.
  • Set the request body's capacity value to 250.
  • Click Execute.

You will see a 200 response with London and 250 in the returned event. The row has now changed inside PostgreSQL.

Update returning an error?

  • Confirm the request body contains values for name. Check location next. Check capacity last.
  • Confirm capacity contains a number without quotation marks.
  • Check that every assignment inside update_event() is indented beneath the function.
  • Still stuck? Help me troubleshoot my event update route.
  • Add the delete route below update_event() by pasting this code:
@app.delete("/events/{event_id}")
async def delete_event(event_id: int, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    await session.delete(event)
    await session.flush()
    return {"deleted": event_id}

What does the delete route do?

  • The route retrieves the event before attempting deletion.
  • A missing event returns the same 404 behavior as the read and update routes.
  • The session.delete() call marks the mapped row for deletion.
  • The response returns the ID that was deleted.
  • Save app/main.py.
  • Switch back to the terminal running Uvicorn.
  • Press Ctrl+C to stop the current server process.
  • Restart the API with the delete route by running this command:
uvicorn example:app

What does this restart confirm?

Uvicorn imports the completed route module. Swagger UI can now document create, list, read, update, and delete operations.

  • Refresh Swagger UI.
  • Confirm that DELETE /events/{event_id} now appears with the other event routes.

Your API surface now contains every operation needed for the event lifecycle. The next substep proves that the routes change the same database rows you can inspect directly.

Delete route missing from Swagger UI?

  • Confirm delete_event() sits below update_event().
  • Check that the decorator begins with @app.delete.
  • Confirm the terminal restarted after you saved app/main.py.
  • Still stuck? Help me find my missing delete route.

✔️ Awesome, I've got everything!

Great. Double check that you saved app/main.py before moving to the lifecycle test.

ⓧ I'd like to double check the full code

Compare your completed app/main.py with this file:

from fastapi import FastAPI, HTTPException, status
from sqlalchemy import select

from app.database import SessionDep
from app.models import Event
from app.schemas import EventData, EventRead

app = FastAPI(title="Async Event API")


@app.get("/health")
async def health(session: SessionDep):
    await session.execute(select(1))
    return {"database": "connected"}


@app.post(
    "/events",
    response_model=EventRead,
    status_code=status.HTTP_201_CREATED,
)
async def create_event(payload: EventData, session: SessionDep):
    event = Event(**payload.model_dump())
    session.add(event)
    await session.flush()
    await session.refresh(event)
    return event


@app.get("/events", response_model=list[EventRead])
async def list_events(session: SessionDep):
    result = await session.scalars(select(Event).order_by(Event.id))
    return result.all()


@app.get("/events/{event_id}", response_model=EventRead)
async def get_event(event_id: int, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")
    return event


@app.put("/events/{event_id}", response_model=EventRead)
async def update_event(event_id: int, payload: EventData, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    event.name = payload.name
    event.location = payload.location
    event.capacity = payload.capacity
    await session.flush()
    await session.refresh(event)
    return event


@app.delete("/events/{event_id}")
async def delete_event(event_id: int, session: SessionDep):
    event = await session.get(Event, event_id)
    if event is None:
        raise HTTPException(status_code=404, detail="Event not found")

    await session.delete(event)
    await session.flush()
    return {"deleted": event_id}

What should match?

Check each decorator first. Then compare the indentation inside every function. The completed file should expose six routes in Swagger UI.

Exercise the full event lifecycle

A lifecycle check follows one row from creation through deletion. You will also inspect the updated row in pgAdmin 4 to prove the API and database show the same state.

Before you start this final check, what sequence of database changes do you expect as one event moves through every route?

  • Return to Swagger UI.
  • Expand POST /events.
  • Click Try it out.
  • Set the request body's name value to Lifecycle Demo.
  • Set the request body's location value to Online.
  • Set the request body's capacity value to 50.
  • Click Execute.
  • Replace the saved ID with the returned id: your event ID.

You will see a 201 response containing Lifecycle Demo. PostgreSQL now holds the new event row.

  • Expand GET /events.
  • Click Try it out.
  • Click Execute.
  • Expand GET /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Click Execute.

What have you proved so far?

The collection response includes Lifecycle Demo. The detail response returns the same row by ID. Both read routes are querying the data created through the API.

  • Expand PUT /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Set the request body's name value to Lifecycle Demo.
  • Set the request body's location value to London.
  • Set the request body's capacity value to 75.
  • Click Execute.

You will see a 200 response containing the updated location and capacity. The API has saved both changes.

  • Switch back to the connected pgAdmin 4 window.
  • Expand event_api in Object Explorer.
  • Expand Schemas.
  • Expand public.
  • Expand Tables.
  • Right-click events.
  • Select View/Edit Data.

You will see the Lifecycle Demo row with London and 75. This is the same updated row returned by the API.

Before the final fetch, do you expect the deleted ID to return an event?

  • Return to Swagger UI.
  • Expand DELETE /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Click Execute.

You will see a 200 response containing the deleted event ID. The row has been removed.

  • Expand GET /events/{event_id}.
  • Click Try it out.
  • Enter your event ID in the event_id field.
  • Click Execute.

You will see a 404 response with Event not found. That completes the create, list, read, update, inspect, delete, and missing-event lifecycle.

Lifecycle result doesn't match?

  • Run GET /events to confirm which event IDs still exist.
  • Use the same saved event ID for the read, update, and delete requests.
  • Refresh the pgAdmin 4 data grid after each API change if the displayed values are stale.
  • Still stuck? Help me trace my event lifecycle.

That is the complete resource lifecycle working against PostgreSQL. Next, you will wrap real HTTP requests in a database transaction and prove that test rows disappear after rollback.

Prove Test Transaction Rollback

Your Event API now handles the full event lifecycle against PostgreSQL. The next risk sits in test data that survives after a test finishes.

Permanent test rows make later runs depend on earlier runs. A surrounding transaction gives each run a clean boundary that can be rolled back.

The pytest runner drives the test. HTTPX calls the FastAPI app in-process. SQLAlchemy keeps every database operation inside the outer transaction.

In this step, get ready to:
  • Create an async integration test with an external transaction.
  • Exercise the complete Event API lifecycle over HTTP.
  • Prove that rollback removes the final probe row.
Create the rollback test boundary

The test records the current event count before it writes anything. That baseline gives the final rollback check a trustworthy comparison point.

  • In the Visual Studio Code workspace from earlier, use the file sidebar to create a tests folder inside event-api.
  • Inside tests, create test_events.py with the file sidebar.

You should see tests/test_events.py in the file sidebar. The empty file is ready for the first runnable test slice.

  • Add the imports plus the baseline count by pasting this code into tests/test_events.py:
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import TEST_DATABASE_URL
from app.database import get_session
from app.main import app
from app.models import Event


@pytest.mark.asyncio
async def test_event_crud_and_transaction_rollback():
    test_engine = create_async_engine(TEST_DATABASE_URL)
    TestSessionFactory = async_sessionmaker(test_engine, expire_on_commit=False)

    async with TestSessionFactory() as baseline_session:
        baseline_count = await baseline_session.scalar(
            select(func.count()).select_from(Event)
        )

What does this test slice do?

  • The test engine targets TEST_DATABASE_URL, which points to event_api_test.
  • The session factory creates database sessions that keep ORM values available after commits.
  • The baseline query records the number of event rows that exist before the test begins.
  • The async marker lets pytest run the coroutine in its event loop.
  • Save tests/test_events.py.

Before you run the first check, how many tests do you expect pytest to collect?

  • Run the first test slice from the terminal in event-api by running this command:
python -m pytest

What should you see?

Pytest should collect one test. The test should pass after reading the current row count from event_api_test.

Your first database test is green. The baseline count is now coming from the real test database.

Test collection or connection failing?

  • Confirm that test_events.py sits inside the tests folder.
  • Check that TEST_DATABASE_URL still points to the migrated event_api_test database.
  • Still stuck? Help me diagnose my async pytest database test.

The next slice opens one database connection for the whole test. Its outer transaction remains in control even when the API session completes its own work.

  • Extend test_event_crud_and_transaction_rollback() directly below the baseline count by pasting this code:
    async with test_engine.connect() as connection:
        outer_transaction = await connection.begin()
        test_session = AsyncSession(
            bind=connection,
            expire_on_commit=False,
            join_transaction_mode="rollback_only",
        )

        async def override_get_session():
            yield test_session

        app.dependency_overrides[get_session] = override_get_session

        try:
            transport = ASGITransport(app=app)
        finally:
            app.dependency_overrides = {}
            await test_session.close()
            await outer_transaction.rollback()

    async with TestSessionFactory() as verification_session:
        after_count = await verification_session.scalar(
            select(func.count()).select_from(Event)
        )

    await test_engine.dispose()
    assert after_count == baseline_count

How does the rollback boundary work?

  • The connection-level outer_transaction becomes the boundary around every database operation in the test.
  • The join_transaction_mode="rollback_only" setting prevents session commits from taking control away from that boundary.
  • The dependency override sends API requests through test_session.
  • The finally block clears the override before rolling back the outer transaction.
  • The fresh verification session checks the database after the rollback completes.
  • Save tests/test_events.py.

Before you rerun the test, do you expect the final row count to differ from the baseline?

  • Check the transaction boundary by running this command:
python -m pytest

What did this run prove?

You should see one passing test. The fresh verification session reads the same count that the baseline session recorded.

The outer rollback now controls the test database boundary. The next substep puts real HTTP writes inside it.

Rollback boundary test failing?

  • Check that join_transaction_mode="rollback_only" matches the test code exactly.
  • Confirm that await outer_transaction.rollback() remains inside the finally block.
  • Need another pair of eyes? Help me troubleshoot my external transaction rollback.
Exercise CRUD inside the transaction

The dependency override makes HTTP requests use the transaction-bound session. This tests the real route behavior while keeping every database write disposable.

  • In tests/test_events.py, find transport = ASGITransport(app=app).
  • Paste this HTTP create, read, and update sequence directly below that line:
            async with AsyncClient(
                transport=transport,
                base_url="http://testserver",
            ) as client:
                created = await client.post(
                    "/events",
                    json={
                        "name": "Test Conference",
                        "location": "Online",
                        "capacity": 50,
                    },
                )
                assert created.status_code == 201
                event_id = created.json()["id"]

                fetched = await client.get(f"/events/{event_id}")
                assert fetched.status_code == 200
                assert fetched.json()["name"] == "Test Conference"

                updated = await client.put(
                    f"/events/{event_id}",
                    json={
                        "name": "Test Conference",
                        "location": "London",
                        "capacity": 75,
                    },
                )
                assert updated.status_code == 200
                assert updated.json()["location"] == "London"

What does this HTTP sequence prove?

  • The in-process transport sends requests directly to the FastAPI application.
  • The create request proves the API can insert an event through the overridden session.
  • The read request confirms that the inserted event remains visible inside the transaction.
  • The update request proves that later operations can modify the same transactional row.
  • Save tests/test_events.py.

Before you run this slice, do you expect the new event to survive after the test completes?

  • Exercise the first three CRUD operations by running this command:
python -m pytest

What should happen now?

You should still see one passing test. The API creates and updates a real row before the outer transaction removes it.

The HTTP layer now participates in the rollback boundary. Your test covers real application wiring without retaining its event.

HTTP assertions failing?

  • Check that the async client uses base_url="http://testserver".
  • Confirm that the create request sends name, location, and capacity values.
  • Need help with a failed response? Help me trace the failing HTTP assertion.

The final probe stays inside the transaction when the test reaches its count assertion. That creates visible evidence that the transaction contains one extra row before rollback.

  • Find assert updated.json()["location"] == "London" in tests/test_events.py.
  • Paste this delete, missing-event, and rollback-probe sequence directly below that assertion:
                deleted = await client.delete(f"/events/{event_id}")
                assert deleted.status_code == 200

                missing = await client.get(f"/events/{event_id}")
                assert missing.status_code == 404

                probe = await client.post(
                    "/events",
                    json={
                        "name": "Rollback Probe",
                        "location": "Temporary",
                        "capacity": 1,
                    },
                )
                assert probe.status_code == 201

            inside_count = await test_session.scalar(
                select(func.count()).select_from(Event)
            )
            assert inside_count == baseline_count + 1

Why leave one probe row behind?

  • The delete request completes the successful event lifecycle.
  • The next read confirms that a deleted event produces the API's documented 404 behavior.
  • The final request creates Rollback Probe without deleting it through the API.
  • The internal count proves that the probe exists before the outer transaction rolls back.
  • The existing post-rollback assertion proves that a fresh session cannot see the probe afterward.
  • Save the completed tests/test_events.py file.

✔️ I have the full test file

Your test now covers Event CRUD plus transaction rollback. Keep tests/test_events.py saved before the final run.

ⓧ I'd like to double check the full code

Compare tests/test_events.py with this complete version:

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.config import TEST_DATABASE_URL
from app.database import get_session
from app.main import app
from app.models import Event


@pytest.mark.asyncio
async def test_event_crud_and_transaction_rollback():
    test_engine = create_async_engine(TEST_DATABASE_URL)
    TestSessionFactory = async_sessionmaker(test_engine, expire_on_commit=False)

    async with TestSessionFactory() as baseline_session:
        baseline_count = await baseline_session.scalar(
            select(func.count()).select_from(Event)
        )

    async with test_engine.connect() as connection:
        outer_transaction = await connection.begin()
        test_session = AsyncSession(
            bind=connection,
            expire_on_commit=False,
            join_transaction_mode="rollback_only",
        )

        async def override_get_session():
            yield test_session

        app.dependency_overrides[get_session] = override_get_session

        try:
            transport = ASGITransport(app=app)
            async with AsyncClient(
                transport=transport,
                base_url="http://testserver",
            ) as client:
                created = await client.post(
                    "/events",
                    json={
                        "name": "Test Conference",
                        "location": "Online",
                        "capacity": 50,
                    },
                )
                assert created.status_code == 201
                event_id = created.json()["id"]

                fetched = await client.get(f"/events/{event_id}")
                assert fetched.status_code == 200
                assert fetched.json()["name"] == "Test Conference"

                updated = await client.put(
                    f"/events/{event_id}",
                    json={
                        "name": "Test Conference",
                        "location": "London",
                        "capacity": 75,
                    },
                )
                assert updated.status_code == 200
                assert updated.json()["location"] == "London"

                deleted = await client.delete(f"/events/{event_id}")
                assert deleted.status_code == 200

                missing = await client.get(f"/events/{event_id}")
                assert missing.status_code == 404

                probe = await client.post(
                    "/events",
                    json={
                        "name": "Rollback Probe",
                        "location": "Temporary",
                        "capacity": 1,
                    },
                )
                assert probe.status_code == 201

            inside_count = await test_session.scalar(
                select(func.count()).select_from(Event)
            )
            assert inside_count == baseline_count + 1
        finally:
            app.dependency_overrides = {}
            await test_session.close()
            await outer_transaction.rollback()

    async with TestSessionFactory() as verification_session:
        after_count = await verification_session.scalar(
            select(func.count()).select_from(Event)
        )

    await test_engine.dispose()
    assert after_count == baseline_count

This complete file sends every HTTP operation through one transaction-bound session. Its final assertion compares a fresh post-rollback count with the original baseline.

Confirm the probe row disappeared

The final run tests both sides of the isolation claim. The probe must exist inside the transaction before disappearing from the database afterward.

Before you run the completed test, do you expect the internal count assertion and the post-rollback count assertion to agree with each other?

  • Run the completed integration test by running this command:
python -m pytest

What proves the test passed?

Pytest should report one passing test. That result means every CRUD response matched its assertion.

It also means the row count increased inside the transaction. The final count returned to the original baseline after rollback.

Completed test failing?

  • Confirm that the events table exists in event_api_test.
  • Check that app.dependency_overrides = {} runs before the test session closes.
  • Check that the outer transaction rolls back before verification_session reads the final count.
  • Still seeing a failure? Help me debug the completed rollback test.

The passing test proves the code-level assertions. A direct database check makes the rollback visible outside the test process.

Before you load the test table, do you expect its name column to contain Rollback Probe?

  • Return to the pgAdmin 4 window from earlier.
  • In Object Explorer, expand event_api_test.
  • Under its public schema, right-click events.
  • Choose View/Edit Data to load the table rows.
  • Check the name column for Rollback Probe.

The Data Grid contains no Rollback Probe row. Any events that existed before the test remain unchanged.

You have proved full-stack test isolation. Real HTTP requests reached PostgreSQL inside a transaction that left the test database at its original row count.

Secret mission

Protect Events That Have Bookings

Bookings make unconditional event deletion unsafe. Extend the database with a bookings table and return a 409 response whenever someone tries to delete an event that still has bookings.

Clean Up Your Resources

Clean Up Your Resources

Everything you created stays on your Windows machine. Your local PostgreSQL databases cost $0 to keep. Choose whether to keep the API available, pause its running server, or delete the databases and project files.

Resources you used:

  • The event-api project folder containing your application code, migrations, tests, and .venv environment.
  • The local event_api database containing events, bookings, and alembic_version tables.
  • The local event_api_test database containing the migrated tables used by your rollback-isolated test.
  • The migration history managed through Alembic inside both databases.
  • The database objects inspected through pgAdmin 4.

Keep everything running

No action needed. Choose this if you plan to keep extending the Event API or experimenting with its booking guard.

  • Your event_api database keeps the event and booking rows you created.
  • Your event_api_test database stays migrated and ready for more isolated integration tests.
  • Your event-api folder keeps the source code, migration revisions, test, and virtual environment.
  • Your Uvicorn server can remain available while you continue working with the API.

Pause - I'll come back to this later

Stop the running API to free the port and terminal. Your project files and both databases remain intact for your return.

  • Return to the PowerShell terminal where Uvicorn is running.
  • Press Ctrl+C to stop the server.

You should see the PowerShell prompt return. The event_api and event_api_test databases remain available in Object Explorer.

Delete - I don't want to use this again

Remove both databases before deleting the project folder. This clears the persisted rows, migration history, source code, and virtual environment from your machine.

  • Return to the PowerShell terminal where Uvicorn is running.
  • Press Ctrl+C to stop the server.
  • Close any pgAdmin 4 Query Tool or View/Edit Data tabs connected to either database.
  • In Object Explorer, use the context menu for event_api to remove the database.
  • Confirm the removal when pgAdmin 4 asks.
  • Use the context menu for event_api_test to remove the test database.
  • Confirm the second removal when pgAdmin 4 asks.
  • Refresh Object Explorer.

You should no longer see event_api or event_api_test in Object Explorer. Removing each database also removes its events, bookings, and alembic_version tables.

  • Close any Visual Studio Code window using the event-api folder.
  • Locate the event-api folder with Windows file management.
  • Use the folder's context menu to delete event-api and its contents.

You should no longer see the event-api folder. Its .venv environment disappears with the rest of the project files.

Nice Work!

Nice Work!

Fantastic work! You've built a FastAPI service that uses async database I/O with PostgreSQL.

You've learned how to:

  • Build an async Event API with complete create, list, read, update, and delete behavior. Verify persisted event rows through Swagger UI and pgAdmin 4.
  • Use reversible Alembic migrations to create matching schemas in event_api and event_api_test. Resolve a real missing-table failure without creating the table manually.
  • Run an async HTTP integration test inside an external transaction. Prove rollback isolation by confirming that the Rollback Probe row disappears after the test.
  • Complete the optional Secret Mission by adding a bookings table plus a booking-aware deletion guard. Return 409 when an event has bookings while keeping the event and booking records intact.

Ready to quiz yourself?