FastAPI Backend + Telegram Storefront

Build a watch inventory API and Telegram bot storefront with lead capture.

Introduction

30 Second Summary

Luxury watch sellers often run their businesses through messaging apps. They share photos, field questions, and track interested buyers manually. A single backend system that manages inventory and connects directly to a chat storefront could automate the entire pipeline.

In this project, you will build a FastAPI backend that manages watch listings and connects them to a Telegram bot storefront. Customers browse your catalog through the bot, press "I want this," and the system captures their intent as a lead for the operator.

What You'll Build

A monolithic FastAPI backend that acts as the system-of-record for a premium watch commerce operation, exposing inventory to a Telegram bot storefront and capturing customer purchase-intent leads.

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

  • A working FastAPI backend with full watch inventory CRUD endpoints and PostgreSQL persistence via async SQLAlchemy models and Alembic migrations.
  • A Telegram bot storefront where customers can browse available watches, view pricing and details, and express purchase intent.
  • A lead capture pipeline that records customer interactions and surfaces them through admin API endpoints with a dashboard summary.
  • Secret Mission: Add inline keyboard pagination so customers can browse watches page-by-page with "Next" and "Previous" buttons.

Are there any prerequisites?

This project assumes intermediate Python experience, familiarity with REST APIs, and basic SQL knowledge. You do not need prior experience with FastAPI or Telegram bots.

Before We Start

You are building a backend system that connects a FastAPI REST API to a Telegram bot storefront. The API manages watch inventory, and the bot lets customers browse listings and express purchase intent.

Before writing any application code, you need three things in place: Python with an async-ready package set, a PostgreSQL database to persist data, and a clean project folder structure. This step gets your development environment ready so the rest of the project runs smoothly.

In this step, get ready to:
  • Install Python 3.12+ and set up a virtual environment with project dependencies.
  • Install PostgreSQL and create the watchbot database.
  • Create the project directory structure.
Install Python and create the virtual environment

This project uses Python 3.12 or higher for its async capabilities and modern type annotation support. You will check your installed version first, then create an isolated virtual environment for the project.

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

✔️ I see version 3.12 or higher

Your Python version is ready. Move on to creating the project folder below.

ⓧ I see an older version

macOS ships with an older Python by default, and Windows may have an outdated version from a previous install. You need to install Python 3.12 or higher alongside your existing version.

  • Download the latest Python 3.12 installer from python.org/downloads.
  • Run the installer and follow the prompts. On Windows, make sure to check Add python.exe to PATH during setup.
  • After installation, close and reopen your terminal, then run python3 --version again.
  • If macOS still shows the older system Python, invoke the new version directly with python3.12 --version. Use python3.12 instead of python3 for the rest of this project.

ⓧ Command not found

Python is not installed on your machine yet.

  • Download the Python 3.12 installer from python.org/downloads.
  • Run the installer. On Windows, check Add python.exe to PATH before clicking Install Now.
  • Close and reopen your terminal after installation.
  • Run python3 --version again to confirm you see 3.12 or higher.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir watchbot-commerce
cd watchbot-commerce
  • Create a virtual environment by running:
python3 -m venv venv
  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

  • Create a requirements.txt file in your watchbot-commerce folder with the following content:
fastapi[standard]==0.136.3
sqlalchemy[asyncio]==2.0.50
asyncpg==0.31.0
alembic==1.18.4
python-telegram-bot==22.7
pydantic-settings>=2.0.0

What are these packages?

  • fastapi[standard] is the web framework for building your REST API with built-in async support.
  • sqlalchemy[asyncio] is the ORM for defining database models and running async queries.
  • asyncpg is the high-performance async PostgreSQL driver.
  • alembic handles database schema migrations so you can evolve your tables over time.
  • python-telegram-bot provides the Telegram Bot API client for building the storefront.
  • pydantic-settings loads environment variables into typed configuration objects.
  • Install all project dependencies by running:
pip install -r requirements.txt

This will download and install all six packages plus their dependencies. You should see a success message once it finishes.

Seeing installation errors?

  • Make sure your virtual environment is active. You should see (venv) in your prompt.
  • If you get a permissions error, do not use sudo. Deactivate and recreate the venv instead.
  • help me debug my pip install errors

✔️ Awesome, I've got everything!

Great. Make sure you have saved your requirements.txt file.

ⓧ I'd like to double check the full code

fastapi[standard]==0.136.3
sqlalchemy[asyncio]==2.0.50
asyncpg==0.31.0
alembic==1.18.4
python-telegram-bot==22.7
pydantic-settings>=2.0.0
Install PostgreSQL and create the database

Your application stores watch listings, customer records, and leads in a PostgreSQL database. You need PostgreSQL running locally and a database named watchbot ready to accept connections.

  • Check if PostgreSQL is already installed by running:
psql --version

✔️ I see a version number

PostgreSQL is installed. Skip ahead to creating the database below.

ⓧ Command not found

PostgreSQL is not installed yet. Follow the instructions for your platform.

macOS

  • Install PostgreSQL using Homebrew by running:
brew install postgresql
  • Start the PostgreSQL service by running:
brew services start postgresql

Windows

  • Download the Windows installer from postgresql.org/download/windows.
  • Run the installer and follow the setup wizard. Keep the default port (5432) and remember the password you set for the postgres user.
  • Make sure to check Command Line Tools during installation so that psql is available in your terminal.
  • Close and reopen your terminal after installation.
  • Verify PostgreSQL is now available by running psql --version again.
  • Create the watchbot database by running:
createdb watchbot

What does createdb do?

createdb is a PostgreSQL utility that creates a new database. It is a wrapper around the SQL command CREATE DATABASE. If the command produces no output, the database was created successfully.

  • Verify the database exists by running:
psql -l | grep watchbot

You should see a line containing watchbot in the output. This confirms your database is ready.

Permission denied or role does not exist?

  • On macOS with Homebrew, PostgreSQL creates a role matching your macOS username. If you see a role error, try createdb -U postgres watchbot instead.
  • On Windows, you may need to specify the user: createdb -U postgres watchbot. Enter the password you set during installation.
  • Make sure the PostgreSQL server is running. On macOS, run brew services start postgresql. On Windows, check that the PostgreSQL service is started in Services.
  • help me fix my PostgreSQL database creation error
Create the project directory structure

Your project separates concerns into distinct folders. The app/ directory holds the FastAPI backend, and bot/ holds the Telegram bot code.

  • Create all required directories by running these commands:
mkdir -p app/models app/routers app/schemas app/core
mkdir bot

What is this folder structure?

  • app/models/ contains SQLAlchemy database model definitions.
  • app/routers/ contains FastAPI route handlers grouped by feature.
  • app/schemas/ contains Pydantic request and response schemas.
  • app/core/ contains shared configuration and database connection setup.
  • bot/ contains the Telegram bot handlers and startup logic.
  • Confirm the directories exist by running:
ls

You should see app, bot, requirements.txt, and venv listed in the output.

  • Verify the nested structure inside app/ by running:
ls app

You should see core, models, routers, and schemas listed.

Missing folders in the output?

  • Make sure you ran the mkdir commands from inside your watchbot-commerce folder. Run pwd to check your current location.
  • The -p flag creates parent directories automatically. If you missed it, just re-run the mkdir -p command.
  • help me fix my project directory structure

Your development environment is fully configured. Next up, you will define the database models that represent watches, leads, and customers.

Set Up Configuration and Database Models

Your project structure is ready and all dependencies are installed. Now you need a foundation before writing any API logic: a configuration layer that reads environment variables, a database connection pool, and the data models that define your schema.

In this step, you will create the settings module, define three SQLAlchemy models (watches, customers, and leads), and use Alembic to generate and run database migrations that turn those models into real PostgreSQL tables.

In this step, get ready to:
  • Create a configuration layer with environment variable loading and an async database session factory.
  • Define SQLAlchemy models for watches, customers, and leads tables.
  • Configure Alembic for async migrations, generate the initial migration, and apply it.
Create the configuration layer

Every file in the app/ package needs an __init__.py to be importable as a Python module. Some of these will stay empty, while others will re-export symbols for convenience.

  • Create the empty init files by running these commands:
touch app/__init__.py app/core/__init__.py app/schemas/__init__.py
  • Confirm the files exist by running:
ls app/__init__.py app/core/__init__.py app/schemas/__init__.py

You should see all three file paths listed in the output.

Next you need an environment file to store connection strings and secrets outside of your code. The .env file keeps sensitive values out of version control.

  • Create a .env file in your project root with the following content:
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/watchbot
TELEGRAM_BOT_TOKEN=your-bot-token-here

What does this file do?

DATABASE_URL tells SQLAlchemy how to connect to your PostgreSQL database using the asyncpg async driver.

TELEGRAM_BOT_TOKEN is a placeholder you will fill in later when you create your Telegram bot.

  • Save .env.

Now create the settings module that reads these environment variables at startup.

  • Create app/core/config.py with the following code:
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Connection string for async PostgreSQL access
    database_url: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/watchbot"
    # Telegram bot token (filled in later)
    telegram_bot_token: str = ""

    class Config:
        env_file = ".env"


# Single instance used across the app
settings = Settings()

What does this code do?

  • Settings inherits from BaseSettings (from pydantic-settings), which automatically loads values from environment variables and .env files.
  • The Config inner class tells pydantic-settings to look for a .env file in the project root.
  • settings = Settings() creates one shared instance that other modules import directly.
  • Save app/core/config.py.

With settings in place, you can now build the database session factory that the rest of your app will use to talk to PostgreSQL.

  • Create app/core/database.py with the following code:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.core.config import settings

# Create the async engine pointing at our PostgreSQL database
engine = create_async_engine(settings.database_url, echo=True)
# Factory that produces AsyncSession instances
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db():
    # Dependency that yields a session and auto-closes it
    async with async_session() as session:
        yield session

What does this code do?

  • create_async_engine opens a connection pool to PostgreSQL using the asyncpg driver. echo=True logs every SQL statement to the console during development.
  • async_sessionmaker is a factory that creates new AsyncSession objects on demand. expire_on_commit=False keeps attribute values accessible after a commit without re-querying.
  • get_db() is a FastAPI dependency. Each request gets its own session that auto-closes when the request finishes.
  • Save app/core/database.py.

Seeing an import error?

If your editor underlines pydantic_settings or sqlalchemy.ext.asyncio, make sure your virtual environment is activated. Check that you ran pip install -r requirements.txt in the previous step.

Help me fix an import error in my config or database module.

✔️ Awesome, I've got everything!

Great. Double check you have saved both config.py and database.py.

ⓧ I'd like to double check the full code

Here is app/core/config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_url: str = "postgresql+asyncpg://postgres:postgres@localhost:5432/watchbot"
    telegram_bot_token: str = ""

    class Config:
        env_file = ".env"


settings = Settings()

Here is app/core/database.py:

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

from app.core.config import settings

engine = create_async_engine(settings.database_url, echo=True)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


async def get_db():
    async with async_session() as session:
        yield session
Define the database models

Models describe the shape of your data in Python classes. SQLAlchemy translates these classes into PostgreSQL tables when you run migrations.

  • Create app/models/base.py with the base class that all models inherit from:
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass
  • Save app/models/base.py.

The Watch model is the core of your inventory system. It tracks each watch listing with pricing, condition, and a stock status that moves through available, reserved, and sold.

  • Create app/models/watch.py with the following code:
import enum
from datetime import datetime

from sqlalchemy import String, Numeric, Text, Boolean, DateTime, Enum
from sqlalchemy.orm import Mapped, mapped_column


from app.models.base import Base


# Enum restricts stock_status to exactly three states
class StockStatus(str, enum.Enum):
    available = "available"
    reserved = "reserved"
    sold = "sold"


class Watch(Base):
    __tablename__ = "watches"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    brand: Mapped[str] = mapped_column(String(100))
    # Numeric(12,2) stores prices up to 9,999,999,999.99
    price: Mapped[float] = mapped_column(Numeric(12, 2))
    currency: Mapped[str] = mapped_column(String(3), default="USD")
    stock_status: Mapped[StockStatus] = mapped_column(
        Enum(StockStatus), default=StockStatus.available
    )
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    condition: Mapped[str | None] = mapped_column(String(50), nullable=True)
    reference_code: Mapped[str | None] = mapped_column(String(100), nullable=True)
    # Controls whether the watch appears in the storefront
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
    )

What does this model do?

  • StockStatus is a Python enum that constrains the stock_status column to three possible values. This prevents invalid states at the database level.
  • is_active lets the admin hide a watch from the storefront without deleting it. The bot only shows watches where is_active == True.
  • updated_at with onupdate=datetime.utcnow automatically refreshes the timestamp whenever any column on the row changes.
  • Save app/models/watch.py.

Leads capture customer purchase intent. Each lead links a customer to a specific watch they want.

  • Create app/models/lead.py with the following code:
from datetime import datetime

from sqlalchemy import String, Integer, DateTime, ForeignKey, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base


# Stores Telegram user details for lead attribution
class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    telegram_id: Mapped[int] = mapped_column(Integer, unique=True, nullable=True)
    first_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
    username: Mapped[str | None] = mapped_column(String(100), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)


# Each lead records one "I want this" interaction
class Lead(Base):
    __tablename__ = "leads"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    watch_id: Mapped[int] = mapped_column(ForeignKey("watches.id"))
    platform: Mapped[str] = mapped_column(String(50), default="telegram")
    message: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # Relationships for easy access to related objects
    customer: Mapped["Customer"] = relationship()
    watch: Mapped["Watch"] = relationship()

What does this code do?

  • Customer stores the Telegram user who expressed interest. The telegram_id column is unique so the same user only gets one customer record.
  • Lead joins a customer to a watch via foreign keys. The platform column defaults to "telegram" so you could extend this to other channels later.
  • relationship() lets you access the full Customer or Watch object from a Lead instance without writing a separate query.
  • Save app/models/lead.py.

Finally, create the models package init that re-exports all models so Alembic can discover them in one import.

  • Create app/models/__init__.py with the following imports:
from app.models.base import Base
from app.models.watch import Watch
from app.models.lead import Lead, Customer

Why re-export here?

Alembic's autogenerate feature scans Base.metadata to discover tables. Importing all models here ensures they register with Base before Alembic runs.

  • Save app/models/__init__.py.

Seeing a circular import error?

If Python complains about circular imports, make sure app/models/lead.py imports Base from app.models.base (not from app.models). The relationship string Mapped["Watch"] uses a forward reference so Python does not need the Watch class imported at define time.

Help me debug a circular import in my SQLAlchemy models.

✔️ Awesome, I've got everything!

Great. Make sure all four model files are saved.

ⓧ I'd like to double check the full code

Here is app/models/base.py:

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

Here is app/models/watch.py:

import enum
from datetime import datetime

from sqlalchemy import String, Numeric, Text, Boolean, DateTime, Enum
from sqlalchemy.orm import Mapped, mapped_column


from app.models.base import Base


class StockStatus(str, enum.Enum):
    available = "available"
    reserved = "reserved"
    sold = "sold"


class Watch(Base):
    __tablename__ = "watches"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    brand: Mapped[str] = mapped_column(String(100))
    price: Mapped[float] = mapped_column(Numeric(12, 2))
    currency: Mapped[str] = mapped_column(String(3), default="USD")
    stock_status: Mapped[StockStatus] = mapped_column(
        Enum(StockStatus), default=StockStatus.available
    )
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    condition: Mapped[str | None] = mapped_column(String(50), nullable=True)
    reference_code: Mapped[str | None] = mapped_column(String(100), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
    )

Here is app/models/lead.py:

from datetime import datetime

from sqlalchemy import String, Integer, DateTime, ForeignKey, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    telegram_id: Mapped[int] = mapped_column(Integer, unique=True, nullable=True)
    first_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
    username: Mapped[str | None] = mapped_column(String(100), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)


class Lead(Base):
    __tablename__ = "leads"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    watch_id: Mapped[int] = mapped_column(ForeignKey("watches.id"))
    platform: Mapped[str] = mapped_column(String(50), default="telegram")
    message: Mapped[str | None] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    customer: Mapped["Customer"] = relationship()
    watch: Mapped["Watch"] = relationship()

Here is app/models/__init__.py:

from app.models.base import Base
from app.models.watch import Watch
from app.models.lead import Lead, Customer
Configure Alembic and run the initial migration

Alembic tracks changes to your models and generates SQL migration scripts that update the database schema to match. You need to initialize its folder structure and then configure it to work with your async engine.

  • Initialize the Alembic migration environment by running:
alembic init alembic

What does this command do?

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

  • Confirm the new files exist by running:
ls alembic/env.py alembic.ini

You should see both paths listed.

The generated alembic.ini has a placeholder database URL that needs updating.

  • In alembic.ini, find this line:
sqlalchemy.url = driver://user:pass@localhost/dbname
  • Replace it with your actual database URL so the line looks like this:
sqlalchemy.url = postgresql+asyncpg://postgres:postgres@localhost:5432/watchbot
  • Save alembic.ini.

The default env.py uses synchronous connections. You need to replace it entirely with an async-compatible version that imports your models and settings.

  • Replace the entire contents of alembic/env.py with the following:
import asyncio
from logging.config import fileConfig

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

from alembic import context

# Import our models and settings so Alembic knows the schema
from app.models import Base
from app.core.config import settings

config = context.config
# Override the ini-file URL with our settings value
config.set_main_option("sqlalchemy.url", settings.database_url)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Alembic compares this metadata to the live database to detect changes
target_metadata = Base.metadata


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


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations():
    # Create an async engine from the alembic config section
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()


def run_migrations_online() -> None:
    # Use asyncio.run to bridge the sync Alembic entry point to async
    asyncio.run(run_async_migrations())


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

What does this code do?

  • from app.models import Base loads all your model classes (since models/__init__.py imports them). Alembic reads Base.metadata to detect what tables should exist.
  • config.set_main_option("sqlalchemy.url", settings.database_url) overrides the static URL in alembic.ini with whatever your .env provides. This avoids duplicating connection strings.
  • run_async_migrations() bridges the gap between Alembic (which is synchronous) and your async database driver by using asyncio.run().
  • Save alembic/env.py.

Now generate the migration script that creates all three tables.

  • Generate the initial migration by running:
alembic revision --autogenerate -m "initial tables"

What does this command do?

--autogenerate compares your model metadata to the current database state and generates the CREATE TABLE statements automatically. The -m flag sets a human-readable message for the revision file.

You should see output showing a new revision file was created inside alembic/versions/.

  • Apply the migration to your database by running:
alembic upgrade head

You should see SQL statements logged to the terminal as Alembic creates the watches, customers, and leads tables.

  • Verify the tables exist by connecting to PostgreSQL and listing them:
psql -d watchbot -c "\dt"

You should see watches, customers, leads, and alembic_version in the table list.

Migration failing to connect?

Make sure PostgreSQL is running and the watchbot database exists. Check that the username and password in your .env match your local PostgreSQL credentials (the default is postgres:postgres on most local installs).

If you see a "target database is not up to date" error, try running alembic stamp head first, then re-run alembic upgrade head.

Help me fix my Alembic migration connection error.

Your database schema is live and your models are ready to back a full CRUD API. Next up, you will build the FastAPI application and expose endpoints for managing watch inventory.

Build the FastAPI Application and Watch CRUD

Your database schema is live with tables for watches, customers, and leads. Now it's time to build the API layer that exposes your watch inventory to the outside world.

FastAPI generates interactive documentation automatically from your Python code. By the end of this step, you'll have a fully functional CRUD API you can test directly in your browser.

In this step, get ready to:
  • Create Pydantic validation schemas for watch request and response data.
  • Build the watch CRUD router with filtering and status management.
  • Test all endpoints using FastAPI's auto-generated Swagger docs.
Define the watch request and response schemas

Before building API endpoints, you need Pydantic models that define what data the API accepts and what it returns. These schemas validate incoming requests and serialize database objects into JSON responses.

  • Create a new file app/schemas/watch.py and add the following schemas:
from pydantic import BaseModel
from datetime import datetime


# Schema for creating a new watch listing
class WatchCreate(BaseModel):
    name: str
    brand: str
    price: float
    currency: str = "USD"
    description: str | None = None
    condition: str | None = None
    reference_code: str | None = None


# Schema for updating an existing watch (all fields optional)
class WatchUpdate(BaseModel):
    name: str | None = None
    brand: str | None = None
    price: float | None = None
    currency: str | None = None
    description: str | None = None
    condition: str | None = None
    reference_code: str | None = None


# Schema for changing stock status only
class WatchStatusUpdate(BaseModel):
    stock_status: str


# Schema for API responses (includes all database fields)
class WatchResponse(BaseModel):
    id: int
    name: str
    brand: str
    price: float
    currency: str
    stock_status: str
    description: str | None
    condition: str | None
    reference_code: str | None
    is_active: bool
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

What do these schemas do?

Each schema serves a different purpose in the API lifecycle:

  • WatchCreate defines required fields for adding a new watch. Fields like currency have defaults so the client doesn't need to send them every time.
  • WatchUpdate makes every field optional so clients can update only what changed.
  • WatchResponse controls what the API sends back. The from_attributes = True setting lets Pydantic read data directly from SQLAlchemy model instances.
  • Save app/schemas/watch.py.
  • Create an empty file app/routers/__init__.py to make the routers directory a Python package:
touch app/routers/__init__.py
Build the watch inventory router

The router groups all watch-related endpoints under the /watches URL prefix. Each endpoint uses the async database session you set up in the previous step.

  • Create app/routers/watches.py and add the imports, router definition, and the list and create endpoints:
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.models.watch import Watch, StockStatus
from app.schemas.watch import WatchCreate, WatchUpdate, WatchStatusUpdate, WatchResponse

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


@router.get("/", response_model=list[WatchResponse])
async def list_watches(
    status: str | None = None,
    brand: str | None = None,
    active_only: bool = True,
    db: AsyncSession = Depends(get_db),
):
    query = select(Watch)
    if active_only:
        query = query.where(Watch.is_active == True)
    if status:
        query = query.where(Watch.stock_status == status)
    if brand:
        query = query.where(Watch.brand.ilike(f"%{brand}%"))
    query = query.order_by(Watch.created_at.desc())
    result = await db.execute(query)
    return result.scalars().all()


@router.post("/", response_model=WatchResponse, status_code=201)
async def create_watch(watch_data: WatchCreate, db: AsyncSession = Depends(get_db)):
    watch = Watch(**watch_data.model_dump())
    db.add(watch)
    await db.commit()
    await db.refresh(watch)
    return watch

What does this code do?

The list_watches endpoint supports three optional query parameters. The active_only filter defaults to True so clients only see active listings unless they explicitly request all watches. The brand filter uses ilike for case-insensitive partial matching.

The create_watch endpoint converts the Pydantic schema to a dictionary with model_dump(), passes it to the SQLAlchemy model constructor, commits to the database, and returns the newly created record with its auto-generated id and timestamps.

  • Save app/routers/watches.py.
  • Now add the update and status change endpoints below the create_watch function in the same file:
@router.put("/{watch_id}", response_model=WatchResponse)
async def update_watch(
    watch_id: int, watch_data: WatchUpdate, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    update_data = watch_data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(watch, key, value)
    await db.commit()
    await db.refresh(watch)
    return watch


@router.patch("/{watch_id}/status", response_model=WatchResponse)
async def update_watch_status(
    watch_id: int, status_data: WatchStatusUpdate, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    watch.stock_status = StockStatus(status_data.stock_status)
    await db.commit()
    await db.refresh(watch)
    return watch

What does this code do?

The update_watch endpoint uses exclude_unset=True when dumping the schema. This means only fields the client actually sent get updated. If the client sends just {"price": 5000}, only the price changes.

The update_watch_status endpoint is a dedicated PATCH route for changing stock status. It validates the incoming string against the StockStatus enum, rejecting invalid values at runtime.

  • Save app/routers/watches.py.

Seeing import errors in your editor?

Make sure your virtual environment is activated. Your editor needs to use the Python interpreter from the virtual environment to resolve the project imports correctly.

Help me fix import errors in my watches router.

Add the activate and delete endpoints, then wire up the app

The final two router endpoints let an admin toggle a watch's visibility and permanently remove a listing. After that, you'll create the FastAPI application entry point that ties everything together.

  • Add the following two endpoints below update_watch_status in app/routers/watches.py:
@router.patch("/{watch_id}/activate", response_model=WatchResponse)
async def toggle_watch_active(
    watch_id: int, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    watch.is_active = not watch.is_active
    await db.commit()
    await db.refresh(watch)
    return watch


@router.delete("/{watch_id}", status_code=204)
async def delete_watch(watch_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    await db.delete(watch)
    await db.commit()

What does this code do?

The toggle_watch_active endpoint flips the is_active boolean. This gives you a soft way to hide watches from the storefront without deleting them from the database.

The delete_watch endpoint permanently removes a record. It returns HTTP 204 (no content) on success, which is the standard response for successful deletions.

  • Save app/routers/watches.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.models.watch import Watch, StockStatus
from app.schemas.watch import WatchCreate, WatchUpdate, WatchStatusUpdate, WatchResponse

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


@router.get("/", response_model=list[WatchResponse])
async def list_watches(
    status: str | None = None,
    brand: str | None = None,
    active_only: bool = True,
    db: AsyncSession = Depends(get_db),
):
    query = select(Watch)
    if active_only:
        query = query.where(Watch.is_active == True)
    if status:
        query = query.where(Watch.stock_status == status)
    if brand:
        query = query.where(Watch.brand.ilike(f"%{brand}%"))
    query = query.order_by(Watch.created_at.desc())
    result = await db.execute(query)
    return result.scalars().all()


@router.post("/", response_model=WatchResponse, status_code=201)
async def create_watch(watch_data: WatchCreate, db: AsyncSession = Depends(get_db)):
    watch = Watch(**watch_data.model_dump())
    db.add(watch)
    await db.commit()
    await db.refresh(watch)
    return watch


@router.put("/{watch_id}", response_model=WatchResponse)
async def update_watch(
    watch_id: int, watch_data: WatchUpdate, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    update_data = watch_data.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(watch, key, value)
    await db.commit()
    await db.refresh(watch)
    return watch


@router.patch("/{watch_id}/status", response_model=WatchResponse)
async def update_watch_status(
    watch_id: int, status_data: WatchStatusUpdate, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    watch.stock_status = StockStatus(status_data.stock_status)
    await db.commit()
    await db.refresh(watch)
    return watch


@router.patch("/{watch_id}/activate", response_model=WatchResponse)
async def toggle_watch_active(
    watch_id: int, db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    watch.is_active = not watch.is_active
    await db.commit()
    await db.refresh(watch)
    return watch


@router.delete("/{watch_id}", status_code=204)
async def delete_watch(watch_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Watch).where(Watch.id == watch_id))
    watch = result.scalar_one_or_none()
    if not watch:
        raise HTTPException(status_code=404, detail="Watch not found")
    await db.delete(watch)
    await db.commit()

Now you need the FastAPI application entry point that ties everything together. The app imports a leads router, which you'll build in a later step. For now a stub is enough.

  • Create app/routers/leads.py with a minimal router stub:
from fastapi import APIRouter

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

Why create a stub?

The main application file imports both the watches router and the leads router. Creating an empty router now lets the app start without errors. You'll add the admin lead endpoints in Step 5.

  • Save app/routers/leads.py.
  • Create app/main.py with the FastAPI application:
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.routers.watches import router as watches_router
from app.routers.leads import router as leads_router


# Lifespan context manager for startup/shutdown events
@asynccontextmanager
async def lifespan(app: FastAPI):
    yield


app = FastAPI(title="WatchBot Commerce Control", lifespan=lifespan)
app.include_router(watches_router)
app.include_router(leads_router)


@app.get("/")
async def root():
    return {"status": "running", "service": "WatchBot Commerce Control"}

What does this code do?

The lifespan context manager is where you'd put startup logic (like database pool warm-up) and shutdown logic (like closing connections). For now it's empty, but it keeps the app structured for future use.

The include_router calls register all endpoint groups with the app. FastAPI automatically merges their prefixes and generates documentation for every registered route.

  • Save app/main.py.
  • Start the development server by running:
fastapi dev app/main.py
  • Open http://127.0.0.1:8000/docs in your browser. You should see the Swagger UI with all your watch endpoints listed under the watches tag.

What is Swagger UI?

FastAPI automatically generates interactive API documentation from your code. The Swagger UI at /docs lets you try each endpoint directly in the browser. You can send requests, see responses, and explore the schema definitions without any external tools.

  • In the Swagger UI, expand the POST /watches/ endpoint.
  • Click Try it out.
  • Replace the request body with this test watch:
{
  "name": "Speedmaster Professional",
  "brand": "Omega",
  "price": 6500.00,
  "description": "Moonwatch, manual wind",
  "condition": "Excellent",
  "reference_code": "311.30.42.30.01.005"
}
  • Click Execute. You should see a 201 response with the full watch record, including its auto-generated id and timestamps.
  • Expand the GET /watches/ endpoint, click Try it out, then click Execute. You should see the watch you just created in the response array.

Server won't start?

If you see an ImportError, double check that app/routers/__init__.py and app/routers/leads.py both exist. The main app imports from both router modules.

If you see a database connection error, make sure PostgreSQL is running and the watchbot database exists.

Help me fix my FastAPI server startup error.

Your watch inventory API is live and serving requests. Next up, you'll connect a Telegram bot that lets customers browse these watches directly in their messaging app.

Build the Telegram Bot Storefront

Your FastAPI backend is serving watch inventory data through a clean REST API. Now you need a customer-facing channel where people can actually browse and express interest in watches.

In this step, you will create a Telegram bot that acts as a storefront. Customers will message the bot, browse available watches, and tap a button to say "I want this."

In this step, get ready to:
  • Create a Telegram bot via BotFather and configure the token.
  • Build the bot handlers for /start and /catalog commands.
  • Add the "I want this" inline keyboard button and lead capture callback.
Create a Telegram bot via BotFather

Every Telegram bot starts with a conversation with @BotFather, Telegram's official bot-management tool. BotFather generates the API token your bot needs to authenticate with Telegram's servers.

  • Open Telegram on your phone or desktop.
  • Tap the search bar and type BotFather.
  • Select the verified account with a blue checkmark next to the name.
  • Tap Start or type /start to begin.
  • Type /newbot and send it.
  • BotFather asks for a display name. Type something like WatchBot Store.
  • BotFather asks for a username. This must end in bot and be globally unique. Try something like watchbot_yourname_bot.
  • BotFather responds with your bot token. Copy this token immediately.

What is a bot token?

The token is a long string that authenticates your bot with the Telegram Bot API. Anyone with the token can control the bot. Keep it private and never commit it to a public repository.

Now store the token in your project's environment variables.

  • In your editor, open the .env file in your project root.
  • Find this line:
TELEGRAM_BOT_TOKEN=your-bot-token-here
  • Replace your-bot-token-here with the actual token you copied from BotFather.
  • Save the .env file.

Lost your token?

Open BotFather again in Telegram, type /mybots, select your bot, and tap API Token to view it again.

Help me retrieve my Telegram bot token

Next, create the bot package directory.

  • Create the empty bot/__init__.py file by running:
touch bot/__init__.py
  • Confirm the file exists by running:
ls bot/

You should see __init__.py listed in the output.

Build the /start and /catalog command handlers

The bot needs handler functions that respond when a customer sends a command. The /start command greets new users. The /catalog command queries the database for available watches and sends each one as a formatted message.

  • Create a new file bot/handlers.py and add the following imports and start handler:
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ContextTypes
from sqlalchemy import select

from app.core.database import async_session
from app.models.watch import Watch, StockStatus
from app.models.lead import Lead, Customer


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        "Welcome to the Watch Store!\n\n"
        "Use /catalog to browse our collection.\n"
        "Tap 'I want this' on any watch to express interest."
    )

What does this code do?

  • The imports pull in Telegram's Update and InlineKeyboardButton classes, your database session factory, and your SQLAlchemy models.
  • The start function sends a welcome message when the customer first opens the bot.
  • Below the start function in the same file, add the catalog handler:
async def catalog(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    async with async_session() as db:
        result = await db.execute(
            select(Watch)
            .where(Watch.is_active == True)
            .where(Watch.stock_status == StockStatus.available)
            .order_by(Watch.created_at.desc())
        )
        watches = result.scalars().all()

    if not watches:
        await update.message.reply_text("No watches available at the moment.")
        return

    for watch in watches:
        text = (
            f"*{watch.name}*\n"
            f"Brand: {watch.brand}\n"
            f"Price: {watch.currency} {watch.price:,.2f}\n"
            f"Condition: {watch.condition or 'N/A'}\n"
            f"Ref: {watch.reference_code or 'N/A'}"
        )
        keyboard = InlineKeyboardMarkup(
            [[InlineKeyboardButton("I want this", callback_data=f"want:{watch.id}")]]
        )
        await update.message.reply_text(
            text, parse_mode="Markdown", reply_markup=keyboard
        )

What does the catalog handler do?

It opens a fresh database session, queries for watches that are both active and available, and loops through each one. Each watch is sent as a Markdown-formatted message with an InlineKeyboardMarkup containing an "I want this" button.

The callback_data on each button encodes the watch ID (e.g. want:3). This value is what your callback handler receives when a customer taps the button.

  • Save bot/handlers.py.

Import errors in your editor?

If your editor shows red underlines on the telegram imports, make sure your virtual environment is activated and python-telegram-bot is installed. These imports resolve at runtime even if your editor's language server lags behind.

Help me fix python-telegram-bot import errors

Add the lead capture callback and wire up the bot

When a customer taps "I want this", Telegram sends a callback query back to the bot. You need a handler that receives this callback, creates a lead record in the database, and confirms the action to the customer.

  • In bot/handlers.py, add the handle_want function below the catalog function:
async def handle_want(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    await query.answer()

    watch_id = int(query.data.split(":")[1])
    telegram_user = query.from_user

    async with async_session() as db:
        result = await db.execute(
            select(Customer).where(Customer.telegram_id == telegram_user.id)
        )
        customer = result.scalar_one_or_none()

        if not customer:
            customer = Customer(
                telegram_id=telegram_user.id,
                first_name=telegram_user.first_name,
                username=telegram_user.username,
            )
            db.add(customer)
            await db.commit()
            await db.refresh(customer)

        lead = Lead(
            customer_id=customer.id,
            watch_id=watch_id,
            platform="telegram",
            message=f"Customer expressed interest via Telegram bot",
        )
        db.add(lead)
        await db.commit()

        result = await db.execute(select(Watch).where(Watch.id == watch_id))
        watch = result.scalar_one_or_none()

    watch_name = watch.name if watch else "this watch"
    await query.edit_message_reply_markup(reply_markup=None)
    await query.message.reply_text(
        f"Your interest in *{watch_name}* has been recorded. "
        f"Our team will reach out to you shortly!",
        parse_mode="Markdown",
    )

What does handle_want do?

When the callback fires, it extracts the watch ID from the callback_data string. It then checks if a Customer record already exists for this Telegram user. If not, it creates one.

Next it creates a Lead record linking the customer to the watch. Finally it removes the inline keyboard from the original message and sends a confirmation reply.

  • Save bot/handlers.py.

Now create the bot application entry point that registers all handlers.

  • Create a new file bot/main.py with the following code:
from telegram.ext import ApplicationBuilder, CommandHandler, CallbackQueryHandler

from app.core.config import settings
from bot.handlers import start, catalog, handle_want


def create_bot_application():
    application = (
        ApplicationBuilder().token(settings.telegram_bot_token).build()
    )
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("catalog", catalog))
    application.add_handler(
        CallbackQueryHandler(handle_want, pattern=r"^want:\d+$")
    )
    return application


if __name__ == "__main__":
    app = create_bot_application()
    print("Bot is running... Press Ctrl+C to stop.")
    app.run_polling()

What does this code do?

  • ApplicationBuilder().token(...).build() creates the bot application using the token from your .env file.
  • CommandHandler maps the /start and /catalog commands to their handler functions.
  • CallbackQueryHandler catches inline keyboard callbacks matching the pattern ^want:\d+$ and routes them to handle_want.
  • run_polling() starts a long-polling loop that fetches updates from Telegram's servers continuously.
  • Save bot/main.py.
  • Make sure your FastAPI server is stopped (press Ctrl+C in the terminal where it was running).
  • Start the bot by running:
python -m bot.main

You should see Bot is running... Press Ctrl+C to stop. printed in the terminal.

  • Open Telegram and search for your bot by the username you chose earlier.
  • Tap Start to trigger the /start command.
  • Type /catalog to browse available watches.

What should I see?

After /start, the bot replies with a welcome message. After /catalog, you see each available watch formatted with its name, brand, price, and an "I want this" button.

If you have no watches in the database yet, the bot replies with "No watches available at the moment."

Add test watches via Swagger

If your catalog is empty, start your FastAPI server in a separate terminal with fastapi dev and use the Swagger docs at http://localhost:8000/docs to create a few watches via the POST /watches/ endpoint. Then try /catalog again in Telegram.

Bot not responding?

  • Verify your .env file has the real token (not the placeholder your-bot-token-here).
  • Make sure PostgreSQL is running. The catalog handler queries the database directly.
  • Check the terminal where python -m bot.main is running for error messages.

Help me debug my Telegram bot

✔️ Awesome, I've got everything!

Great. Double check you have saved both bot/handlers.py and bot/main.py.

ⓧ I'd like to double check the full code

Here is the complete bot/handlers.py:

from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ContextTypes
from sqlalchemy import select

from app.core.database import async_session
from app.models.watch import Watch, StockStatus
from app.models.lead import Lead, Customer


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
        "Welcome to the Watch Store!\n\n"
        "Use /catalog to browse our collection.\n"
        "Tap 'I want this' on any watch to express interest."
    )


async def catalog(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    async with async_session() as db:
        result = await db.execute(
            select(Watch)
            .where(Watch.is_active == True)
            .where(Watch.stock_status == StockStatus.available)
            .order_by(Watch.created_at.desc())
        )
        watches = result.scalars().all()

    if not watches:
        await update.message.reply_text("No watches available at the moment.")
        return

    for watch in watches:
        text = (
            f"*{watch.name}*\n"
            f"Brand: {watch.brand}\n"
            f"Price: {watch.currency} {watch.price:,.2f}\n"
            f"Condition: {watch.condition or 'N/A'}\n"
            f"Ref: {watch.reference_code or 'N/A'}"
        )
        keyboard = InlineKeyboardMarkup(
            [[InlineKeyboardButton("I want this", callback_data=f"want:{watch.id}")]]
        )
        await update.message.reply_text(
            text, parse_mode="Markdown", reply_markup=keyboard
        )


async def handle_want(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = update.callback_query
    await query.answer()

    watch_id = int(query.data.split(":")[1])
    telegram_user = query.from_user

    async with async_session() as db:
        result = await db.execute(
            select(Customer).where(Customer.telegram_id == telegram_user.id)
        )
        customer = result.scalar_one_or_none()

        if not customer:
            customer = Customer(
                telegram_id=telegram_user.id,
                first_name=telegram_user.first_name,
                username=telegram_user.username,
            )
            db.add(customer)
            await db.commit()
            await db.refresh(customer)

        lead = Lead(
            customer_id=customer.id,
            watch_id=watch_id,
            platform="telegram",
            message=f"Customer expressed interest via Telegram bot",
        )
        db.add(lead)
        await db.commit()

        result = await db.execute(select(Watch).where(Watch.id == watch_id))
        watch = result.scalar_one_or_none()

    watch_name = watch.name if watch else "this watch"
    await query.edit_message_reply_markup(reply_markup=None)
    await query.message.reply_text(
        f"Your interest in *{watch_name}* has been recorded. "
        f"Our team will reach out to you shortly!",
        parse_mode="Markdown",
    )

Here is the complete bot/main.py:

from telegram.ext import ApplicationBuilder, CommandHandler, CallbackQueryHandler

from app.core.config import settings
from bot.handlers import start, catalog, handle_want


def create_bot_application():
    application = (
        ApplicationBuilder().token(settings.telegram_bot_token).build()
    )
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("catalog", catalog))
    application.add_handler(
        CallbackQueryHandler(handle_want, pattern=r"^want:\d+$")
    )
    return application


if __name__ == "__main__":
    app = create_bot_application()
    print("Bot is running... Press Ctrl+C to stop.")
    app.run_polling()

Your storefront is live. Customers can browse the catalog and express purchase intent directly from Telegram. Next up, you will build the admin API endpoints that surface these leads and provide an operational dashboard.

Wire Up Lead Capture and Admin API

Your Telegram bot can now display watches and capture purchase intent through the "I want this" button.

Captured leads are useless if the operator cannot see them. In this step, you will build admin-facing API endpoints that surface leads and dashboard statistics, then wire the bot and API to run together as one system.

In this step, get ready to:
  • Create Pydantic response schemas for leads and the admin dashboard.
  • Build admin API endpoints that list leads, retrieve lead details, and return dashboard counts.
  • Wire up the bot and FastAPI to run concurrently, then test the full flow end-to-end.
Create the lead response schemas

The admin endpoints need Pydantic models to serialize lead data and dashboard statistics into JSON responses. You will define two schemas: one for individual lead records and one for the aggregated dashboard counts.

  • Create a new file app/schemas/lead.py and add the following code:
from pydantic import BaseModel
from datetime import datetime


class LeadResponse(BaseModel):
    id: int
    customer_id: int
    watch_id: int
    platform: str
    message: str | None
    created_at: datetime

    class Config:
        from_attributes = True


class DashboardResponse(BaseModel):
    total_watches: int
    active_listings: int
    total_leads: int
    leads_today: int

What does this code do?

LeadResponse serializes a single lead record from the database into JSON. The from_attributes = True setting tells Pydantic to read values directly from SQLAlchemy model attributes.

DashboardResponse defines the shape of the operator's at-a-glance summary: total watches, active listings, total leads, and leads captured today.

  • Save app/schemas/lead.py.
Build the admin leads and dashboard endpoints

The operator needs a way to see incoming leads without opening Telegram. You will create an admin router with endpoints for listing leads, viewing a single lead, and fetching a summary dashboard.

  • Create a new file app/routers/leads.py and add the imports, router setup, and lead listing endpoints:
from datetime import datetime, date

from fastapi import APIRouter, Depends
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.models.lead import Lead
from app.models.watch import Watch
from app.schemas.lead import LeadResponse, DashboardResponse

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


@router.get("/leads", response_model=list[LeadResponse])
async def list_leads(
    platform: str | None = None,
    db: AsyncSession = Depends(get_db),
):
    query = select(Lead).order_by(Lead.created_at.desc())
    if platform:
        query = query.where(Lead.platform == platform)
    result = await db.execute(query)
    return result.scalars().all()


@router.get("/leads/{lead_id}", response_model=LeadResponse)
async def get_lead(lead_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Lead).where(Lead.id == lead_id))
    lead = result.scalar_one_or_none()
    if not lead:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Lead not found")
    return lead

What does this code do?

  • The router uses the prefix /admin so all endpoints live under /admin/leads and /admin/dashboard.
  • list_leads returns all leads ordered by most recent. The optional platform query parameter lets the admin filter by source (e.g. "telegram").
  • get_lead fetches a single lead by ID and raises a 404 if it does not exist.
  • Now add the dashboard endpoint below the get_lead function in the same file:
@router.get("/dashboard", response_model=DashboardResponse)
async def get_dashboard(db: AsyncSession = Depends(get_db)):
    total_watches = await db.scalar(select(func.count(Watch.id)))
    active_listings = await db.scalar(
        select(func.count(Watch.id)).where(Watch.is_active == True)
    )
    total_leads = await db.scalar(select(func.count(Lead.id)))
    today_start = datetime.combine(date.today(), datetime.min.time())
    leads_today = await db.scalar(
        select(func.count(Lead.id)).where(Lead.created_at >= today_start)
    )
    return DashboardResponse(
        total_watches=total_watches or 0,
        active_listings=active_listings or 0,
        total_leads=total_leads or 0,
        leads_today=leads_today or 0,
    )

What does this code do?

The dashboard endpoint runs four separate COUNT queries against the database. It uses func.count from SQLAlchemy to perform aggregation at the database level rather than loading all rows into Python.

The leads_today count filters leads created after midnight today. The or 0 fallback handles the case where tables are empty and the scalar returns None.

  • Save app/routers/leads.py.

Why does the app already import this router?

If you look at app/main.py, it already includes from app.routers.leads import router as leads_router. That import was added in the previous step in anticipation of this file. Now that the file exists, the FastAPI app can start without an import error.

Seeing a ModuleNotFoundError?

  • Make sure you created the file at app/routers/leads.py (not app/leads.py or routers/leads.py).
  • Verify that app/schemas/lead.py exists and contains both LeadResponse and DashboardResponse.

Help me debug my ModuleNotFoundError for the leads router.

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Here is the complete app/schemas/lead.py:

from pydantic import BaseModel
from datetime import datetime


class LeadResponse(BaseModel):
    id: int
    customer_id: int
    watch_id: int
    platform: str
    message: str | None
    created_at: datetime

    class Config:
        from_attributes = True


class DashboardResponse(BaseModel):
    total_watches: int
    active_listings: int
    total_leads: int
    leads_today: int

And the complete app/routers/leads.py:

from datetime import datetime, date

from fastapi import APIRouter, Depends
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.models.lead import Lead
from app.models.watch import Watch
from app.schemas.lead import LeadResponse, DashboardResponse

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


@router.get("/leads", response_model=list[LeadResponse])
async def list_leads(
    platform: str | None = None,
    db: AsyncSession = Depends(get_db),
):
    query = select(Lead).order_by(Lead.created_at.desc())
    if platform:
        query = query.where(Lead.platform == platform)
    result = await db.execute(query)
    return result.scalars().all()


@router.get("/leads/{lead_id}", response_model=LeadResponse)
async def get_lead(lead_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Lead).where(Lead.id == lead_id))
    lead = result.scalar_one_or_none()
    if not lead:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Lead not found")
    return lead


@router.get("/dashboard", response_model=DashboardResponse)
async def get_dashboard(db: AsyncSession = Depends(get_db)):
    total_watches = await db.scalar(select(func.count(Watch.id)))
    active_listings = await db.scalar(
        select(func.count(Watch.id)).where(Watch.is_active == True)
    )
    total_leads = await db.scalar(select(func.count(Lead.id)))
    today_start = datetime.combine(date.today(), datetime.min.time())
    leads_today = await db.scalar(
        select(func.count(Lead.id)).where(Lead.created_at >= today_start)
    )
    return DashboardResponse(
        total_watches=total_watches or 0,
        active_listings=active_listings or 0,
        total_leads=total_leads or 0,
        leads_today=leads_today or 0,
    )
Create the concurrent runner

Right now the bot and the API are separate processes. The bot uses run_polling() in its own script, and the API uses fastapi dev. You need a single entry point that runs both concurrently so the system operates as one unit.

  • Create a new file run.py in your project root (the same level as app/ and bot/) and add the following code:
import asyncio
import threading

import uvicorn

from bot.main import create_bot_application


def run_api():
    # Start FastAPI on port 8000 in a background thread
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=False)


async def run_bot():
    # Initialize and start the Telegram bot with polling
    application = create_bot_application()
    await application.initialize()
    await application.start()
    await application.updater.start_polling()
    print("Bot is running...")
    try:
        # Keep the bot alive until interrupted
        await asyncio.Event().wait()
    finally:
        # Graceful shutdown sequence
        await application.updater.stop()
        await application.stop()
        await application.shutdown()


if __name__ == "__main__":
    # Launch the API in a daemon thread
    api_thread = threading.Thread(target=run_api, daemon=True)
    api_thread.start()
    # Run the bot in the main asyncio loop
    asyncio.run(run_bot())

What does this code do?

  • run_api() starts the FastAPI server using uvicorn on port 8000. It runs in a daemon thread so it shuts down automatically when the main process exits.
  • run_bot() initializes the Telegram bot application, starts polling for updates, and waits indefinitely until you press Ctrl+C.
  • The finally block runs a graceful shutdown sequence so the bot cleanly disconnects from the Telegram API.

Why a thread for the API?

The Telegram bot needs the main asyncio event loop to receive updates. Uvicorn creates its own event loop internally. Running them in separate threads avoids conflicts between the two loops while still sharing the same database session factory.

  • Save run.py.

Seeing an ImportError for bot.main?

  • Make sure bot/__init__.py exists (it can be empty). Without it, Python does not recognize bot/ as a package.
  • Confirm you are running from the project root directory (the folder that contains both app/ and bot/).

Help me fix this ImportError when running run.py.

✔️ Awesome, I've got everything!

Great. Double check that run.py is saved at the project root.

ⓧ I'd like to double check the full code

import asyncio
import threading

import uvicorn

from bot.main import create_bot_application


def run_api():
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=False)


async def run_bot():
    application = create_bot_application()
    await application.initialize()
    await application.start()
    await application.updater.start_polling()
    print("Bot is running...")
    try:
        await asyncio.Event().wait()
    finally:
        await application.updater.stop()
        await application.stop()
        await application.shutdown()


if __name__ == "__main__":
    api_thread = threading.Thread(target=run_api, daemon=True)
    api_thread.start()
    asyncio.run(run_bot())
Test the complete flow end-to-end

With all the pieces in place, you can now start the entire system and verify the pipeline from Telegram interaction to admin API response.

  • Make sure your .env file has your real Telegram bot token in the TELEGRAM_BOT_TOKEN field.
  • Start the system by running this command:
python run.py
  • You should see uvicorn start on port 8000, followed by "Bot is running..." in your terminal.

Now test the Telegram-to-API pipeline.

  • Open Telegram and send /catalog to your bot.
  • Tap the I want this button on any watch listing.
  • You should see a confirmation message that your interest has been recorded.

Now verify the lead appears in the admin API.

  • Open your browser and navigate to http://localhost:8000/docs to open the Swagger UI.
  • Expand the GET /admin/leads endpoint and click Try it out, then Execute.
  • You should see a JSON array containing the lead you just created via Telegram.
  • Try the GET /admin/dashboard endpoint the same way. You should see counts for total watches, active listings, total leads, and leads today.

Not seeing any leads?

  • Make sure you have at least one watch in the database with is_active = True and stock_status = 'available'. The /catalog command only shows watches that match both conditions.
  • If /catalog returns "No watches available", use Swagger to create a watch first via POST /watches.
  • Check your terminal for any database connection errors. Make sure PostgreSQL is running and the watchbot database exists.

Help me debug why my admin leads endpoint returns an empty array.

Your chat-commerce pipeline is fully operational. The bot captures purchase intent from customers, persists it to the database, and the admin API surfaces those leads on demand. Next up, you will extend the bot with paginated catalog browsing in the Secret Mission.

Secret mission

Paginated Catalog with Inline Keyboards

Your catalog currently dumps every watch into the chat at once. With 10+ watches, that's an overwhelming wall of text. Real storefronts paginate. Can you add inline keyboard navigation so customers browse 3 watches at a time with Next and Previous buttons?

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • watchbot PostgreSQL database (local).
  • Project directory containing app/, bot/, run.py, alembic/, and .env.
  • Telegram bot (created via @BotFather).

Keep everything running

No action needed. Choose this if you are still actively building or want to extend the project further.

  • Your PostgreSQL database and project files remain on your machine.
  • The Telegram bot stays registered under your BotFather account. It only runs when you start the run.py script.
  • Python and PostgreSQL remain installed for future projects.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and database stay intact for when you return.

  • Press Ctrl+C in any terminal running the bot or API to stop the processes.
  • Deactivate your virtual environment by running:
deactivate
  • Optionally stop the PostgreSQL service if you do not need it for other work. You can restart it when you return.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop any running processes by pressing Ctrl+C in your terminal.
  • Deactivate your virtual environment by running:
deactivate
  • Drop the watchbot database by running:
dropdb watchbot
  • Delete the project directory. Navigate to the folder that contains your project and run:

macOS

cd ~/Desktop
rm -rf watchbot-commerce-control

Windows

cd ~\Desktop
rmdir /s /q watchbot-commerce-control
  • Delete the Telegram bot by opening @BotFather in Telegram, sending /deletebot, and selecting your bot from the list. This action cannot be undone.

Nice Work!

Nice Work!

Nice work! You just built a complete chat-commerce backend system from scratch, connecting a REST API to a messaging platform with a real lead capture pipeline.

You've learned how to:

  • Build a FastAPI backend with full CRUD operations, async SQLAlchemy models, and Alembic database migrations for PostgreSQL.
  • Create a Telegram bot storefront that fetches live inventory data and presents it to customers with inline keyboard interactions.
  • Implement a lead capture pipeline that turns customer purchase-intent signals into queryable records surfaced through admin API endpoints.
  • Add paginated inline keyboard navigation to the bot catalog so customers can browse watches page-by-page with Next and Previous buttons.

Ready to quiz yourself?