Concurrency-Safe Inventory API: Part 4

Introduction

30 Second Summary

Two shoppers can reach checkout while the same last item still appears to be available. Without careful handling, both orders can be approved.

In this project, you will build a searchable product inventory API using Python, FastAPI, and PostgreSQL. You will prove that it can filter a large catalog while protecting scarce stock from simultaneous reservations.

What You'll Build

You'll search a 50-product catalog, narrow the results to the products you need, and watch two simultaneous reservations compete for limited stock while only one succeeds.

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

• A searchable product catalog where you combine text, category, and price filters to find matching products.
• A deterministic browsing experience where every page includes sorted products plus clear pagination metadata.
• A concurrency-safe reservation flow where simultaneous requests cannot oversell stock, backed by automated tests that prove the final quantity stays correct.
• Secret Mission: Add a grouped category summary that reports each category's product count and total available stock.

Are there any prerequisites?

This hard project assumes you completed Project 2 or have equivalent familiarity with FastAPI, Tortoise ORM, PostgreSQL, asynchronous Python, and pytest.

You'll also need a Windows machine with Python 3.10+, psql, PostgreSQL, and Visual Studio Code available.

Before We Start

Before the hands-on work begins, lock in who uses this inventory API and why accurate stock matters. This gives every search and reservation feature a clear purpose.

Set Up the Windows Project and Database

The concurrency work ahead depends on every request reaching the same PostgreSQL database. A pinned Python environment keeps every request on the same dependency versions.

This step creates that shared foundation with Windows PowerShell and Visual Studio Code. You will install the pinned FastAPI stack before creating the local database.

In this step, get ready to:

• Create an isolated Python environment with the pinned project packages.
• Create the local PostgreSQL database with a process-scoped connection string.
• Add the empty application packages before verifying the complete setup.

Create the pinned Python environment

A virtual environment keeps this project's packages separate from other Python projects on your machine. The version pins make the setup reproducible when you run the API and its tests.

• Press the Windows key to open Windows search.
• Type PowerShell into the search bar.
• Press Enter to open Windows PowerShell.
• Confirm Visual Studio Code is available from this PowerShell session by running:

code --version

What does this check show?

The command prints the Visual Studio Code version details. It also confirms that PowerShell can launch the editor through the code command.

You should see version information followed by additional build details.

PowerShell Cannot Find Visual Studio Code?

The editor can be installed while its command is unavailable to the current terminal. Install the Windows User Setup through WinGet by running:

winget install Microsoft.VisualStudioCode

This installs Visual Studio Code for your Windows user. Close PowerShell after the installation completes.

• Reopen PowerShell through Windows search.
• Run the version check again.

If the command remains unavailable, help me make the Visual Studio Code command available in PowerShell.

• Create the product-inventory folder on your Desktop by running these commands:

Set-Location ~\Desktop
New-Item -Path . -Name "product-inventory" -ItemType "Directory"
Set-Location -Path .\product-inventory -PassThru

What do these commands do?

• The first command moves PowerShell to your Desktop.
• The second command creates the product-inventory folder.
• The final command enters that folder. Its output shows the full project path.

You should see a path ending in Desktop\product-inventory.

Does the project folder already exist?

A folder from an earlier attempt can stop the creation command. Rename or delete that old product-inventory folder before running the commands again.

If you need help protecting files from an earlier attempt, help me safely resolve an existing project folder.

• Prepare the isolated Python environment by running these commands:

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python --version
python -m pip --version

How does the virtual environment help?

The first command creates .venv inside product-inventory. The activation command points this PowerShell session at its isolated Python installation.

The final checks confirm that the active environment has Python 3.10 or higher. They also confirm that its package installer is ready.

Your PowerShell prompt should now begin with (.venv). The version output should report Python 3.10 or higher.

Did the virtual environment fail to activate?

Confirm that PowerShell is still inside Desktop\product-inventory. The activation path only works after the .venv folder has been created there.

For an environment-specific activation problem, help me activate my Windows Python virtual environment.

• Open the current product-inventory folder in Visual Studio Code by running:

code .

What does the dot mean?

The dot represents the folder currently open in PowerShell. Visual Studio Code opens product-inventory without changing the active terminal session.

• Select the Explorer icon in the left sidebar.
• Click the New File icon at the top of the Explorer.
• Enter requirements.txt as the file name.
• Press Enter to create the file.
• Paste the pinned dependency list into requirements.txt from the code below:

fastapi==0.162.0
pydantic==2.14.3
tortoise-orm==1.1.9
asyncpg==0.32.0
uvicorn==0.54.0
httpx==0.28.1
pytest==9.2.0
anyio==4.16.0
asgi-lifespan==2.1.0

What do these pins control?

• FastAPI defines the API routes. Uvicorn runs the local web server.
• Pydantic validates request data. Tortoise ORM maps Python models to database tables.
• asyncpg connects Tortoise ORM to PostgreSQL.
• HTTPX, pytest, AnyIO, and asgi-lifespan support the asynchronous test suite.

• Save requirements.txt with Ctrl+S.
• Return to the active PowerShell window.
• Install the pinned packages before listing the installed versions by running:

python -m pip install -r requirements.txt
python -m pip list

What happens during installation?

The first command reads every pin from requirements.txt. It installs those packages into the active .venv environment.

The second command lists the resulting environment. This makes each installed package version visible before you continue.

You should see rows for all nine pinned project packages. Their versions should match requirements.txt.

Did a package fail to install?

Check that your prompt starts with (.venv). Confirm that requirements.txt is saved inside product-inventory.

If the failure mentions compatibility or downloading, help me install the pinned inventory API packages.

✔️ Awesome, I've got everything!

Great. Save requirements.txt before continuing with the database.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference.

fastapi==0.162.0
pydantic==2.14.3
tortoise-orm==1.1.9
asyncpg==0.32.0
uvicorn==0.54.0
httpx==0.28.1
pytest==9.2.0
anyio==4.16.0
asgi-lifespan==2.1.0

Create the PostgreSQL database

The API needs one real database that every request can share. That shared state makes the later concurrency test meaningful.

• Return to the PowerShell window with the active .venv environment.
• Create the local product_inventory database by running this command:

createdb -U postgres -W product_inventory

What does this database command do?

The command connects as the postgres user. The -W flag requests the PostgreSQL password before creating product_inventory.

The password prompt is expected. Enter the working PostgreSQL password from your earlier project.

Before you check the result, do you expect PostgreSQL to report the new database name?

• Verify the database by running:

psql -U postgres -d product_inventory -c "SELECT current_database();"

What does this query prove?

The command connects directly to product_inventory. The query asks PostgreSQL to return the name of the active database.

A successful result proves that the database exists. It also proves that the postgres user can connect to it.

You should see product_inventory in the result.

Could PostgreSQL Not Create or Open the Database?

Confirm that you entered the working password for the local postgres user. A different Windows password does not authenticate that database role.

If the database already exists, run the verification command directly. For another database error, help me verify my local product_inventory database.

Your connection string contains a database password. This assignment keeps it in the current PowerShell process so the project files do not store that credential.

• Replace your-password with the working PostgreSQL password in the command below.
• Set the process-scoped connection string by running the edited command:

$Env:DATABASE_URL = "postgres://postgres:your-password@localhost:5432/product_inventory"

Where is the connection string stored?

PowerShell stores DATABASE_URL in the current process. Visual Studio Code and the API processes launched from this session can read it.

Closing this PowerShell window removes the process-scoped value. Keep this session open while you work through the project.

PowerShell should return to the prompt without printing the connection string. Your database password stays out of the project files.

Did the connection string assignment fail?

Keep the double quotes around the complete URL. Replace only the plain your-password placeholder.

For help with characters in your PostgreSQL password, help me set DATABASE_URL safely in PowerShell.

Finish the project structure

Python package files give the application and test code stable import paths. The files begin empty because later steps add the actual catalog behavior.

• Create the app package and the tests package by running these commands:

New-Item -ItemType "Directory" -Path ".\app", ".\tests"
New-Item -ItemType "File" -Path ".\app\__init__.py", ".\tests\__init__.py"

What does this structure prepare?

The first command creates the two package folders. The second command adds an empty __init__.py file to each folder.

These files let Python import code from app and tests as the project grows.

• Return to Visual Studio Code.
• Select the Explorer icon in the left sidebar.

You should see requirements.txt beside the app and tests folders. Each folder should contain an empty __init__.py file.

Before the final check, which result should confirm the database connection?

• Run the final database check and package check from the same active PowerShell session:

psql -U postgres -d product_inventory -c "SELECT current_database();"
python -m pip list

What does the final check confirm?

The first command proves that the local database is reachable from this session. The second command shows the packages installed inside the active virtual environment.

Together they confirm that the project uses one prepared PowerShell session for the database, Python environment, and editor.

The database query should return product_inventory.

The package list should include FastAPI 0.162.0, Pydantic 2.14.3, Tortoise ORM 1.1.9, asyncpg 0.32.0, Uvicorn 0.54.0, HTTPX 0.28.1, pytest 9.2.0, AnyIO 4.16.0, and asgi-lifespan 2.1.0.

That's the shared foundation in place. Next, you will build the catalog models before seeding all 50 products into PostgreSQL.

Build and Seed the Catalog

Your pinned environment now points to one local PostgreSQL database. This gives every API request the same durable source of inventory data.

A complete catalog response gives you an early working result. Its fifty product objects also reveal why a growing catalog needs search plus pagination.

In this step, get ready to:

• Define the database configuration plus the category, product, and reservation models.
• Build a temporary FastAPI endpoint that returns every stored product.
• Migrate the database before seeding fifty deterministic products.

Create the database foundation

The database configuration connects Tortoise ORM to the connection string in your active PowerShell session. The models describe the three tables that your first migration creates.

• Switch back to Visual Studio Code.
• Create app/config.py inside the existing app folder from the Explorer sidebar.
• Paste the database configuration into app/config.py using the code below:

import os

DATABASE_URL = os.getenv("DATABASE_URL")

if not DATABASE_URL:
    raise RuntimeError(
        "DATABASE_URL is not set. Set it in PowerShell before running the app."
    )

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

What does this configuration do?

• The DATABASE_URL value reads the connection string from your active PowerShell session.
• The runtime check stops startup when that connection string is missing.
• The TORTOISE_ORM dictionary connects the default database to app.models.
• The migration setting places generated migration packages under app.migrations.

• Save app/config.py.
• Create app/models.py inside the existing app folder from the Explorer sidebar.
• Add the field imports plus the category model to app/models.py using the code below:

from tortoise import fields
from tortoise.models import Model


class Category(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100, unique=True)

    class Meta:
        table = "categories"

What does the category model do?

• The Category model gives each category a numeric primary key.
• The unique name field prevents duplicate category names.
• The metadata maps the model to the categories table.

• Save app/models.py.
• Switch back to the PowerShell session from the previous step.
• Create the configured migration package by running this command:

tortoise -c app.config.TORTOISE_ORM init

What does this command do?

The command reads app.config.TORTOISE_ORM. It creates an importable migration package for the configured models.

You should see command output confirming that the migration package was initialized. The new package now lives under app/migrations.

Migration package did not initialize?

Confirm that the virtual environment remains active. Check that the same PowerShell session still holds DATABASE_URL.

Check that app/config.py points to app.models.

Use help me troubleshoot Tortoise migration initialization if the command still fails.

• Switch back to app/models.py in Visual Studio Code.
• Add the product model below the Category model using the code below:

class Product(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=200)
    description = fields.TextField()
    price = fields.DecimalField(max_digits=10, decimal_places=2)
    category = fields.ForeignKeyField(
        "models.Category",
        related_name="products",
        on_delete=fields.OnDelete.CASCADE,
    )
    stock = fields.IntField(default=0)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "products"

What does the product model do?

• The decimal price field stores currency values with two decimal places.
• The category foreign key links each product to one category.
• The stock field stores the available inventory quantity.
• The created_at field records when each product row is created.

• Add the reservation model below the Product model using the code below:

class Reservation(Model):
    id = fields.IntField(primary_key=True)
    product = fields.ForeignKeyField(
        "models.Product",
        related_name="reservations",
        on_delete=fields.OnDelete.RESTRICT,
    )
    quantity = fields.IntField()
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "reservations"

What does the reservation model do?

• Each reservation points to the product whose stock was claimed.
• The quantity field records how many units were reserved.
• The restricted delete rule protects durable reservation history.

• Save app/models.py.
• Switch back to the PowerShell session from earlier.
• Generate the first migration before applying it to PostgreSQL by running these commands:

tortoise -c app.config.TORTOISE_ORM makemigrations
tortoise -c app.config.TORTOISE_ORM migrate

What do these commands do?

• The first command compares the registered models with the migration history.
• The second command applies the generated migration to product_inventory.
• The applied migration creates the categories, products, plus reservations tables.

The migration output should confirm that the new migration was applied. Your Python models now have matching PostgreSQL tables.

Migration did not apply?

Confirm that Category, Product, plus Reservation all inherit from Model.

Check the parentheses around both ForeignKeyField definitions. Check the indentation of each nested Meta class.

Use help me fix my first Tortoise migration if the migration still fails.

✔️ Awesome, I've got everything!

Your database configuration is saved. Your three models now have matching PostgreSQL tables.

ⓧ I'd like to double check the full code

Compare your complete app/config.py file with this reference.

import os

DATABASE_URL = os.getenv("DATABASE_URL")

if not DATABASE_URL:
    raise RuntimeError(
        "DATABASE_URL is not set. Set it in PowerShell before running the app."
    )

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

Compare your complete app/models.py file with this reference.

from tortoise import fields
from tortoise.models import Model


class Category(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=100, unique=True)

    class Meta:
        table = "categories"


class Product(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=200)
    description = fields.TextField()
    price = fields.DecimalField(max_digits=10, decimal_places=2)
    category = fields.ForeignKeyField(
        "models.Category",
        related_name="products",
        on_delete=fields.OnDelete.CASCADE,
    )
    stock = fields.IntField(default=0)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "products"


class Reservation(Model):
    id = fields.IntField(primary_key=True)
    product = fields.ForeignKeyField(
        "models.Product",
        related_name="reservations",
        on_delete=fields.OnDelete.RESTRICT,
    )
    quantity = fields.IntField()
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "reservations"

Build the temporary catalog API

A response schema controls the product fields that clients receive. Pydantic validates that shape before FastAPI sends each product to the browser.

• Create app/schemas.py inside the existing app folder from the Explorer sidebar.
• Add the product response schema to app/schemas.py using the code below:

from datetime import datetime

from pydantic import BaseModel


class ProductOut(BaseModel):
    id: int
    name: str
    description: str
    price: float
    category_id: int
    stock: int
    created_at: datetime

What does the response schema do?

• The ProductOut schema lists the exact fields returned for one product.
• The schema converts each database price into a JSON-friendly floating-point value.
• The category_id field exposes the linked category identifier.

• Save app/schemas.py.

✔️ Awesome, I've got everything!

Your response schema now describes every field in one catalog product.

ⓧ I'd like to double check the full code

Compare your complete app/schemas.py file with this reference.

from datetime import datetime

from pydantic import BaseModel


class ProductOut(BaseModel):
    id: int
    name: str
    description: str
    price: float
    category_id: int
    stock: int
    created_at: datetime

• Create app/main.py inside the existing app folder from the Explorer sidebar.
• Add the application lifecycle plus the FastAPI instance to app/main.py using the code below:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from tortoise.contrib.fastapi import RegisterTortoise

from app.config import TORTOISE_ORM
from app.models import Product
from app.schemas import ProductOut


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)

What does the application lifecycle do?

• The lifespan context connects Tortoise ORM when FastAPI starts.
• The same context closes the database integration during application shutdown.
• The app object gives the server a FastAPI application to run.

• Save app/main.py.

The server keeps PowerShell busy

Uvicorn stays attached to PowerShell while the API is running. You will stop this first check before continuing to edit the route.

• Switch back to the PowerShell session from earlier.
• Check the application lifecycle by starting the server with this command:

uvicorn app.main:app --reload

What does this server command do?

The command imports the app object from app.main. Reload mode watches the project files for saved changes.

The terminal should confirm that application startup finished. This proves the lifecycle can connect to the migrated database.

• Press Ctrl+C in PowerShell to stop the server.

Application did not start?

Check that yield sits inside both context blocks. Confirm that lifespan=lifespan remains inside the FastAPI call.

A database connection problem usually means the current PowerShell session cannot read DATABASE_URL.

Use help me start the FastAPI lifecycle if startup still fails.

• Switch back to app/main.py in Visual Studio Code.
• Add the temporary product route below the app definition using the code below:

@app.get("/products", response_model=list[ProductOut])
async def list_products() -> list[ProductOut]:
    rows = await Product.all().values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    return [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

What does the product route do?

• The GET /products route requests every product row from PostgreSQL.
• The values query retrieves only fields declared by ProductOut.
• The list conversion validates each row before FastAPI serializes the response.

• Save app/main.py.

✔️ Awesome, I've got everything!

Your temporary endpoint is ready to return every stored product.

ⓧ I'd like to double check the full code

Compare your complete app/main.py file with this reference.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from tortoise.contrib.fastapi import RegisterTortoise

from app.config import TORTOISE_ORM
from app.models import Product
from app.schemas import ProductOut


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/products", response_model=list[ProductOut])
async def list_products() -> list[ProductOut]:
    rows = await Product.all().values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    return [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

Seed the catalog

A seed script fills an empty database with repeatable fixtures. Deterministic names, prices, plus stock values make the later search tests predictable.

• Create seed.py in the Product Inventory project folder from the Explorer sidebar.
• Add the imports plus the first category fixture to seed.py using the code below:

import asyncio
from decimal import Decimal

from tortoise import Tortoise

from app.config import TORTOISE_ORM
from app.models import Category, Product, Reservation


CATALOG = {
    "Electronics": (
        Decimal("25.00"),
        [
            "Laptop Stand",
            "Wireless Mouse",
            "Mechanical Keyboard",
            "USB-C Hub",
            "Webcam",
            "Portable SSD",
            "Noise-Cancelling Headphones",
            "Monitor Light",
            "Bluetooth Speaker",
            "Ergonomic Trackball",
        ],
    ),
}

What does this catalog fixture hold?

• The dictionary groups ten product names under one category.
• The decimal value gives that category a repeatable starting price.
• The seed function uses each product position to calculate its final price plus stock.

• Add the initial seed function below the closing brace in seed.py using the code below:

async def seed() -> None:
    await Tortoise.init(config=TORTOISE_ORM)

    await Reservation.all().delete()
    await Product.all().delete()
    await Category.all().delete()

    product_count = 0

    for category_index, (category_name, catalog_data) in enumerate(
        CATALOG.items()
    ):
        base_price, product_names = catalog_data
        category = await Category.create(name=category_name)

    print(
        f"Seeded {product_count} products across {len(CATALOG)} categories."
    )
    await Tortoise.close_connections()


if __name__ == "__main__":
    asyncio.run(seed())

What does this first seed pass do?

• The function initializes Tortoise with the same database configuration as the API.
• The delete operations clear dependent records in foreign-key order.
• The outer loop creates each category currently listed in CATALOG.
• The final call closes the database connections after the script finishes.

• Save seed.py.
• Switch back to the PowerShell session from earlier.
• Run the first seed pass with this command:

python seed.py

What does this first run prove?

The script should report one created category with zero products. This proves database initialization plus category creation work before you add the product loop.

• Switch back to seed.py in Visual Studio Code.
• Find the line that creates the category inside seed().
• Add the product creation loop below that line using the code below:

        for product_index, product_name in enumerate(product_names):
            await Product.create(
                name=product_name,
                description=(
                    f"A practical {product_name.lower()} for work, home, or travel."
                ),
                price=(
                    base_price
                    + Decimal(product_index * 7)
                    + Decimal("0.99")
                ),
                category=category,
                stock=(category_index + 1) * 5 + product_index,
            )
            product_count += 1

How are the products generated?

• The inner loop creates one row for every product name in the current category.
• The price calculation combines the category base price with a repeatable product offset.
• The stock calculation combines the category position with the product position.
• The counter records how many product rows were created.

• Save seed.py.
• Switch back to the PowerShell session from earlier.
• Seed the first ten product rows by running this command:

python seed.py

What changed in the database?

The script should now report ten products across one category. Each run clears the old fixtures before rebuilding them.

• Switch back to the CATALOG dictionary in seed.py.
• Insert the Home fixture after the Electronics tuple using the code below:

    "Home": (
        Decimal("12.00"),
        [
            "Desk Lamp",
            "Storage Basket",
            "Cotton Throw",
            "Kitchen Scale",
            "Ceramic Planter",
            "Wall Clock",
            "Water Filter Jug",
            "Tool Organizer",
            "Reading Cushion",
            "Laundry Hamper",
        ],
    ),

What does this fixture add?

The Home tuple adds ten household products. Its lower base price produces a different deterministic price range.

• Save seed.py.
• Switch back to the PowerShell session from earlier.
• Rebuild the fixtures with the second category by running this command:

python seed.py

What should the second category produce?

The seed script should now report twenty products across two categories. This confirms the same loop can process another fixture group.

• Switch back to the CATALOG dictionary in seed.py.
• Insert the Fitness fixture after the Home tuple using the code below:

    "Fitness": (
        Decimal("18.00"),
        [
            "Yoga Mat",
            "Resistance Bands",
            "Foam Roller",
            "Kettlebell",
            "Jump Rope",
            "Training Gloves",
            "Balance Board",
            "Running Belt",
            "Exercise Ball",
            "Recovery Strap",
        ],
    ),

What does this fixture add?

The Fitness tuple contributes ten products with a separate base price. The category position also gives these rows a higher stock range.

• Save seed.py.
• Switch back to the PowerShell session from earlier.
• Rebuild the fixtures with the third category by running this command:

python seed.py

What should the third category produce?

The script should now report thirty products across three categories. Your growing catalog is using one repeatable creation path.

• Switch back to the CATALOG dictionary in seed.py.
• Insert the Books fixture after the Fitness tuple using the code below:

    "Books": (
        Decimal("8.00"),
        [
            "Async Python Handbook",
            "Database Design Patterns",
            "Practical API Testing",
            "PostgreSQL Field Guide",
            "Reliable Systems Workbook",
            "Clean Backend Architecture",
            "Concurrency Case Studies",
            "Query Optimization Notes",
            "Python Type Hints Guide",
            "Service Design Journal",
        ],
    ),

What does this fixture add?

The Books tuple adds ten technical titles. Its base price gives the final catalog a broader range for later price filters.

• Save seed.py.
• Switch back to the PowerShell session from earlier.
• Rebuild the fixtures with the fourth category by running this command:

python seed.py

What should the fourth category produce?

The script should now report forty products across four categories. The database is one category away from the complete fixture set.

• Switch back to the CATALOG dictionary in seed.py.
• Insert the Outdoors fixture after the Books tuple using the code below:

    "Outdoors": (
        Decimal("20.00"),
        [
            "Camping Lantern",
            "Trail Backpack",
            "Insulated Bottle",
            "Picnic Blanket",
            "Compact Binoculars",
            "Hiking Poles",
            "Dry Bag",
            "Pocket Compass",
            "Travel Hammock",
            "First Aid Pouch",
        ],
    ),

What does this fixture add?

The Outdoors tuple completes the five-category fixture set. The same nested loop now creates all fifty products.

• Save seed.py.

Before you run the final seed, how many products do you expect the script to create?

• Switch back to the PowerShell session from earlier.
• Build the complete catalog by running this command:

python seed.py

What should the final seed report?

You should see Seeded 50 products across 5 categories. in PowerShell. The database now contains the complete deterministic catalog.

Seed command failed?

Confirm that the migration completed before running the seed script. Check that the active PowerShell session still holds DATABASE_URL.

Check the closing brackets around every category fixture. Confirm that the product loop remains inside the category loop.

Use help me troubleshoot the catalog seed script if the database still does not seed.

✔️ Awesome, I've got everything!

Your seed script now rebuilds five categories plus fifty deterministic products.

ⓧ I'd like to double check the full code

Compare your complete seed.py file with this reference.

import asyncio
from decimal import Decimal

from tortoise import Tortoise

from app.config import TORTOISE_ORM
from app.models import Category, Product, Reservation


CATALOG = {
    "Electronics": (
        Decimal("25.00"),
        [
            "Laptop Stand",
            "Wireless Mouse",
            "Mechanical Keyboard",
            "USB-C Hub",
            "Webcam",
            "Portable SSD",
            "Noise-Cancelling Headphones",
            "Monitor Light",
            "Bluetooth Speaker",
            "Ergonomic Trackball",
        ],
    ),
    "Home": (
        Decimal("12.00"),
        [
            "Desk Lamp",
            "Storage Basket",
            "Cotton Throw",
            "Kitchen Scale",
            "Ceramic Planter",
            "Wall Clock",
            "Water Filter Jug",
            "Tool Organizer",
            "Reading Cushion",
            "Laundry Hamper",
        ],
    ),
    "Fitness": (
        Decimal("18.00"),
        [
            "Yoga Mat",
            "Resistance Bands",
            "Foam Roller",
            "Kettlebell",
            "Jump Rope",
            "Training Gloves",
            "Balance Board",
            "Running Belt",
            "Exercise Ball",
            "Recovery Strap",
        ],
    ),
    "Books": (
        Decimal("8.00"),
        [
            "Async Python Handbook",
            "Database Design Patterns",
            "Practical API Testing",
            "PostgreSQL Field Guide",
            "Reliable Systems Workbook",
            "Clean Backend Architecture",
            "Concurrency Case Studies",
            "Query Optimization Notes",
            "Python Type Hints Guide",
            "Service Design Journal",
        ],
    ),
    "Outdoors": (
        Decimal("20.00"),
        [
            "Camping Lantern",
            "Trail Backpack",
            "Insulated Bottle",
            "Picnic Blanket",
            "Compact Binoculars",
            "Hiking Poles",
            "Dry Bag",
            "Pocket Compass",
            "Travel Hammock",
            "First Aid Pouch",
        ],
    ),
}


async def seed() -> None:
    await Tortoise.init(config=TORTOISE_ORM)

    await Reservation.all().delete()
    await Product.all().delete()
    await Category.all().delete()

    product_count = 0

    for category_index, (category_name, catalog_data) in enumerate(
        CATALOG.items()
    ):
        base_price, product_names = catalog_data
        category = await Category.create(name=category_name)

        for product_index, product_name in enumerate(product_names):
            await Product.create(
                name=product_name,
                description=(
                    f"A practical {product_name.lower()} for work, home, or travel."
                ),
                price=(
                    base_price
                    + Decimal(product_index * 7)
                    + Decimal("0.99")
                ),
                category=category,
                stock=(category_index + 1) * 5 + product_index,
            )
            product_count += 1

    print(
        f"Seeded {product_count} products across {len(CATALOG)} categories."
    )
    await Tortoise.close_connections()


if __name__ == "__main__":
    asyncio.run(seed())

Before you start the API, what do you expect the unpaginated endpoint to look like with fifty product objects?

Keep this server running

The server stays attached to PowerShell during the final check. Leave that terminal running while you inspect the endpoint in your browser.

• Start the seeded Product Inventory API by running this command:

uvicorn app.main:app --reload

What happens during startup?

Uvicorn loads the FastAPI app. The lifespan connects Tortoise ORM to the seeded PostgreSQL database.

• Open http://127.0.0.1:8000/products in your browser.

You should see fifty product objects backed by PostgreSQL. Each object includes a description, price, category ID, stock quantity, plus creation timestamp.

That fifty-item response is the intended result. The terminal should also record the successful request.

Products did not load?

Check the Uvicorn terminal for an import problem or database connection problem. Confirm that the browser address ends with /products.

Run the seed script again if the endpoint returns an empty list. Restart the server after the seed command completes.

Use help me load the fifty seeded products if the endpoint still does not work.

That is the first working catalog slice complete: your migrated PostgreSQL data now flows through FastAPI. Next, you will replace the fifty-item wall with validated search, sorting, plus pagination.

Add Search, Sorting, and Pagination

Your seeded FastAPI catalog can read all 50 products from PostgreSQL. That full response becomes difficult to navigate as the catalog grows.

This step gives clients precise query controls. The database handles filtering, ordering, counting, and slicing.

In this step, get ready to:

• Validate eight search, filter, sorting, and pagination parameters.
• Return deterministic pages with result metadata.
• Prove the catalog contract with asynchronous HTTP tests.

Define validated query parameters

A Pydantic model creates one validation boundary for every catalog query. Invalid input is rejected before the route builds a database query.

• Switch back to app/schemas.py in Visual Studio Code.
• Replace the existing imports at the top of app/schemas.py with the imports below.
• Add ProductQuery directly above the existing ProductOut class by using this code:

from datetime import datetime
from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ProductQuery(BaseModel):
    model_config = ConfigDict(extra="forbid")

    q: str | None = Field(default=None, min_length=1, max_length=100)
    category_id: int | None = Field(default=None, ge=1)
    min_price: Decimal | None = Field(default=None, ge=0)
    max_price: Decimal | None = Field(default=None, ge=0)
    sort_by: Literal["name", "price", "created_at"] = "created_at"
    sort_direction: Literal["asc", "desc"] = "asc"
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1, le=50)

    @model_validator(mode="after")
    def validate_price_range(self):
        if (
            self.min_price is not None
            and self.max_price is not None
            and self.min_price > self.max_price
        ):
            raise ValueError("min_price must be less than or equal to max_price")
        return self

What does this model validate?

• The extra="forbid" configuration rejects parameters outside the API contract.
• The Field constraints protect search length, positive identifiers, non-negative prices, and page limits.
• The Literal types allow three sorting fields plus two sorting directions.
• The validate_price_range method rejects a minimum price above the maximum price.

• Keep the existing ProductOut class below ProductQuery.
• Add ProductPage below ProductOut by using this code:

class ProductPage(BaseModel):
    items: list[ProductOut]
    total: int
    page: int
    page_size: int
    total_pages: int

What does the page model add?

The items field carries the current product slice. The remaining fields describe its position within the filtered result set.

• Save app/schemas.py in Visual Studio Code.
• Switch back to the activated PowerShell session from the previous step.
• Check that the application can import the updated schemas by running:

uvicorn app.main:app --reload

What does this check prove?

Uvicorn imports app.main. That module loads app.schemas, so a successful startup confirms that the new models are valid.

• Return to the browser from the previous step.
• Refresh http://127.0.0.1:8000/products to confirm the temporary product route still responds.

You should still see the seeded products. Your application now imports the query contract successfully.

• Stop Uvicorn by pressing Ctrl+C in PowerShell.

Schema import failing?

Confirm that ProductQuery appears above ProductOut. Confirm that ProductPage appears below it.

Compare your imports with the reference. A missing Decimal or model_validator import prevents the module from loading.

Share the traceback with AI using help me diagnose my schema import failure.

✔️ Awesome, I've got everything!

Your query model and page model are saved. The application can import both definitions.

ⓧ I'd like to double check the full code

Compare your complete app/schemas.py file with this reference.

from datetime import datetime
from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ProductQuery(BaseModel):
    model_config = ConfigDict(extra="forbid")

    q: str | None = Field(default=None, min_length=1, max_length=100)
    category_id: int | None = Field(default=None, ge=1)
    min_price: Decimal | None = Field(default=None, ge=0)
    max_price: Decimal | None = Field(default=None, ge=0)
    sort_by: Literal["name", "price", "created_at"] = "created_at"
    sort_direction: Literal["asc", "desc"] = "asc"
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1, le=50)

    @model_validator(mode="after")
    def validate_price_range(self):
        if (
            self.min_price is not None
            and self.max_price is not None
            and self.min_price > self.max_price
        ):
            raise ValueError("min_price must be less than or equal to max_price")
        return self


class ProductOut(BaseModel):
    id: int
    name: str
    description: str
    price: float
    category_id: int
    stock: int
    created_at: datetime


class ProductPage(BaseModel):
    items: list[ProductOut]
    total: int
    page: int
    page_size: int
    total_pages: int

What should match?

Check all eight query fields. Check the price-range validator plus the five pagination response fields.

Build the lazy catalog query

A lazy Tortoise ORM queryset collects each requested filter before PostgreSQL executes it. Counting and slicing the filtered query keeps the response efficient.

• Switch back to app/main.py in Visual Studio Code.
• Replace the imports at the top of app/main.py with this import section:

from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q

from app.config import TORTOISE_ORM
from app.models import Product
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
)

Why are these imports needed?

• The Query marker tells FastAPI to populate ProductQuery from the URL query string.
• The Q expression combines name search with description search.
• The ceil function converts the filtered count into a whole number of pages.

• Select the temporary list_products route from its decorator through its return statement.
• Replace the selected route so that app/main.py matches the full-file reference below.

from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q

from app.config import TORTOISE_ORM
from app.models import Product
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/products", response_model=ProductPage)
async def list_products(
    params: Annotated[ProductQuery, Query()],
) -> ProductPage:
    queryset = Product.all()

    if params.q is not None:
        queryset = queryset.filter(
            Q(name__icontains=params.q) | Q(description__icontains=params.q)
        )
    if params.category_id is not None:
        queryset = queryset.filter(category_id=params.category_id)
    if params.min_price is not None:
        queryset = queryset.filter(price__gte=params.min_price)
    if params.max_price is not None:
        queryset = queryset.filter(price__lte=params.max_price)

    total = await queryset.count()
    total_pages = ceil(total / params.page_size) if total else 0

    if params.page > max(total_pages, 1):
        raise HTTPException(status_code=422, detail="page exceeds available pages")

    primary_order = (
        params.sort_by
        if params.sort_direction == "asc"
        else f"-{params.sort_by}"
    )
    offset = (params.page - 1) * params.page_size

    rows = await queryset.order_by(primary_order, "id").offset(offset).limit(
        params.page_size
    ).values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    items = [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

    return ProductPage(
        items=items,
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=total_pages,
    )

How does the route work?

The route begins with Product.all() as a lazy queryset. Each supplied parameter adds a matching database filter.

The filtered count determines total_pages. Requests beyond that boundary receive a validation response.

The requested field supplies the primary order. The id field provides stable ordering when primary values match.

The offset skips earlier pages. The limit retrieves only the current slice.

• Save app/main.py in Visual Studio Code.
• Switch back to the activated PowerShell session.
• Start the updated API by running:

uvicorn app.main:app --reload

What does reload mode do?

Uvicorn serves the FastAPI application from app.main. Reload mode watches the project files while the local server runs.

Before checking the route, how many products do you expect the page-size limit to allow?

• Enter http://127.0.0.1:8000/products?q=desk&sort_by=price&sort_direction=desc&page=1&page_size=5 in your browser address bar.

You should see an items array with no more than five products. The response also includes total, page, page_size, and total_pages.

Good progress. Your catalog now returns a focused database-backed page instead of an unwieldy full list.

• Stop Uvicorn by pressing Ctrl+C in PowerShell.

Endpoint failing after the route change?

Confirm that the old unpaginated route has been removed. Only one list_products function should remain.

Check the response body if a supported parameter is rejected. Its spelling must match a field in ProductQuery.

Share the traceback or response body with AI using help me debug my paginated product route.

✔️ Awesome, I've got everything!

Your endpoint validates query parameters. It returns a deterministic slice from the filtered catalog.

ⓧ I'd like to double check the full code

Compare your complete app/main.py file with this reference.

from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q

from app.config import TORTOISE_ORM
from app.models import Product
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/products", response_model=ProductPage)
async def list_products(
    params: Annotated[ProductQuery, Query()],
) -> ProductPage:
    queryset = Product.all()

    if params.q is not None:
        queryset = queryset.filter(
            Q(name__icontains=params.q) | Q(description__icontains=params.q)
        )
    if params.category_id is not None:
        queryset = queryset.filter(category_id=params.category_id)
    if params.min_price is not None:
        queryset = queryset.filter(price__gte=params.min_price)
    if params.max_price is not None:
        queryset = queryset.filter(price__lte=params.max_price)

    total = await queryset.count()
    total_pages = ceil(total / params.page_size) if total else 0

    if params.page > max(total_pages, 1):
        raise HTTPException(status_code=422, detail="page exceeds available pages")

    primary_order = (
        params.sort_by
        if params.sort_direction == "asc"
        else f"-{params.sort_by}"
    )
    offset = (params.page - 1) * params.page_size

    rows = await queryset.order_by(primary_order, "id").offset(offset).limit(
        params.page_size
    ).values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    items = [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

    return ProductPage(
        items=items,
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=total_pages,
    )

What should match?

Check the query-model annotation, conditional filters, filtered count, boundary check, ordering, offset, limit, and response metadata.

The id tie-breaker must follow the requested primary order.

Prove the catalog contract

One browser request proves the successful path. An asynchronous pytest matrix checks valid combinations plus every important rejection path.

The test client uses HTTPX against the application in memory. Each test receives clean registered tables.

• Create tests/conftest.py with the new-file control in the Visual Studio Code sidebar.
• Add the asynchronous client fixture by using this code:

import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from tortoise.contrib.test import truncate_all_models

from app.main import app


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def client():
    async with LifespanManager(app) as manager:
        await truncate_all_models()
        transport = ASGITransport(app=manager.app)
        async with AsyncClient(
            transport=transport,
            base_url="http://test",
        ) as test_client:
            yield test_client

What does the client fixture do?

• The anyio_backend fixture selects the asynchronous backend.
• The LifespanManager starts the application lifecycle before each test.
• The table cleanup isolates every test case.
• The AsyncClient sends requests directly to the application.

• Save tests/conftest.py in Visual Studio Code.
• Create tests/test_products.py with the new-file control in the editor sidebar.
• Build the catalog test module so that it matches this complete reference:

from decimal import Decimal

import pytest

from app.models import Category, Product

pytestmark = pytest.mark.anyio


async def seed_small_catalog():
    office = await Category.create(name="Office")
    home = await Category.create(name="Home")

    products = [
        await Product.create(
            name="Desk Lamp",
            description="Adjustable desk lighting",
            price=Decimal("30.00"),
            category=home,
            stock=10,
        ),
        await Product.create(
            name="Desk Mat",
            description="Protective office desk mat",
            price=Decimal("20.00"),
            category=office,
            stock=12,
        ),
        await Product.create(
            name="Monitor Stand",
            description="Raised desk monitor platform",
            price=Decimal("40.00"),
            category=office,
            stock=8,
        ),
        await Product.create(
            name="Storage Basket",
            description="Woven home storage",
            price=Decimal("20.00"),
            category=home,
            stock=15,
        ),
    ]
    return office, home, products


async def test_combined_filters_sorting_and_pagination(client):
    office, _, _ = await seed_small_catalog()

    response = await client.get(
        "/products",
        params={
            "q": "desk",
            "category_id": office.id,
            "min_price": "10.00",
            "max_price": "35.00",
            "sort_by": "price",
            "sort_direction": "desc",
            "page": 1,
            "page_size": 2,
        },
    )

    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 1
    assert body["total_pages"] == 1
    assert [item["name"] for item in body["items"]] == ["Desk Mat"]


async def test_metadata_and_page_boundary(client):
    await seed_small_catalog()

    response = await client.get(
        "/products",
        params={"page": 2, "page_size": 3, "sort_by": "name"},
    )

    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 4
    assert body["page"] == 2
    assert body["page_size"] == 3
    assert body["total_pages"] == 2
    assert len(body["items"]) == 1

    beyond = await client.get(
        "/products",
        params={"page": 3, "page_size": 3},
    )
    assert beyond.status_code == 422


async def test_sort_tie_uses_id_as_tiebreaker(client):
    _, _, products = await seed_small_catalog()

    response = await client.get(
        "/products",
        params={
            "sort_by": "price",
            "sort_direction": "asc",
            "page_size": 10,
        },
    )

    assert response.status_code == 200
    equal_price_ids = [
        item["id"]
        for item in response.json()["items"]
        if item["price"] == 20.0
    ]
    expected_ids = sorted([products[1].id, products[3].id])
    assert equal_price_ids == expected_ids


@pytest.mark.parametrize(
    "params",
    [
        {"q": ""},
        {"category_id": 0},
        {"min_price": -1},
        {"max_price": -1},
        {"min_price": 50, "max_price": 10},
        {"sort_by": "rating"},
        {"sort_direction": "sideways"},
        {"page": 0},
        {"page_size": 0},
        {"page_size": 51},
        {"unknown": "value"},
    ],
)
async def test_invalid_query_parameters_return_422(client, params):
    await seed_small_catalog()
    response = await client.get("/products", params=params)
    assert response.status_code == 422

What does the test matrix cover?

The small catalog creates predictable categories, prices, and search matches. Two equal prices create a deliberate ordering tie.

The successful tests cover combined filters, metadata, page boundaries, and stable sorting. The parametrized test checks every invalid input against the same typed contract.

• Save tests/test_products.py in Visual Studio Code.
• Switch back to the activated PowerShell session.

Before running the suite, do you expect an unknown query parameter to reach list_products?

• Run the complete catalog test matrix by running:

python -m pytest tests/test_products.py -q

What does the passing matrix prove?

The suite exercises the API through asynchronous HTTP requests. Passing results prove filtering, validation, pagination metadata, page boundaries, and deterministic ties.

You should see every catalog test pass. Unsupported query strings are rejected before the route builds its queryset.

Catalog matrix still failing?

Start with the first failing test. Its assertion identifies whether the mismatch belongs to validation, filtering, metadata, or ordering.

Confirm that PowerShell still has DATABASE_URL configured. The test lifespan uses the same local database connection as the application.

Share the first failure with AI using help me debug my catalog test matrix.

✔️ Awesome, I've got everything!

Your asynchronous catalog suite passes. The read path now has repeatable coverage for successful requests plus rejected input.

ⓧ I'd like to double check the full code

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

import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from tortoise.contrib.test import truncate_all_models

from app.main import app


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def client():
    async with LifespanManager(app) as manager:
        await truncate_all_models()
        transport = ASGITransport(app=manager.app)
        async with AsyncClient(
            transport=transport,
            base_url="http://test",
        ) as test_client:
            yield test_client

What should match in the fixture?

Check the lifespan manager, table cleanup, ASGI transport, base URL, and yielded asynchronous client.

Compare your complete tests/test_products.py file with this reference.

from decimal import Decimal

import pytest

from app.models import Category, Product

pytestmark = pytest.mark.anyio


async def seed_small_catalog():
    office = await Category.create(name="Office")
    home = await Category.create(name="Home")

    products = [
        await Product.create(
            name="Desk Lamp",
            description="Adjustable desk lighting",
            price=Decimal("30.00"),
            category=home,
            stock=10,
        ),
        await Product.create(
            name="Desk Mat",
            description="Protective office desk mat",
            price=Decimal("20.00"),
            category=office,
            stock=12,
        ),
        await Product.create(
            name="Monitor Stand",
            description="Raised desk monitor platform",
            price=Decimal("40.00"),
            category=office,
            stock=8,
        ),
        await Product.create(
            name="Storage Basket",
            description="Woven home storage",
            price=Decimal("20.00"),
            category=home,
            stock=15,
        ),
    ]
    return office, home, products


async def test_combined_filters_sorting_and_pagination(client):
    office, _, _ = await seed_small_catalog()

    response = await client.get(
        "/products",
        params={
            "q": "desk",
            "category_id": office.id,
            "min_price": "10.00",
            "max_price": "35.00",
            "sort_by": "price",
            "sort_direction": "desc",
            "page": 1,
            "page_size": 2,
        },
    )

    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 1
    assert body["total_pages"] == 1
    assert [item["name"] for item in body["items"]] == ["Desk Mat"]


async def test_metadata_and_page_boundary(client):
    await seed_small_catalog()

    response = await client.get(
        "/products",
        params={"page": 2, "page_size": 3, "sort_by": "name"},
    )

    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 4
    assert body["page"] == 2
    assert body["page_size"] == 3
    assert body["total_pages"] == 2
    assert len(body["items"]) == 1

    beyond = await client.get(
        "/products",
        params={"page": 3, "page_size": 3},
    )
    assert beyond.status_code == 422


async def test_sort_tie_uses_id_as_tiebreaker(client):
    _, _, products = await seed_small_catalog()

    response = await client.get(
        "/products",
        params={
            "sort_by": "price",
            "sort_direction": "asc",
            "page_size": 10,
        },
    )

    assert response.status_code == 200
    equal_price_ids = [
        item["id"]
        for item in response.json()["items"]
        if item["price"] == 20.0
    ]
    expected_ids = sorted([products[1].id, products[3].id])
    assert equal_price_ids == expected_ids


@pytest.mark.parametrize(
    "params",
    [
        {"q": ""},
        {"category_id": 0},
        {"min_price": -1},
        {"max_price": -1},
        {"min_price": 50, "max_price": 10},
        {"sort_by": "rating"},
        {"sort_direction": "sideways"},
        {"page": 0},
        {"page_size": 0},
        {"page_size": 51},
        {"unknown": "value"},
    ],
)
async def test_invalid_query_parameters_return_422(client, params):
    await seed_small_catalog()
    response = await client.get("/products", params=params)
    assert response.status_code == 422

What should match in the test module?

Check all four asynchronous test functions plus every invalid parameter case. The equal-price assertion must compare IDs in sorted order.

The tests replace the seeded catalog with controlled fixtures. Reseeding restores the 50-product browser demo.

• Restore the full catalog by running:

python seed.py

What does reseeding restore?

The seed script clears the test fixtures. It recreates five categories plus 50 deterministic products.

You should see confirmation that 50 products were seeded across five categories.

• Start the API for the final verification by running:

uvicorn app.main:app --reload

What is ready to verify?

The restored catalog now runs through the validated route. The next request checks search, descending price order, and page metadata together.

Before opening the URL, do you expect the response to contain more than five products?

• Enter http://127.0.0.1:8000/products?q=desk&sort_by=price&sort_direction=desc&page=1&page_size=5 in your browser address bar.

You should see filtered products inside items plus total, page, page_size, and total_pages.

The page contains no more than five products. PostgreSQL now handles the search plus the ordered slice.

• Stop Uvicorn by pressing Ctrl+C in PowerShell.

Your catalog now supports focused reads with predictable pages and tested validation. Next, you will put the inventory write path under simultaneous pressure.

Reproduce the Reservation Race

Your searchable PostgreSQL catalog now handles flexible reads with passing tests. Inventory reservations introduce a shared write path where timing affects correctness.

A single reservation can look correct while two simultaneous requests act on the same stock snapshot. This race condition can approve more inventory than the product holds.

In this step, get ready to:

• Define validated reservation request and response schemas.
• Add a deliberately unsafe reservation endpoint.
• Run concurrent requests that expose overselling.

Add the unsafe reservation path

The reservation endpoint needs a typed quantity input plus a predictable success response. Pydantic rejects quantities outside the allowed range before the route changes stock.

• Switch back to app/schemas.py in Visual Studio Code.
• Add these reservation schemas below ProductPage by pasting the following code:

class ReserveRequest(BaseModel):
    quantity: int = Field(ge=1, le=1000)


class ReservationResult(BaseModel):
    reservation_id: int
    product_id: int
    quantity: int
    remaining_stock: int

What Do These Schemas Define?

• ReserveRequest accepts quantities from 1 through 1000.
• ReservationResult gives successful requests a reservation ID.
• remaining_stock lets the client observe the stock left after a reservation.

• Save app/schemas.py.
• Confirm the catalog still works by running this command in the PowerShell session from earlier:

python -m pytest tests/test_products.py -q

What Does This Check Prove?

This reruns the existing catalog test matrix. A passing result proves the new schemas have not changed product search or pagination.

You should see the catalog tests pass. Your new reservation contracts now coexist with the product response models.

Seeing a Schema Error?

Check that both classes sit below ProductPage in app/schemas.py. Confirm that BaseModel plus Field remain imported at the top.

help me compare my reservation schemas with the required Pydantic models.

✔️ My Schema Tests Pass

Good work. Your reservation input now rejects invalid quantities before they reach the stock logic.

ⓧ I'd Like to Double Check the Full Code

Compare your complete app/schemas.py with this version.

from datetime import datetime
from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ProductQuery(BaseModel):
    model_config = ConfigDict(extra="forbid")

    q: str | None = Field(default=None, min_length=1, max_length=100)
    category_id: int | None = Field(default=None, ge=1)
    min_price: Decimal | None = Field(default=None, ge=0)
    max_price: Decimal | None = Field(default=None, ge=0)
    sort_by: Literal["name", "price", "created_at"] = "created_at"
    sort_direction: Literal["asc", "desc"] = "asc"
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1, le=50)

    @model_validator(mode="after")
    def validate_price_range(self):
        if (
            self.min_price is not None
            and self.max_price is not None
            and self.min_price > self.max_price
        ):
            raise ValueError("min_price must be less than or equal to max_price")
        return self


class ProductOut(BaseModel):
    id: int
    name: str
    description: str
    price: float
    category_id: int
    stock: int
    created_at: datetime


class ProductPage(BaseModel):
    items: list[ProductOut]
    total: int
    page: int
    page_size: int
    total_pages: int


class ReserveRequest(BaseModel):
    quantity: int = Field(ge=1, le=1000)


class ReservationResult(BaseModel):
    reservation_id: int
    product_id: int
    quantity: int
    remaining_stock: int

What Should Match?

The product schemas remain unchanged. The two reservation schemas appear at the bottom of the file.

The first reservation implementation uses a read-check-write sequence. A short asynchronous pause widens the timing window so concurrent requests can overlap consistently.

• Switch back to app/main.py in Visual Studio Code.
• Replace the import section at the top with this expanded import group:

import asyncio
from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q

from app.config import TORTOISE_ORM
from app.models import Product, Reservation
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
    ReservationResult,
    ReserveRequest,
)

Why Expand the Imports?

• asyncio provides the deliberate scheduling pause used to expose overlapping requests.
• Reservation stores a durable record for each approved request.
• ReserveRequest plus ReservationResult type the new endpoint.

• Save app/main.py.
• Verify the expanded imports by running the catalog tests again:

python -m pytest tests/test_products.py -q

What Is This Run Checking?

This loads the FastAPI application through the existing test fixture. Passing tests confirm every new import resolves correctly.

You should see the catalog suite pass again. The application can now load the reservation models plus schemas.

• Add the reservation route below list_products by pasting this code:

@app.post(
    "/products/{product_id}/reservations",
    response_model=ReservationResult,
    status_code=201,
)
async def reserve_product(product_id: int, request: ReserveRequest) -> ReservationResult:
    product = await Product.get_or_none(id=product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="product not found")
    if product.stock < request.quantity:
        raise HTTPException(status_code=409, detail="insufficient stock")
    await asyncio.sleep(0.05)
    product.stock -= request.quantity
    await product.save(update_fields=["stock"])
    reservation = await Reservation.create(
        product=product,
        quantity=request.quantity,
    )
    return ReservationResult(
        reservation_id=reservation.id,
        product_id=product.id,
        quantity=reservation.quantity,
        remaining_stock=product.stock,
    )

What Does This Route Do?

• get_or_none() loads the requested product or produces a missing-product response.
• product.stock is checked against the validated quantity before any change is saved.
• asyncio.sleep(0.05) allows another request to run after the stock check.
• Reservation.create() records every request that reaches the success path.

• Save app/main.py.
• Confirm the existing product endpoint remains intact by running:

python -m pytest tests/test_products.py -q

What Does This Regression Check Cover?

The command runs the catalog tests against the application with its new route. A passing result confirms search plus pagination still behave as before.

You should see every catalog test pass. The reservation path is now available without changing the read path.

Application Fails to Import?

Confirm that ReserveRequest plus ReservationResult are present in app/schemas.py. Check that Reservation is imported from app.models.

help me debug the imports for my new reservation endpoint.

✔️ The Catalog Tests Still Pass

That confirms the new reservation route loads without disturbing your searchable catalog.

ⓧ I'd Like to Double Check the Full Code

Compare your complete app/main.py with this intentionally unsafe version.

import asyncio
from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q

from app.config import TORTOISE_ORM
from app.models import Product, Reservation
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
    ReservationResult,
    ReserveRequest,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/products", response_model=ProductPage)
async def list_products(
    params: Annotated[ProductQuery, Query()],
) -> ProductPage:
    queryset = Product.all()

    if params.q is not None:
        queryset = queryset.filter(
            Q(name__icontains=params.q) | Q(description__icontains=params.q)
        )
    if params.category_id is not None:
        queryset = queryset.filter(category_id=params.category_id)
    if params.min_price is not None:
        queryset = queryset.filter(price__gte=params.min_price)
    if params.max_price is not None:
        queryset = queryset.filter(price__lte=params.max_price)

    total = await queryset.count()
    total_pages = ceil(total / params.page_size) if total else 0

    if params.page > max(total_pages, 1):
        raise HTTPException(status_code=422, detail="page exceeds available pages")

    primary_order = (
        params.sort_by
        if params.sort_direction == "asc"
        else f"-{params.sort_by}"
    )
    offset = (params.page - 1) * params.page_size

    rows = await queryset.order_by(primary_order, "id").offset(offset).limit(
        params.page_size
    ).values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    items = [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

    return ProductPage(
        items=items,
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=total_pages,
    )


@app.post(
    "/products/{product_id}/reservations",
    response_model=ReservationResult,
    status_code=201,
)
async def reserve_product(product_id: int, request: ReserveRequest) -> ReservationResult:
    product = await Product.get_or_none(id=product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="product not found")
    if product.stock < request.quantity:
        raise HTTPException(status_code=409, detail="insufficient stock")
    await asyncio.sleep(0.05)
    product.stock -= request.quantity
    await product.save(update_fields=["stock"])
    reservation = await Reservation.create(
        product=product,
        quantity=request.quantity,
    )
    return ReservationResult(
        reservation_id=reservation.id,
        product_id=product.id,
        quantity=reservation.quantity,
        remaining_stock=product.stock,
    )

What Should Match?

The product route remains unchanged. The reservation route reads stock before its pause without coordinating concurrent requests.

Build the concurrent inventory test

An ordinary request cannot reveal a timing bug. The new HTTPX test sends two reservations together against a product holding five units.

• Create tests/test_inventory.py in the Visual Studio Code Explorer sidebar.
• Add the imports plus the low-stock product helper by pasting this code:

import asyncio
from decimal import Decimal

import pytest

from app.models import Category, Product, Reservation

pytestmark = pytest.mark.anyio


async def create_low_stock_product():
    category = await Category.create(name="Limited")
    return await Product.create(
        name="Limited Keyboard",
        description="A deliberately scarce product",
        price=Decimal("99.00"),
        category=category,
        stock=5,
    )

What Does This Test Setup Create?

• pytestmark enables asynchronous tests throughout this file.
• create_low_stock_product() creates one product with exactly five available units.
• Reservation lets the test inspect the audit records created by the endpoint.

• Save tests/test_inventory.py.
• Confirm test_inventory.py appears inside the tests folder in the Visual Studio Code Explorer sidebar.

Test File Missing from the Explorer?

Confirm that test_inventory.py sits inside the existing tests folder. Check the filename for extra extensions such as .txt.

help me place my inventory test in the correct project folder.

• Add the concurrent reservation test below create_low_stock_product() by pasting this code:

async def test_concurrent_reservations_do_not_oversell(client):
    product = await create_low_stock_product()

    first, second = await asyncio.gather(
        client.post(
            f"/products/{product.id}/reservations",
            json={"quantity": 3},
        ),
        client.post(
            f"/products/{product.id}/reservations",
            json={"quantity": 3},
        ),
    )

    assert sorted([first.status_code, second.status_code]) == [201, 409]

    refreshed = await Product.get_or_none(id=product.id)
    quantities = await Reservation.all().values_list("quantity", flat=True)

    assert refreshed is not None
    assert refreshed.stock == 2
    assert quantities == [3]
    assert sum(quantities) + refreshed.stock == 5

What Does the Concurrency Test Prove?

• asyncio.gather() launches both three-unit requests concurrently.
• The status assertion requires one success response plus one insufficient-stock response.
• The database assertions require one reservation with two units remaining.
• The final assertion accounts for all five starting units.

• Save tests/test_inventory.py.

Before you run this, predict how two requests might behave when both read stock before either saves.

• Expose the race by running this command in the PowerShell session from earlier:

python -m pytest tests/test_inventory.py -q

What Is This Command Testing?

The command runs only the new inventory tests with concise pytest output. The application still uses the deliberate scheduling pause.

The concurrency test fails because both requests can return 201. Two reservations can be approved from the same five-unit snapshot.

This Failure Is Intentional

The assertion expects one 201 plus one 409. The failing run proves that request validation alone cannot coordinate access to shared stock.

Cover the basic error paths

The race test isolates the timing problem. A second test confirms that ordinary missing-product plus invalid-quantity behavior still works.

• Add the error-path test below the concurrency test by pasting this code:

async def test_reservation_errors(client):
    product = await create_low_stock_product()

    missing = await client.post(
        "/products/999999/reservations",
        json={"quantity": 1},
    )
    insufficient = await client.post(
        f"/products/{product.id}/reservations",
        json={"quantity": 6},
    )
    invalid = await client.post(
        f"/products/{product.id}/reservations",
        json={"quantity": 0},
    )

    assert missing.status_code == 404
    assert insufficient.status_code == 409
    assert invalid.status_code == 422

What Do These Checks Cover?

• A missing product must return 404.
• A request larger than the available stock must return 409.
• A quantity below the schema minimum must return 422.

• Save tests/test_inventory.py.

Before you run the completed file, predict whether the basic error test can pass while the concurrency test still fails.

• Verify both inventory behaviors by running:

python -m pytest tests/test_inventory.py -q

What Should This Final Run Reveal?

The basic error test passes. The concurrent reservation test still fails because both requests can approve three units from the same starting stock.

You have reproduced the race reliably. The failing test now defines the exact safe behavior the next implementation must satisfy.

Concurrency Test Passes Unexpectedly?

Confirm that await asyncio.sleep(0.05) appears after the stock check in reserve_product(). Confirm that the test sends both requests through one asyncio.gather() call.

help me verify that my test reliably reproduces the reservation race.

✔️ I Reproduced the Race

You now have repeatable evidence that the read-check-write reservation can oversell stock.

ⓧ I'd Like to Double Check the Full Code

Compare your complete tests/test_inventory.py with this version.

import asyncio
from decimal import Decimal

import pytest

from app.models import Category, Product, Reservation

pytestmark = pytest.mark.anyio


async def create_low_stock_product():
    category = await Category.create(name="Limited")
    return await Product.create(
        name="Limited Keyboard",
        description="A deliberately scarce product",
        price=Decimal("99.00"),
        category=category,
        stock=5,
    )


async def test_concurrent_reservations_do_not_oversell(client):
    product = await create_low_stock_product()

    first, second = await asyncio.gather(
        client.post(
            f"/products/{product.id}/reservations",
            json={"quantity": 3},
        ),
        client.post(
            f"/products/{product.id}/reservations",
            json={"quantity": 3},
        ),
    )

    assert sorted([first.status_code, second.status_code]) == [201, 409]

    refreshed = await Product.get_or_none(id=product.id)
    quantities = await Reservation.all().values_list("quantity", flat=True)

    assert refreshed is not None
    assert refreshed.stock == 2
    assert quantities == [3]
    assert sum(quantities) + refreshed.stock == 5


async def test_reservation_errors(client):
    product = await create_low_stock_product()

    missing = await client.post(
        "/products/999999/reservations",
        json={"quantity": 1},
    )
    insufficient = await client.post(
        f"/products/{product.id}/reservations",
        json={"quantity": 6},
    )
    invalid = await client.post(
        f"/products/{product.id}/reservations",
        json={"quantity": 0},
    )

    assert missing.status_code == 404
    assert insufficient.status_code == 409
    assert invalid.status_code == 422

What Should Match?

The file contains one low-stock fixture plus two tests. The concurrency test deliberately demands the safe result that the current route cannot provide.

You have turned a hidden timing risk into a repeatable failing test. Next, you will protect the stock update plus reservation record as one concurrency-safe operation.

Lock Inventory Reservations

Your failing concurrency test exposed the exact overselling path. Both buyers can read stock five before either save becomes visible.

Your FastAPI route now uses Tortoise ORM to create a transaction. PostgreSQL then applies a row-level lock to the selected product.

This keeps the stock update plus the reservation insert on one connection. The failed test becomes proof that competing requests cannot oversell the product.

In this step, get ready to:

• Open one transaction around each reservation attempt.
• Lock the selected product row before checking its stock.
• Run the complete test suite to prove only one competing request succeeds.

Connect the route to a transaction

The transaction defines one boundary for reading stock plus recording a reservation. Every database operation inside that boundary needs the same connection.

Why use a row lock?

A conditional update can protect one stock value. This project also commits a reservation record in the same unit of work.

A row lock makes the check-lock-update-commit sequence visible. Optimistic locking would add version fields plus retry logic outside this project's scope.

• Switch back to app/main.py in Visual Studio Code.
• Select the import section from the first line through the closing parenthesis after ReserveRequest.
• Replace the selected imports with this complete import section:

from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q
from tortoise.transactions import in_transaction

from app.config import TORTOISE_ORM
from app.models import Product, Reservation
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
    ReservationResult,
    ReserveRequest,
)

What do these imports change?

• The existing imports continue to support the application lifecycle plus the product query route.
• The new in_transaction import provides a connection-bound transaction for each reservation attempt.
• The obsolete asyncio import disappears because the artificial race delay is no longer needed.

Lock and update the product row

The unsafe route checks stock from an unlocked snapshot. The safe route makes a competing request wait before it reads that product again.

This replacement is the fiddliest edit in the step because the existing route decorator must stay in place. You only replace the function beneath it.

The existing @app.post decorator already defines the reservation path plus its response model.

• Select the code from async def reserve_product( through the final closing parenthesis of ReservationResult(.
• Replace the selected function with this transactional version:

async def reserve_product(
    product_id: int,
    request: ReserveRequest,
) -> ReservationResult:
    async with in_transaction() as connection:
        product = await Product.select_for_update(
            using_db=connection
        ).get_or_none(id=product_id)

        if product is None:
            raise HTTPException(status_code=404, detail="product not found")

        if product.stock < request.quantity:
            raise HTTPException(status_code=409, detail="insufficient stock")

        product.stock -= request.quantity
        await product.save(using_db=connection, update_fields=["stock"])

        reservation = await Reservation.create(
            using_db=connection,
            product=product,
            quantity=request.quantity,
        )

        return ReservationResult(
            reservation_id=reservation.id,
            product_id=product.id,
            quantity=reservation.quantity,
            remaining_stock=product.stock,
        )

How does the lock prevent overselling?

• The in_transaction() context opens one transaction connection for the reservation attempt.
• The select_for_update query locks the matching product row before the stock check runs.
• A competing request waits for the first transaction to finish. It then reads the committed remaining stock.
• The stock save plus Reservation.create use the same connection. They commit together as one unit.
• The route preserves 404 for a missing product. It preserves 409 for insufficient stock.
• The existing ReserveRequest schema continues to return 422 for invalid quantities.

• Save app/main.py.

✔️ Awesome, I've got everything!

Your reservation route now locks one product row inside a shared transaction.

ⓧ I'd like to double check the full code

• Compare your complete app/main.py with this reference file.

from contextlib import asynccontextmanager
from math import ceil
from typing import Annotated

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.expressions import Q
from tortoise.transactions import in_transaction

from app.config import TORTOISE_ORM
from app.models import Product, Reservation
from app.schemas import (
    ProductOut,
    ProductPage,
    ProductQuery,
    ReservationResult,
    ReserveRequest,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with RegisterTortoise(app, config=TORTOISE_ORM):
        yield


app = FastAPI(
    title="Product Inventory API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/products", response_model=ProductPage)
async def list_products(
    params: Annotated[ProductQuery, Query()],
) -> ProductPage:
    queryset = Product.all()

    if params.q is not None:
        queryset = queryset.filter(
            Q(name__icontains=params.q) | Q(description__icontains=params.q)
        )
    if params.category_id is not None:
        queryset = queryset.filter(category_id=params.category_id)
    if params.min_price is not None:
        queryset = queryset.filter(price__gte=params.min_price)
    if params.max_price is not None:
        queryset = queryset.filter(price__lte=params.max_price)

    total = await queryset.count()
    total_pages = ceil(total / params.page_size) if total else 0

    if params.page > max(total_pages, 1):
        raise HTTPException(status_code=422, detail="page exceeds available pages")

    primary_order = (
        params.sort_by
        if params.sort_direction == "asc"
        else f"-{params.sort_by}"
    )
    offset = (params.page - 1) * params.page_size

    rows = await queryset.order_by(primary_order, "id").offset(offset).limit(
        params.page_size
    ).values(
        "id",
        "name",
        "description",
        "price",
        "category_id",
        "stock",
        "created_at",
    )

    items = [
        ProductOut(
            id=row["id"],
            name=row["name"],
            description=row["description"],
            price=float(row["price"]),
            category_id=row["category_id"],
            stock=row["stock"],
            created_at=row["created_at"],
        )
        for row in rows
    ]

    return ProductPage(
        items=items,
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=total_pages,
    )


@app.post(
    "/products/{product_id}/reservations",
    response_model=ReservationResult,
    status_code=201,
)
async def reserve_product(
    product_id: int,
    request: ReserveRequest,
) -> ReservationResult:
    async with in_transaction() as connection:
        product = await Product.select_for_update(
            using_db=connection
        ).get_or_none(id=product_id)

        if product is None:
            raise HTTPException(status_code=404, detail="product not found")

        if product.stock < request.quantity:
            raise HTTPException(status_code=409, detail="insufficient stock")

        product.stock -= request.quantity
        await product.save(using_db=connection, update_fields=["stock"])

        reservation = await Reservation.create(
            using_db=connection,
            product=product,
            quantity=request.quantity,
        )

        return ReservationResult(
            reservation_id=reservation.id,
            product_id=product.id,
            quantity=reservation.quantity,
            remaining_stock=product.stock,
        )

What should match?

The final file has no asyncio import. The reservation function uses one transaction connection for the lock plus both writes.

Prove the reservation is safe

The complete suite checks the existing catalog behavior plus every reservation outcome. Its concurrency test rereads the product and reservation rows after both requests finish.

Before you run the suite, consider whether both reservation requests can still return success.

• Run every catalog plus inventory test from the activated PowerShell environment with this command:

python -m pytest -q

What does this check?

The command runs the complete test suite with compact output. It covers product queries plus reservation errors.

The concurrent test sends two three-unit requests against stock five. Its database assertions prove that only one reservation commits.

You should see every catalog plus inventory test pass. The concurrency test now proves the sorted statuses are [201, 409].

The same passing test proves the final stock is 2. It also proves the only reservation quantity is [3].

Still seeing the concurrency test fail?

• Confirm app/main.py imports in_transaction.
• Confirm select_for_update receives using_db=connection.
• Confirm both product.save plus Reservation.create receive the same connection.
• Confirm the old asyncio.sleep(0.05) line is gone.

help me debug the remaining reservation race in my transactional FastAPI route.

• Return to tests/test_inventory.py in Visual Studio Code.
• Confirm the status assertion expects [201, 409].
• Confirm the refreshed product assertion expects stock 2.
• Confirm the reservation assertion expects quantities [3].
• Confirm the conservation assertion adds the successful quantity to the remaining stock for a total of 5.

That is the race closed. Your API now preserves inventory when competing buyers arrive together.

Secret mission

Add a Category Inventory Summary

Your product routes answer questions about individual records. In this mission, you will add a grouped category summary that reports product counts and total stock directly from the database.

Clean Up Your Resources

Clean Up Your Resources

The project runs entirely on your Windows machine with no service charges. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

• The local PostgreSQL database product_inventory with its catalog, reservation, and migration data.
• The project folder currently open in Visual Studio Code with its virtual environment, source files, seed script, and tests.
• The Uvicorn server process if it is still running.

Keep everything running

No action is needed. Choose this if you are still actively building or want the API available for another demonstration.

• Leave the Uvicorn server running if you want the API available now.
• Leave product_inventory intact so the API keeps its catalog data.
• Keep the project folder so its application code and tests remain available.
• Keep seed.py for restoring the deterministic fifty-product catalog after test runs.

Pause - I'll come back to this later

This option frees the terminal while preserving the database and project files for your next session.

• Switch back to the activated PowerShell terminal.
• Press Ctrl+C to stop Uvicorn if it is still running.
• Leave the product_inventory database intact.
• Keep the project folder unchanged.

Delete - I don't want to use this again

This option removes the local database and project folder so you can start fresh.

Deletion is permanent. The Keep and Pause options remain available if you want another demo later.

• Switch back to the activated PowerShell terminal.
• Press Ctrl+C to stop Uvicorn if it is still running.
• Delete the local database by running this command:

dropdb product_inventory

What does this command do?

The dropdb command permanently removes the local product_inventory database.

Its tables and rows disappear together. Your files remain until you delete the project folder.

• Confirm PowerShell returns to its prompt without reporting a database deletion error.

Database Not Deleted?

• Make sure Uvicorn has stopped before you retry the database deletion.
• Check that your PostgreSQL role has permission to delete product_inventory.
• Help me troubleshoot the database deletion.

• Press the Windows key to open search.
• Type File Explorer into the search field.
• Press Enter to open File Explorer.
• Navigate to the location containing the project folder currently open in Visual Studio Code.
• Close the Visual Studio Code window.
• Right-click the project folder that contains requirements.txt.
• Select Delete.
• Confirm the project folder no longer appears in File Explorer.

Your local database and project folder are now removed. The API can no longer run from these deleted resources.

Nice Work!

Nice Work!

You made it through the full inventory workflow! Your local FastAPI backend now supports flexible catalog browsing while keeping inventory correct under concurrent requests.

You've learned how to:

• Build a seeded PostgreSQL catalog containing 50 products. Shape catalog requests through Pydantic validation. Return search filters. Keep sorting deterministic across pages. Include pagination metadata in every response.
• Reproduce a read-check-write race with simultaneous requests. Prove the overselling bug through an asynchronous HTTPX test.
• Protect inventory with a Tortoise ORM transaction. Serialize competing reservations with a row-level lock. Keep each stock update with its reservation audit record. Prove stock conservation through concurrent pytest coverage.
• Secret Mission: Added a database-side category summary that reports product counts plus total available stock for each category.

Ready to quiz yourself?