Product Inventory & Order API

Introduction

30 Second Summary

Every time you add something to a shopping cart online, a backend API is checking stock levels, reserving inventory, and recording your order across multiple database tables. If any of those steps fail halfway through, the system needs to undo everything cleanly.

In this project, you will build a Product Inventory & Order API that manages products with stock levels and processes customer orders using database transactions. You will implement cursor-based pagination, filtering, and sales analytics endpoints.

What You'll Build

A FastAPI backend connected to PostgreSQL that handles product listings, transactional order placement, and sales aggregation queries.

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

• A multi-table PostgreSQL schema with products, orders, and order items linked by foreign keys, managed through Alembic migrations.
• Product listing endpoints with cursor-based pagination, category filtering, price range filtering, and sort options.
• An order placement endpoint that uses database transactions to atomically validate stock and decrement inventory, preventing overselling.
• Secret Mission: Add JWT authentication with role-based access control so only admins can view analytics.

Are there any prerequisites?

This project assumes you have basic experience building a CRUD API, familiarity with Python fundamentals, and an understanding of SQL basics like SELECT and INSERT statements. You should also have Python 3 and PostgreSQL installed on your machine.

Before We Start

Every production API starts with a solid foundation. Before writing any business logic, you need a Python environment that isolates your dependencies and a database ready to accept connections.

In this step, you'll scaffold the project directory, install the packages your API needs, create a PostgreSQL database, and lay out the file structure.

In this step, get ready to:

• Set up a Python virtual environment and install project dependencies.
• Create the PostgreSQL database.
• Initialize the project file structure.

Set up the project and install dependencies

A virtual environment keeps your project's packages separate from your system Python. This prevents version conflicts between projects.

• Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
• Type Terminal and press Enter to open it.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create your project folder and move into it by running these commands:

mkdir inventory-api
cd inventory-api

• Create a Python virtual environment by running:

python3 -m venv venv

• Activate the virtual environment.

macOS

• Run this command to activate:

source venv/bin/activate

Windows

• Run this command to activate:

venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms you're working inside the virtual environment.

Don't see (venv) in your prompt?

Make sure you ran the activate command from inside the inventory-api folder. Run ls and confirm you see a venv folder listed.

On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned first, then try activating again.

Help me activate my Python virtual environment.

• Install all the project dependencies by running:

pip install "fastapi[standard]==0.138.0" sqlalchemy==2.0.51 psycopg2-binary==2.9.12 alembic==1.18.4 pydantic-settings

You should see output ending with a Successfully installed message listing all the packages.

What are these packages?

• fastapi[standard] is the web framework for building your API. The [standard] extra includes the development server.
• sqlalchemy is the ORM that maps Python classes to database tables.
• psycopg2-binary is the PostgreSQL driver that lets Python talk to your database.
• alembic manages database schema migrations.
• pydantic-settings handles configuration and environment variables.

• Verify the FastAPI CLI is available by running:

fastapi --version

You should see a version number printed in your terminal. This confirms the CLI was installed correctly.

Command not found?

Make sure your virtual environment is active. You should see (venv) in your prompt. If not, re-run the activate command from above.

If you still see an error, try reinstalling with pip install "fastapi[standard]==0.138.0" and check the output for errors.

Help me troubleshoot my FastAPI installation.

Create the PostgreSQL database

Your API needs a dedicated database to store products, orders, and order items. PostgreSQL's createdb utility creates one from the command line.

• Create the database by running:

createdb inventory_api

• Verify the database is accessible by running:

psql inventory_api -c "SELECT 1;"

You should see a small table with a single row containing the value 1. This confirms PostgreSQL is running and your database exists.

Seeing a connection error?

If you see a "connection refused" error, your PostgreSQL server might not be running. On macOS, start it with brew services start postgresql. On Windows, check that the PostgreSQL service is running in the Services app.

If you see "role does not exist", your system user may not have a matching PostgreSQL role. Run createuser -s $(whoami) to create one.

Help me connect to my PostgreSQL database.

Initialize the project file structure

Your API will be organized into separate files for database configuration, models, schemas, and route handlers. Setting up this structure now keeps the project clean as it grows.

• Create the router directory and all project files by running these commands:

mkdir routers
touch main.py database.py models.py schemas.py
touch routers/__init__.py routers/products.py routers/orders.py routers/analytics.py

What is this file structure?

• main.py is the entry point that creates and configures the FastAPI app.
• database.py holds the database connection and session configuration.
• models.py defines your SQLAlchemy table models.
• schemas.py defines the request and response shapes using Pydantic.
• The routers/ directory separates endpoint logic by domain (products, orders, analytics).

• Confirm all files were created by running:

ls
ls routers/

The first ls should show main.py, database.py, models.py, schemas.py, routers, and venv. The second should show __init__.py, products.py, orders.py, and analytics.py.

Missing files in the output?

Make sure you ran both commands from inside the inventory-api folder. Run pwd to check your current directory.

If routers/ is missing, the mkdir routers command may have failed silently. Re-run it and then re-run the touch routers/... commands.

Help me fix my project file structure.

Your project skeleton is ready. Next up, you'll define the database models and configure Alembic to manage your schema migrations.

Set Up the Project

Your project directory, virtual environment, dependencies, and database are all in place. Now it's time to write the foundational code that connects everything together.

Every FastAPI application needs two things to talk to a database: a connection layer that manages how Python communicates with PostgreSQL, and an entry point that wires routes and error handling together. In this step, you'll write both.

In this step, get ready to:

• Write the database connection module with SQLAlchemy.
• Set up the FastAPI application with routing and error handling.
• Verify the server starts and the health endpoint responds.

Write the database connection module

The database.py module is the bridge between your Python code and PostgreSQL. It defines how connections are created, how sessions are managed, and provides a base class that all your future models will inherit from.

• In VS Code, open database.py from the Explorer sidebar.
• Replace the placeholder comment with the following code:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# Connection string pointing to your local PostgreSQL database
DATABASE_URL = "postgresql://localhost/inventory_api"

# Engine handles the actual database connections
engine = create_engine(DATABASE_URL, echo=True)
# SessionLocal is a factory that produces new database sessions
SessionLocal = sessionmaker(bind=engine)


# Base class that all SQLAlchemy models will inherit from
class Base(DeclarativeBase):
    pass


def create_tables():
    # Creates all tables defined by models inheriting from Base
    Base.metadata.create_all(bind=engine)


def get_db():
    # Dependency injection helper: opens a session, yields it, then closes
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

What does this code do?

• DATABASE_URL tells SQLAlchemy where your PostgreSQL database lives. Since it's running locally with no password, the connection string is short.
• echo=True makes the engine print every SQL statement it runs to the terminal. This is invaluable for debugging during development.
• get_db() is a generator that FastAPI will use as a dependency. It opens a database session for each request, then guarantees the session closes afterwards.

• Save database.py.

You won't see a change yet. The next file (main.py) imports from this module and actually starts the server.

Seeing a red underline on the import?

If VS Code shows a warning on from sqlalchemy import create_engine, your editor might not be using the virtual environment's Python interpreter.

• Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and select Python: Select Interpreter.
• Choose the interpreter inside your venv folder.

Help me fix my Python interpreter in VS Code

✔️ Awesome, I've got everything!

Great. Make sure you've saved database.py before moving on.

ⓧ I'd like to double check the full code

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

DATABASE_URL = "postgresql://localhost/inventory_api"

engine = create_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


def create_tables():
    Base.metadata.create_all(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Write the application entry point

The main.py file is where your FastAPI app is created and configured. It wires together your database startup, route modules, and error handling into a single application.

• Open main.py in VS Code.
• Replace the placeholder comment with the following code to set up the app foundation:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager

from database import create_tables
from routers import products, orders, analytics


# Lifespan runs create_tables() once when the server starts
@asynccontextmanager
async def lifespan(app: FastAPI):
    create_tables()
    yield


# Create the FastAPI app and attach the lifespan hook
app = FastAPI(title="Inventory & Order API", lifespan=lifespan)

# Register route modules so their endpoints become part of the app
app.include_router(products.router)
app.include_router(orders.router)
app.include_router(analytics.router)

What does this code do?

• The lifespan function is FastAPI's startup hook. Code before yield runs once when the server starts, so your database tables are created automatically.
• include_router() attaches each route module to the app. This keeps your endpoints organized in separate files instead of one giant main.py.

• Now add the error handler and health endpoint below the router registrations:

# Catch ValueError exceptions and return a structured JSON error
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={
            "error": "bad_request",
            "detail": str(exc),
        },
    )


# Simple endpoint to verify the server is running
@app.get("/health")
def health_check():
    return {"status": "healthy"}

What does this code do?

• The exception handler catches any ValueError raised anywhere in your app and returns a consistent JSON structure. This prevents raw Python tracebacks from leaking to API consumers.
• The /health endpoint is a lightweight check that confirms your server is alive. You'll use it in a moment to verify everything works.

• Save main.py.

Before the server can start, the router files need to export a router object. You'll build these out fully in later steps, but they need minimal stubs right now.

• Open routers/products.py and replace the placeholder with this stub:

from fastapi import APIRouter

router = APIRouter(prefix="/products", tags=["products"])

• Open routers/orders.py and replace the placeholder with this stub:

from fastapi import APIRouter

router = APIRouter(prefix="/orders", tags=["orders"])

• Open routers/analytics.py and replace the placeholder with this stub:

from fastapi import APIRouter

router = APIRouter(prefix="/analytics", tags=["analytics"])

• Save all three router files.

Why stubs?

main.py imports products.router, orders.router, and analytics.router at startup. Without these stubs, Python would crash with an AttributeError because the modules wouldn't have a router attribute. You'll replace these stubs with full endpoint logic in the upcoming steps.

✔️ Awesome, I've got everything!

Great. Make sure you've saved all files before moving on to verification.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager

from database import create_tables
from routers import products, orders, analytics


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_tables()
    yield


app = FastAPI(title="Inventory & Order API", lifespan=lifespan)

app.include_router(products.router)
app.include_router(orders.router)
app.include_router(analytics.router)


@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={
            "error": "bad_request",
            "detail": str(exc),
        },
    )


@app.get("/health")
def health_check():
    return {"status": "healthy"}

Verify the health endpoint

With both files saved, it's time to confirm the server starts and responds to requests.

• Start the development server by running this command in your terminal:

fastapi dev main.py

You should see output indicating the server is running at http://127.0.0.1:8000. You'll also see SQL statements logged (from echo=True) as the server creates tables on startup.

• Open a second terminal tab or window (keep the server running in the first one).
• Test the health endpoint by running:

curl http://127.0.0.1:8000/health

You should see this response:

{"status":"healthy"}

What about the auto-generated docs?

FastAPI automatically generates interactive API documentation. Open http://127.0.0.1:8000/docs in your browser to explore it. You'll see the /health endpoint listed along with the empty router groups for products, orders, and analytics.

Server won't start?

• If you see an AttributeError mentioning router, double check that all three router files (routers/products.py, routers/orders.py, routers/analytics.py) contain the stub code with router = APIRouter(...).
• If you see a connection error mentioning PostgreSQL, make sure the inventory_api database exists and PostgreSQL is running.
• If the curl command returns "Connection refused", make sure the server is still running in your first terminal tab.

Help me debug my FastAPI server startup error

Your foundation is solid. The database layer is connected, the app starts cleanly, and the health check proves everything is wired together. Next up, you'll define the data models and use Alembic to create the actual database tables.

Define Database Models and Run Migrations

Your API server is running, but it has no data to serve. Before you can build product listings or order placement, you need tables in PostgreSQL to hold your products, orders, and order line items.

In this step, you will define the database schema using SQLAlchemy models, set up Alembic for migration management, and seed the database with sample products so there is data to query.

In this step, get ready to:

• Define SQLAlchemy models for products, orders, and order items with foreign key relationships.
• Configure Alembic and generate your first database migration.
• Apply the migration and seed the database with sample products.

Define the SQLAlchemy models

Your API needs three tables that relate to each other. A product has stock, an order belongs to a customer, and order items link specific products to a specific order.

• Open models.py in VS Code.
• Replace the placeholder comment with the following imports and the Product model:

from datetime import datetime, timezone
from sqlalchemy import Integer, String, Boolean, DateTime, ForeignKey, Index
from sqlalchemy.orm import Mapped, mapped_column, relationship

from database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str] = mapped_column(String(1000), nullable=True)
    category: Mapped[str] = mapped_column(String(100), nullable=False)
    price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
    stock_quantity: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    # Relationship to order items that reference this product
    order_items: Mapped[list["OrderItem"]] = relationship(back_populates="product")

    # Database indexes for faster filtering
    __table_args__ = (
        Index("ix_products_category", "category"),
        Index("ix_products_price_cents", "price_cents"),
    )

What does this code do?

Each Mapped[type] declaration tells SQLAlchemy the column type using Python type hints. This is the SQLAlchemy 2.0 declarative style.

Prices are stored as integer cents (price_cents) to avoid floating-point precision issues. The __table_args__ tuple adds database indexes on category and price_cents so filtering queries stay fast as the table grows.

• Add the Order and OrderItem models below the Product class:

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    customer_name: Mapped[str] = mapped_column(String(200), nullable=False)
    status: Mapped[str] = mapped_column(String(50), nullable=False, default="confirmed")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    # All line items belonging to this order
    items: Mapped[list["OrderItem"]] = relationship(back_populates="order")


class OrderItem(Base):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), nullable=False)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), nullable=False)
    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
    unit_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)

    # Relationships back to parent order and product
    order: Mapped["Order"] = relationship(back_populates="items")
    product: Mapped["Product"] = relationship(back_populates="order_items")

What are foreign keys doing here?

ForeignKey("orders.id") on order_id creates a database-level link between order_items and orders. PostgreSQL will reject any insert that references a non-existent order.

The relationship() calls let you traverse these links in Python. For example, given an order object, order.items returns all its line items without writing a manual JOIN.

• Save models.py.

Next, you need Pydantic schemas that control how data enters and leaves the API. These are separate from SQLAlchemy models because input validation and response shaping have different concerns than database storage.

• Open schemas.py and replace the placeholder with the following Pydantic models:

from pydantic import BaseModel
from datetime import datetime


# Response schema for a single product
class ProductOut(BaseModel):
    id: int
    name: str
    description: str | None
    category: str
    price_cents: int
    stock_quantity: int
    created_at: datetime

    model_config = {"from_attributes": True}


# Paginated product list with cursor
class ProductListResponse(BaseModel):
    data: list[ProductOut]
    next_cursor: int | None


# Input schema for a single order item
class OrderItemIn(BaseModel):
    product_id: int
    quantity: int


# Input schema for placing an order
class OrderIn(BaseModel):
    customer_name: str
    items: list[OrderItemIn]


# Response schema for an order item
class OrderItemOut(BaseModel):
    id: int
    product_id: int
    quantity: int
    unit_price_cents: int

    model_config = {"from_attributes": True}


# Response schema for a full order
class OrderOut(BaseModel):
    id: int
    customer_name: str
    status: str
    created_at: datetime
    items: list[OrderItemOut]

    model_config = {"from_attributes": True}


# Analytics response for top products
class TopProductOut(BaseModel):
    product_id: int
    product_name: str
    total_quantity_sold: int
    total_revenue_cents: int


# Analytics response for revenue summary
class RevenueSummaryOut(BaseModel):
    total_revenue_cents: int
    total_orders: int
    average_order_value_cents: int

What does model_config do?

model_config = {"from_attributes": True} tells Pydantic to read data from SQLAlchemy model attributes (like product.name) instead of requiring a dictionary. Without it, FastAPI could not serialize your database objects into JSON responses.

• Save schemas.py.

Seeing import errors in your editor?

If VS Code shows red underlines on from database import Base, make sure your terminal's virtual environment is activated and your Python interpreter in VS Code points to the .venv folder.

help me fix import errors in my SQLAlchemy models file

✔️ Awesome, I've got everything!

Great. Make sure both models.py and schemas.py are saved.

ⓧ I'd like to double check the full code

Here is the complete models.py:

from datetime import datetime, timezone
from sqlalchemy import Integer, String, Boolean, DateTime, ForeignKey, Index
from sqlalchemy.orm import Mapped, mapped_column, relationship

from database import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str] = mapped_column(String(1000), nullable=True)
    category: Mapped[str] = mapped_column(String(100), nullable=False)
    price_cents: Mapped[int] = mapped_column(Integer, nullable=False)
    stock_quantity: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    order_items: Mapped[list["OrderItem"]] = relationship(back_populates="product")

    __table_args__ = (
        Index("ix_products_category", "category"),
        Index("ix_products_price_cents", "price_cents"),
    )


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    customer_name: Mapped[str] = mapped_column(String(200), nullable=False)
    status: Mapped[str] = mapped_column(String(50), nullable=False, default="confirmed")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

    items: Mapped[list["OrderItem"]] = relationship(back_populates="order")


class OrderItem(Base):
    __tablename__ = "order_items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"), nullable=False)
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"), nullable=False)
    quantity: Mapped[int] = mapped_column(Integer, nullable=False)
    unit_price_cents: Mapped[int] = mapped_column(Integer, nullable=False)

    order: Mapped["Order"] = relationship(back_populates="items")
    product: Mapped["Product"] = relationship(back_populates="order_items")

Here is the complete schemas.py:

from pydantic import BaseModel
from datetime import datetime


class ProductOut(BaseModel):
    id: int
    name: str
    description: str | None
    category: str
    price_cents: int
    stock_quantity: int
    created_at: datetime

    model_config = {"from_attributes": True}


class ProductListResponse(BaseModel):
    data: list[ProductOut]
    next_cursor: int | None


class OrderItemIn(BaseModel):
    product_id: int
    quantity: int


class OrderIn(BaseModel):
    customer_name: str
    items: list[OrderItemIn]


class OrderItemOut(BaseModel):
    id: int
    product_id: int
    quantity: int
    unit_price_cents: int

    model_config = {"from_attributes": True}


class OrderOut(BaseModel):
    id: int
    customer_name: str
    status: str
    created_at: datetime
    items: list[OrderItemOut]

    model_config = {"from_attributes": True}


class TopProductOut(BaseModel):
    product_id: int
    product_name: str
    total_quantity_sold: int
    total_revenue_cents: int


class RevenueSummaryOut(BaseModel):
    total_revenue_cents: int
    total_orders: int
    average_order_value_cents: int

Configure Alembic for database migrations

Alembic tracks what your database schema looks like and generates migration scripts when your models change. This is how production teams manage schema changes without manually writing SQL.

• Initialize the Alembic migration environment by running this command:

alembic init alembic

What did this create?

This command creates an alembic/ directory (with env.py and a versions/ folder for migration scripts) and an alembic.ini config file in your project root.

• Confirm the files were created by running:

ls alembic

You should see env.py, README, script.py.mako, and a versions directory listed.

Now you need to point Alembic at your database and tell it about your models.

• Open alembic.ini in VS Code.
• Find the line that starts with sqlalchemy.url:

sqlalchemy.url = driver://user:pass@localhost/dbname

• Replace it with your project's database URL:

sqlalchemy.url = postgresql://localhost/inventory_api

• Save alembic.ini.
• Open alembic/env.py.
• Near the top of the file, find the commented-out section that ends with this line:

target_metadata = None

• Replace target_metadata = None with these two lines:

import models  # noqa: F401 - registers models with Base
from database import Base
target_metadata = Base.metadata

Why import models here?

Importing models forces Python to execute the model class definitions, which registers the tables with Base.metadata. Without this import, Alembic would see an empty metadata object and generate a blank migration.

• Save alembic/env.py.
• Generate the initial migration by running:

alembic revision --autogenerate -m "initial tables"

You should see output indicating a new migration file was generated inside alembic/versions/.

Seeing a ModuleNotFoundError?

If Python cannot find the models or database module, make sure you are running the command from your project's root folder (the one containing models.py and database.py). Also confirm your virtual environment is activated.

help me fix ModuleNotFoundError when running alembic autogenerate

Apply the migration and seed the database

The migration file describes what tables to create, but it has not touched the database yet. You need to apply it, then insert sample data so you have something to query.

• Apply all pending migrations by running:

alembic upgrade head

You should see output showing the migration being applied, ending with a confirmation that the upgrade completed.

What does upgrade head mean?

head refers to the latest migration revision. Running alembic upgrade head applies every migration that the database has not seen yet, bringing the schema up to date with your models.

• Create a seed script to populate the database with sample products by creating a new file called seed.py with the following content:

from database import SessionLocal
from models import Product


def seed():
    db = SessionLocal()
    products = [
        Product(name="Mechanical Keyboard", description="Cherry MX Brown switches", category="electronics", price_cents=8999, stock_quantity=50),
        Product(name="USB-C Hub", description="7-port hub with HDMI", category="electronics", price_cents=3499, stock_quantity=120),
        Product(name="Standing Desk Mat", description="Anti-fatigue comfort mat", category="furniture", price_cents=4599, stock_quantity=30),
        Product(name="Monitor Arm", description="Dual monitor mount", category="furniture", price_cents=12999, stock_quantity=25),
        Product(name="Webcam HD", description="1080p with auto-focus", category="electronics", price_cents=5999, stock_quantity=80),
        Product(name="Desk Lamp", description="LED with adjustable brightness", category="furniture", price_cents=2999, stock_quantity=60),
        Product(name="Noise Cancelling Headphones", description="Over-ear wireless", category="electronics", price_cents=24999, stock_quantity=15),
        Product(name="Ergonomic Mouse", description="Vertical design", category="electronics", price_cents=4999, stock_quantity=0),
    ]
    db.add_all(products)
    db.commit()
    db.close()
    print(f"Seeded {len(products)} products.")


if __name__ == "__main__":
    seed()

What does this seed script do?

It creates 8 sample products across two categories: electronics and furniture. One product (Ergonomic Mouse) intentionally has stock_quantity=0 so you can test out-of-stock filtering later.

• Save seed.py, then run it:

python seed.py

You should see Seeded 8 products. printed in the terminal.

Seeing a connection error?

Make sure your PostgreSQL server is running and the inventory_api database exists. If you skipped the database creation step earlier, run createdb inventory_api first, then try alembic upgrade head and python seed.py again.

help me fix a PostgreSQL connection error when running my seed script

Your database schema is live and loaded with sample data. Next, you will build the product listing endpoint with cursor-based pagination and filtering.

Build Product Endpoints with Pagination and Filtering

Your database is seeded with products, and your schema models are ready to serialize them. Now you need the API layer that lets clients retrieve those products efficiently.

Real product catalogs contain thousands of items. Returning them all in one response would be slow and wasteful. In this step, you will build endpoints that support cursor-based pagination, filtering by category and price, and flexible sorting.

In this step, get ready to:

• Build a product listing endpoint with cursor-based pagination.
• Add category, price, and stock filtering plus sort options.
• Verify structured error responses and the JSON response envelope.

Build the product listing and detail endpoints

Most paginated APIs use offset pagination (skip N rows, take M). That approach breaks at scale because inserting or deleting rows shifts every page after the change.

Cursor-based pagination avoids this by remembering the last item the client saw and fetching the next batch after it. Stripe, GitHub, and Twitter all use this pattern.

Why cursor-based pagination instead of offset?

With offset pagination, page 3 might show a product that was already on page 2 if a new item was inserted. Cursor pagination anchors to an ID, so the client always moves forward without duplicates or gaps.

• In VS Code, open routers/products.py and replace the empty placeholder with the following imports and router setup:

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import select

from database import get_db
from models import Product
from schemas import ProductOut, ProductListResponse

router = APIRouter(prefix="/products", tags=["products"])

What does this code do?

• APIRouter groups related endpoints under a shared prefix (/products).
• Depends(get_db) is FastAPI's dependency injection. It gives each request handler a fresh database session and closes it automatically.
• The select import is SQLAlchemy's 2.0-style query builder.

• Below the router declaration, add the list_products endpoint by pasting this function:

@router.get("", response_model=ProductListResponse)
def list_products(
    cursor: int | None = Query(None, description="Product ID to start after"),
    limit: int = Query(20, ge=1, le=100),
    category: str | None = Query(None),
    min_price: int | None = Query(None, description="Minimum price in cents"),
    max_price: int | None = Query(None, description="Maximum price in cents"),
    in_stock: bool | None = Query(None),
    sort_by: str = Query("id", pattern="^(id|price_cents|name)$"),
    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
    db: Session = Depends(get_db),
):
    stmt = select(Product)

    if cursor is not None:
        stmt = stmt.where(Product.id > cursor)

    if category is not None:
        stmt = stmt.where(Product.category == category)
    if min_price is not None:
        stmt = stmt.where(Product.price_cents >= min_price)
    if max_price is not None:
        stmt = stmt.where(Product.price_cents <= max_price)
    if in_stock is True:
        stmt = stmt.where(Product.stock_quantity > 0)
    elif in_stock is False:
        stmt = stmt.where(Product.stock_quantity == 0)

    sort_column = getattr(Product, sort_by)
    if sort_order == "desc":
        stmt = stmt.order_by(sort_column.desc())
    else:
        stmt = stmt.order_by(sort_column.asc())

    stmt = stmt.limit(limit + 1)

    results = db.execute(stmt).scalars().all()

    if len(results) > limit:
        next_cursor = results[limit - 1].id
        products = results[:limit]
    else:
        next_cursor = None
        products = results

    return ProductListResponse(data=products, next_cursor=next_cursor)

What does this code do?

This function handles three responsibilities in a single query:

• Cursor pagination: The cursor parameter is a product ID. The query fetches only rows with id > cursor. To detect whether more pages exist, it fetches limit + 1 rows. If the extra row comes back, there are more results and next_cursor is set.
• Filtering: Each optional query parameter conditionally appends a .where() clause. Only non-None filters are applied.
• Sorting: sort_by maps to a model column via getattr. The pattern validator on Query restricts input to safe column names only.

• Below list_products, add the single-product endpoint:

@router.get("/{product_id}", response_model=ProductOut)
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = db.get(Product, product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

What does this code do?

• db.get(Product, product_id) fetches a product by primary key. This is the most efficient single-row lookup.
• If no product exists, HTTPException(status_code=404) returns a structured JSON error to the client.

• Save routers/products.py.
• Switch back to the terminal running fastapi dev main.py. The server auto-reloads on save.
• Open a second terminal and test the product detail endpoint by running:

curl http://127.0.0.1:8000/products/1

You should see a JSON object with the Mechanical Keyboard product, including its id, name, price_cents, and stock_quantity fields.

Seeing a connection error or 404?

• Make sure the fastapi dev main.py process is still running in your first terminal.
• If you see a module import error, confirm you saved routers/products.py and that the file is inside the routers/ directory.
• If the response says "Not Found", check that your seed data was loaded. Re-run python seed.py if needed.

Help me debug my products endpoint.

Test filtering and sorting with curl

The list_products function already accepts filtering and sorting parameters. Time to verify they work against your seeded data.

• Filter by category to see only electronics products by running:

curl "http://127.0.0.1:8000/products?category=electronics"

You should see 5 products (Mechanical Keyboard, USB-C Hub, Webcam HD, Noise Cancelling Headphones, and Ergonomic Mouse). Furniture products are excluded.

• Filter by price range to find mid-priced items by running:

curl "http://127.0.0.1:8000/products?min_price=4000&max_price=10000"

You should see products priced between 4000 and 10000 cents ($40 to $100).

• Sort by price descending to see the most expensive items first by running:

curl "http://127.0.0.1:8000/products?sort_by=price_cents&sort_order=desc"

Noise Cancelling Headphones (24999 cents) should appear first in the results.

• Filter for out-of-stock products by running:

curl "http://127.0.0.1:8000/products?in_stock=false"

You should see only the Ergonomic Mouse, which has a stock_quantity of 0.

Why use query parameters for filtering?

Query parameters keep the URL structure clean. Clients can combine any filters without needing separate endpoints for each combination.

The pattern validator on sort_by prevents injection by only allowing id, price_cents, or name as values.

Filters returning unexpected results?

• Make sure you wrap the URL in quotes. Without quotes, the shell interprets & as a background operator and cuts off your query string.
• Check that min_price and max_price are in cents (not dollars). 5000 means $50.00.

Help me debug my product filtering.

Verify the response envelope and error handling

Production APIs return responses in a consistent structure so clients can parse them predictably. Your API already has this built in through two mechanisms: the ProductListResponse schema and the ValueError exception handler in main.py.

• Test cursor pagination by limiting results to 3 items:

curl "http://127.0.0.1:8000/products?limit=3"

The response wraps products in a data array and includes a next_cursor field. Since you have 8 products and requested only 3, next_cursor will contain the ID of the last returned product.

• Use the next_cursor value to fetch the next page by running (replace 3 with whatever next_cursor value you received):

curl "http://127.0.0.1:8000/products?limit=3&cursor=3"

You should see the next batch of products. When you reach the last page, next_cursor will be null.

What is a response envelope?

A response envelope wraps the actual data inside a known structure. Your ProductListResponse always returns {"data": [...], "next_cursor": ...}. Clients know exactly where to find the products array and how to request the next page.

• Test the 404 error response by requesting a product that does not exist:

curl http://127.0.0.1:8000/products/999

You should see a structured error: {"detail": "Product not found"} with a 404 status code.

How does the exception handler work?

Your main.py has a value_error_handler that catches any ValueError raised in your code and returns it as a 400 response with a consistent {"error": "bad_request", "detail": "..."} shape.

FastAPI also handles HTTPException natively, returning the status_code and detail you specify. This gives you two layers: validation errors (400) and resource errors (404, 409).

• Open your browser to http://127.0.0.1:8000/docs to see the auto-generated OpenAPI documentation for your new endpoints.

You should see GET /products and GET /products/{product_id} listed with all their query parameters documented.

✔️ Awesome, I've got everything!

Great. Make sure you have saved routers/products.py and confirmed both endpoints respond in your terminal.

ⓧ I'd like to double check the full code

Here is the complete routers/products.py file:

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import select

from database import get_db
from models import Product
from schemas import ProductOut, ProductListResponse

router = APIRouter(prefix="/products", tags=["products"])


@router.get("", response_model=ProductListResponse)
def list_products(
    cursor: int | None = Query(None, description="Product ID to start after"),
    limit: int = Query(20, ge=1, le=100),
    category: str | None = Query(None),
    min_price: int | None = Query(None, description="Minimum price in cents"),
    max_price: int | None = Query(None, description="Maximum price in cents"),
    in_stock: bool | None = Query(None),
    sort_by: str = Query("id", pattern="^(id|price_cents|name)$"),
    sort_order: str = Query("asc", pattern="^(asc|desc)$"),
    db: Session = Depends(get_db),
):
    stmt = select(Product)

    if cursor is not None:
        stmt = stmt.where(Product.id > cursor)

    if category is not None:
        stmt = stmt.where(Product.category == category)
    if min_price is not None:
        stmt = stmt.where(Product.price_cents >= min_price)
    if max_price is not None:
        stmt = stmt.where(Product.price_cents <= max_price)
    if in_stock is True:
        stmt = stmt.where(Product.stock_quantity > 0)
    elif in_stock is False:
        stmt = stmt.where(Product.stock_quantity == 0)

    sort_column = getattr(Product, sort_by)
    if sort_order == "desc":
        stmt = stmt.order_by(sort_column.desc())
    else:
        stmt = stmt.order_by(sort_column.asc())

    stmt = stmt.limit(limit + 1)

    results = db.execute(stmt).scalars().all()

    if len(results) > limit:
        next_cursor = results[limit - 1].id
        products = results[:limit]
    else:
        next_cursor = None
        products = results

    return ProductListResponse(data=products, next_cursor=next_cursor)


@router.get("/{product_id}", response_model=ProductOut)
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = db.get(Product, product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

Your product endpoints are live and queryable. Next up, you will build the order placement endpoint that uses database transactions to prevent overselling.

Implement Order Placement with Transactions

Your product listing endpoint works. Customers can browse, filter, and paginate through the catalog.

Now you need the endpoint that actually makes money: placing orders. The tricky part is that two customers might try to buy the last item at the same time. Without proper safeguards, both orders succeed and you oversell inventory. Database transactions solve this by grouping reads and writes into an atomic unit that either fully succeeds or fully rolls back.

In this step, get ready to:

• Build a POST /orders endpoint that validates stock within a database transaction.
• Atomically decrement stock and create order records, rolling back on failure.
• Build a GET /orders/{id} endpoint and test the oversell protection with curl.

Build the POST /orders endpoint with transaction validation

The order endpoint needs to do multiple things in sequence: check that each requested product exists, verify enough stock is available, decrement the stock, and create the order records. If any check fails partway through, everything must revert.

• In VS Code, open routers/orders.py and replace the placeholder comment with the following code:

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import select

from database import get_db
from models import Product, Order, OrderItem
from schemas import OrderIn, OrderOut

router = APIRouter(prefix="/orders", tags=["orders"])


@router.post("", response_model=OrderOut, status_code=201)
def create_order(order_in: OrderIn, db: Session = Depends(get_db)):
    # Reject empty orders immediately
    if not order_in.items:
        raise HTTPException(status_code=400, detail="Order must contain at least one item")

    # Create the order shell (not yet committed)
    order = Order(customer_name=order_in.customer_name)
    db.add(order)

    for item in order_in.items:
        # Lock the product row to prevent concurrent modifications
        product = db.execute(
            select(Product).where(Product.id == item.product_id).with_for_update()
        ).scalar_one_or_none()

        # Validate product exists
        if product is None:
            db.rollback()
            raise HTTPException(
                status_code=404,
                detail=f"Product with id {item.product_id} not found",
            )

        # Validate sufficient stock
        if product.stock_quantity < item.quantity:
            db.rollback()
            raise HTTPException(
                status_code=409,
                detail=f"Insufficient stock for product '{product.name}'. "
                f"Available: {product.stock_quantity}, requested: {item.quantity}",
            )

        # Decrement stock atomically
        product.stock_quantity -= item.quantity

        # Create the line item record
        order_item = OrderItem(
            order=order,
            product_id=product.id,
            quantity=item.quantity,
            unit_price_cents=product.price_cents,
        )
        db.add(order_item)

    # All items validated and decremented - commit the whole batch
    db.commit()
    db.refresh(order)
    return order

What does this code do?

The create_order function processes an order inside a single database transaction. Everything between the session start and db.commit() is grouped together.

• .with_for_update() adds a SELECT FOR UPDATE lock on each product row. This prevents another request from reading or modifying that row until your transaction finishes.
• If a product doesn't exist or has insufficient stock, db.rollback() undoes all changes made so far in the transaction (including any stock decrements from earlier loop iterations).
• The 409 status code signals a conflict. The client's request was valid, but the current server state (low stock) prevents fulfilling it.
• db.commit() only runs if the entire loop completes without errors, making the operation atomic.

• Save routers/orders.py.

Seeing an import error on startup?

Make sure you're importing OrderIn and OrderOut from schemas. These were already defined in the previous step. Check that schemas.py is saved with the OrderIn, OrderItemIn, and OrderOut classes.

Help me debug an import error in my orders router.

Understand the atomic rollback behavior

The key insight in this endpoint is the word "atomic." Either every product's stock gets decremented and the order is created, or nothing changes at all.

Why with_for_update() instead of a simple read?

Imagine two customers ordering the last unit of the same product at the same time. Without row locking, both requests read stock_quantity = 1, both pass validation, and both decrement to 0. The product is now oversold.

SELECT FOR UPDATE is a pessimistic locking strategy. The first transaction locks the row. The second transaction waits until the first commits or rolls back. By the time the second request reads the row, stock is already decremented and the validation catches it correctly.

What happens on rollback?

When the loop encounters an invalid product or insufficient stock, db.rollback() undoes all pending changes in the session. If your order had 3 items and the 3rd fails validation, the stock decrements from items 1 and 2 are also reverted.

This is why the entire loop runs inside one transaction rather than committing after each item. Partial orders with inconsistent stock are impossible.

Build GET /orders/{id} and test the transaction

Customers need to retrieve their order after placing it. The GET endpoint returns the order with all its line items joined.

• In routers/orders.py, add the following function below create_order:

@router.get("/{order_id}", response_model=OrderOut)
def get_order(order_id: int, db: Session = Depends(get_db)):
    order = db.get(Order, order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    return order

What does this code do?

db.get(Order, order_id) fetches an order by its primary key. The OrderOut schema includes a nested items list, so SQLAlchemy automatically loads the related OrderItem records through the relationship defined in your model.

• Save routers/orders.py.
• Switch to the terminal where fastapi dev is running. It should auto-reload with no errors.

Now test the full transaction flow. Open a second terminal and place a successful order by running:

curl -X POST http://127.0.0.1:8000/orders \
  -H "Content-Type: application/json" \
  -d '{"customer_name": "Alice", "items": [{"product_id": 1, "quantity": 2}]}'

You should see a JSON response with status 201 containing the order ID, customer name, status confirmed, and the items array.

• Retrieve the order you just created by running:

curl http://127.0.0.1:8000/orders/1

You should see the same order with its nested items.

• Now test the oversell protection. The Ergonomic Mouse (product 8) has 0 stock. Attempt to order it by running:

curl -X POST http://127.0.0.1:8000/orders \
  -H "Content-Type: application/json" \
  -d '{"customer_name": "Bob", "items": [{"product_id": 8, "quantity": 1}]}'

You should see a 409 response with a message about insufficient stock. The transaction rolled back and no order was created.

Getting a 500 error instead of 409?

Check that you have db.rollback() before each raise HTTPException in the loop. Without the explicit rollback, SQLAlchemy may try to commit a dirty session when the next request arrives, causing unexpected errors.

Help me debug a 500 error on my POST /orders endpoint.

✔️ Awesome, I've got everything!

Great. Double check you've saved routers/orders.py and your dev server reloaded without errors.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import select

from database import get_db
from models import Product, Order, OrderItem
from schemas import OrderIn, OrderOut

router = APIRouter(prefix="/orders", tags=["orders"])


@router.post("", response_model=OrderOut, status_code=201)
def create_order(order_in: OrderIn, db: Session = Depends(get_db)):
    if not order_in.items:
        raise HTTPException(status_code=400, detail="Order must contain at least one item")

    order = Order(customer_name=order_in.customer_name)
    db.add(order)

    for item in order_in.items:
        product = db.execute(
            select(Product).where(Product.id == item.product_id).with_for_update()
        ).scalar_one_or_none()

        if product is None:
            db.rollback()
            raise HTTPException(
                status_code=404,
                detail=f"Product with id {item.product_id} not found",
            )

        if product.stock_quantity < item.quantity:
            db.rollback()
            raise HTTPException(
                status_code=409,
                detail=f"Insufficient stock for product '{product.name}'. "
                f"Available: {product.stock_quantity}, requested: {item.quantity}",
            )

        product.stock_quantity -= item.quantity

        order_item = OrderItem(
            order=order,
            product_id=product.id,
            quantity=item.quantity,
            unit_price_cents=product.price_cents,
        )
        db.add(order_item)

    db.commit()
    db.refresh(order)
    return order


@router.get("/{order_id}", response_model=OrderOut)
def get_order(order_id: int, db: Session = Depends(get_db)):
    order = db.get(Order, order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    return order

Your API now handles the most critical business operation safely. No matter how many concurrent requests arrive, stock can never go negative. Next up, you'll build analytics endpoints that aggregate this order data into sales insights.

Build Sales Aggregation Endpoints

Your order placement endpoint is working and the database is accumulating order data. The next question stakeholders always ask: which products are selling the most, and how much revenue is coming in?

In this step, you'll build analytics endpoints that answer these questions using SQL GROUP BY queries. These endpoints aggregate data across multiple tables and return computed summaries instead of raw rows.

In this step, get ready to:

• Build a top-products endpoint using GROUP BY with SUM.
• Build a revenue-summary endpoint with aggregate functions.
• Add date range filtering to both analytics endpoints.

Build the top-products endpoint

The top-products endpoint needs to join three tables: order_items, products, and orders. It groups results by product and sums the quantities sold across all orders.

• In routers/analytics.py, replace the empty placeholder with these imports and router setup:

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from sqlalchemy import select, func

from database import get_db
from models import Order, OrderItem, Product
from schemas import TopProductOut

# Set up the analytics router with /analytics prefix
router = APIRouter(prefix="/analytics", tags=["analytics"])

What does this code do?

These imports bring in the SQLAlchemy func module, which gives you access to SQL aggregate functions like SUM, COUNT, and COALESCE from Python.

The router is prefixed with /analytics so all endpoints in this file are grouped under that path.

• Below the router line, add the top_products endpoint function:

@router.get("/top-products", response_model=list[TopProductOut])
def top_products(
    limit: int = Query(10, ge=1, le=50),
    db: Session = Depends(get_db),
):
    # Build a query that joins order_items to products and orders
    stmt = (
        select(
            OrderItem.product_id,
            Product.name.label("product_name"),
            func.sum(OrderItem.quantity).label("total_quantity_sold"),
            func.sum(OrderItem.quantity * OrderItem.unit_price_cents).label(
                "total_revenue_cents"
            ),
        )
        .join(Product, OrderItem.product_id == Product.id)
        .join(Order, OrderItem.order_id == Order.id)
    )

    # Group by product and order by highest quantity sold
    stmt = (
        stmt.group_by(OrderItem.product_id, Product.name)
        .order_by(func.sum(OrderItem.quantity).desc())
        .limit(limit)
    )

    rows = db.execute(stmt).all()

    # Map each row to the response schema
    return [
        TopProductOut(
            product_id=row.product_id,
            product_name=row.product_name,
            total_quantity_sold=row.total_quantity_sold,
            total_revenue_cents=row.total_revenue_cents,
        )
        for row in rows
    ]

What does this code do?

This endpoint performs a three-table join starting from order_items. It joins to products for the product name and to orders so date filtering can be added later.

The GROUP BY clause collapses all order items for the same product into a single row. func.sum(OrderItem.quantity) totals how many units sold, and func.sum(OrderItem.quantity * OrderItem.unit_price_cents) calculates total revenue per product.

Results are ordered by total quantity descending, so the best sellers appear first.

• Save routers/analytics.py.
• Test the endpoint by running this in a new terminal:

curl http://127.0.0.1:8000/analytics/top-products

You should see a JSON array with your ordered products ranked by quantity sold. If you placed an order in the previous step, those products appear here with their totals.

Seeing an empty array?

An empty [] response means no orders exist in the database yet. Go back and place at least one order using the POST /orders endpoint from the previous step, then try again.

If you see a 500 error, check that your import of func from sqlalchemy is correct and that you saved the file.

help me debug my /analytics/top-products endpoint

Build the revenue-summary endpoint

The revenue summary answers a different question: across all orders, what's the total revenue, how many orders exist, and what's the average order value? This uses COALESCE, COUNT DISTINCT, and integer division.

• In routers/analytics.py, find the schemas import line:

from schemas import TopProductOut

• Add RevenueSummaryOut to the import so the line now looks like this:

from schemas import TopProductOut, RevenueSummaryOut

• Below the top_products function, add the revenue_summary endpoint:

@router.get("/revenue-summary", response_model=RevenueSummaryOut)
def revenue_summary(
    db: Session = Depends(get_db),
):
    # Calculate total revenue and order count in one query
    stmt = select(
        func.coalesce(func.sum(OrderItem.quantity * OrderItem.unit_price_cents), 0).label(
            "total_revenue_cents"
        ),
        func.count(func.distinct(Order.id)).label("total_orders"),
    ).join(Order, OrderItem.order_id == Order.id)

    row = db.execute(stmt).one()
    total_revenue = row.total_revenue_cents
    total_orders = row.total_orders
    # Integer division avoids floating point cents
    avg_value = total_revenue // total_orders if total_orders > 0 else 0

    return RevenueSummaryOut(
        total_revenue_cents=total_revenue,
        total_orders=total_orders,
        average_order_value_cents=avg_value,
    )

What does this code do?

This query runs a single SQL statement that computes two aggregates at once. func.coalesce(..., 0) ensures the result is 0 instead of NULL when no orders exist.

The func.count(func.distinct(Order.id)) counts unique orders rather than counting every order_item row. Without DISTINCT, an order with 3 items would be counted 3 times.

Average order value uses integer division (//) to keep the result in whole cents. The guard if total_orders > 0 prevents a division-by-zero error.

• Save routers/analytics.py.
• Test the revenue summary by running:

curl http://127.0.0.1:8000/analytics/revenue-summary

You should see a JSON object with total_revenue_cents, total_orders, and average_order_value_cents values based on the orders you placed earlier.

Getting a division by zero error?

This happens if you have no orders in the database. The if total_orders > 0 guard should prevent it. Double check that line is present and uses // (integer division), not / (float division).

help me fix my revenue-summary endpoint

Add date range filtering to both endpoints

Analytics are most useful when you can scope them to a time window. You'll add optional start_date and end_date query parameters that filter on the order's created_at timestamp.

• In routers/analytics.py, find the first import line:

from fastapi import APIRouter, Depends, Query

• Add a new import line above it so the top of the file starts with:

from datetime import date, datetime, timezone
from fastapi import APIRouter, Depends, Query

• Now replace the entire top_products function with this updated version that accepts date parameters:

@router.get("/top-products", response_model=list[TopProductOut])
def top_products(
    start_date: date | None = Query(None),
    end_date: date | None = Query(None),
    limit: int = Query(10, ge=1, le=50),
    db: Session = Depends(get_db),
):
    stmt = (
        select(
            OrderItem.product_id,
            Product.name.label("product_name"),
            func.sum(OrderItem.quantity).label("total_quantity_sold"),
            func.sum(OrderItem.quantity * OrderItem.unit_price_cents).label(
                "total_revenue_cents"
            ),
        )
        .join(Product, OrderItem.product_id == Product.id)
        .join(Order, OrderItem.order_id == Order.id)
    )

    # Filter by date range if provided
    if start_date is not None:
        stmt = stmt.where(
            Order.created_at >= datetime(start_date.year, start_date.month, start_date.day, tzinfo=timezone.utc)
        )
    if end_date is not None:
        end_dt = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59, tzinfo=timezone.utc)
        stmt = stmt.where(Order.created_at <= end_dt)

    stmt = (
        stmt.group_by(OrderItem.product_id, Product.name)
        .order_by(func.sum(OrderItem.quantity).desc())
        .limit(limit)
    )

    rows = db.execute(stmt).all()

    return [
        TopProductOut(
            product_id=row.product_id,
            product_name=row.product_name,
            total_quantity_sold=row.total_quantity_sold,
            total_revenue_cents=row.total_revenue_cents,
        )
        for row in rows
    ]

How does the date filtering work?

The start_date and end_date parameters use Python's date type. FastAPI automatically parses query strings like ?start_date=2026-01-01 into date objects.

The code converts the date to a timezone-aware datetime for comparison with the created_at timestamp column. The end_date gets set to 23:59:59 to include the entire final day.

• Replace the revenue_summary function with this updated version that also accepts date parameters:

@router.get("/revenue-summary", response_model=RevenueSummaryOut)
def revenue_summary(
    start_date: date | None = Query(None),
    end_date: date | None = Query(None),
    db: Session = Depends(get_db),
):
    stmt = select(
        func.coalesce(func.sum(OrderItem.quantity * OrderItem.unit_price_cents), 0).label(
            "total_revenue_cents"
        ),
        func.count(func.distinct(Order.id)).label("total_orders"),
    ).join(Order, OrderItem.order_id == Order.id)

    # Filter by date range if provided
    if start_date is not None:
        stmt = stmt.where(
            Order.created_at >= datetime(start_date.year, start_date.month, start_date.day, tzinfo=timezone.utc)
        )
    if end_date is not None:
        end_dt = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59, tzinfo=timezone.utc)
        stmt = stmt.where(Order.created_at <= end_dt)

    row = db.execute(stmt).one()
    total_revenue = row.total_revenue_cents
    total_orders = row.total_orders
    avg_value = total_revenue // total_orders if total_orders > 0 else 0

    return RevenueSummaryOut(
        total_revenue_cents=total_revenue,
        total_orders=total_orders,
        average_order_value_cents=avg_value,
    )

Why the same pattern in both functions?

Both endpoints filter on the same column (Order.created_at) using the same logic. This consistency means clients can use identical query parameters across analytics endpoints without learning different date formats per endpoint.

• Save routers/analytics.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before testing.

ⓧ I'd like to double check the full code

from datetime import date, datetime, timezone
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from sqlalchemy import select, func

from database import get_db
from models import Order, OrderItem, Product
from schemas import TopProductOut, RevenueSummaryOut

router = APIRouter(prefix="/analytics", tags=["analytics"])


@router.get("/top-products", response_model=list[TopProductOut])
def top_products(
    start_date: date | None = Query(None),
    end_date: date | None = Query(None),
    limit: int = Query(10, ge=1, le=50),
    db: Session = Depends(get_db),
):
    stmt = (
        select(
            OrderItem.product_id,
            Product.name.label("product_name"),
            func.sum(OrderItem.quantity).label("total_quantity_sold"),
            func.sum(OrderItem.quantity * OrderItem.unit_price_cents).label(
                "total_revenue_cents"
            ),
        )
        .join(Product, OrderItem.product_id == Product.id)
        .join(Order, OrderItem.order_id == Order.id)
    )

    if start_date is not None:
        stmt = stmt.where(
            Order.created_at >= datetime(start_date.year, start_date.month, start_date.day, tzinfo=timezone.utc)
        )
    if end_date is not None:
        end_dt = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59, tzinfo=timezone.utc)
        stmt = stmt.where(Order.created_at <= end_dt)

    stmt = (
        stmt.group_by(OrderItem.product_id, Product.name)
        .order_by(func.sum(OrderItem.quantity).desc())
        .limit(limit)
    )

    rows = db.execute(stmt).all()

    return [
        TopProductOut(
            product_id=row.product_id,
            product_name=row.product_name,
            total_quantity_sold=row.total_quantity_sold,
            total_revenue_cents=row.total_revenue_cents,
        )
        for row in rows
    ]


@router.get("/revenue-summary", response_model=RevenueSummaryOut)
def revenue_summary(
    start_date: date | None = Query(None),
    end_date: date | None = Query(None),
    db: Session = Depends(get_db),
):
    stmt = select(
        func.coalesce(func.sum(OrderItem.quantity * OrderItem.unit_price_cents), 0).label(
            "total_revenue_cents"
        ),
        func.count(func.distinct(Order.id)).label("total_orders"),
    ).join(Order, OrderItem.order_id == Order.id)

    if start_date is not None:
        stmt = stmt.where(
            Order.created_at >= datetime(start_date.year, start_date.month, start_date.day, tzinfo=timezone.utc)
        )
    if end_date is not None:
        end_dt = datetime(end_date.year, end_date.month, end_date.day, 23, 59, 59, tzinfo=timezone.utc)
        stmt = stmt.where(Order.created_at <= end_dt)

    row = db.execute(stmt).one()
    total_revenue = row.total_revenue_cents
    total_orders = row.total_orders
    avg_value = total_revenue // total_orders if total_orders > 0 else 0

    return RevenueSummaryOut(
        total_revenue_cents=total_revenue,
        total_orders=total_orders,
        average_order_value_cents=avg_value,
    )

• Test the top-products endpoint with a date range by running:

curl "http://127.0.0.1:8000/analytics/top-products?start_date=2026-01-01&end_date=2026-12-31"

• Test the revenue summary with the same date range by running:

curl "http://127.0.0.1:8000/analytics/revenue-summary?start_date=2026-01-01&end_date=2026-12-31"

You should see the same data as before (since your orders fall within this range). Try narrowing the range to a date before your orders were placed to confirm filtering works. You should see zeroes for revenue and an empty array for top products.

Seeing a validation error on the date params?

Make sure your dates use the YYYY-MM-DD format in the URL. FastAPI expects ISO 8601 date strings for the date type.

Also check that the URL is wrapped in quotes in your terminal. Without quotes, the & character is interpreted as a shell background operator instead of a query parameter separator.

help me fix date filtering on my analytics endpoints

Your analytics endpoints are live. You now have a complete API that handles product listings with pagination, transactional orders, and aggregation queries. Next up, you'll tackle the Secret Mission and add authentication to protect these endpoints.

Secret mission

JWT Authentication with Role-Based Access

Your API is fully functional, but anyone can hit any endpoint. In a production system, sensitive data like sales analytics should be restricted to admins, and order placement should require an authenticated user. Add JWT-based authentication with role-based access control to lock down your API.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

• The inventory_api PostgreSQL database.
• Your project folder on the Desktop (includes virtual environment, code files, and Alembic migrations).
• The FastAPI dev server process (if still running).

Keep everything running

No action needed. Choose this if you are still actively building on the API or plan to extend it further.

• Your FastAPI dev server continues to serve requests on http://127.0.0.1:8000.
• The inventory_api database stays available for queries and new migrations.
• Your virtual environment and installed packages remain on disk.

Pause - I'll come back to this later

Stop the running dev server to free up the terminal, but keep your files and database intact so you can resume later.

• In the terminal running the FastAPI dev server, press Ctrl+C to stop the process.
• If your virtual environment is active, deactivate it by running:

deactivate

Your project files and database remain untouched. When you come back, activate the virtual environment and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your database, code files, and virtual environment permanently.

• Stop the FastAPI dev server by pressing Ctrl+C in its terminal.
• Deactivate your virtual environment by running:

deactivate

• Drop the PostgreSQL database by running:

dropdb inventory_api

• Delete your project folder from the Desktop by running:

cd ~/Desktop
rm -rf inventory-api

Using a different folder name?

Replace inventory-api in the command above with whatever you named your project folder in Step 1.

• Confirm the folder is gone by running:

ls

You should no longer see your project folder in the output.

Nice Work!

Nice Work!

You just built a production-patterned REST API with multi-table relationships, transactional order processing, and analytics queries.

You've learned how to:

• Design a multi-table PostgreSQL schema with foreign key relationships and manage it with Alembic migrations.
• Implement cursor-based pagination and multi-field filtering for scalable product listings.
• Build transactional order placement that atomically validates and decrements stock using SELECT FOR UPDATE locking.
• Secret Mission: Added JWT authentication with role-based access control to protect endpoints with admin and customer roles.

Ready to quiz yourself?