Build a Real-Time Chat API with FastAPI

Introduction

30 Second Summary

Every time you send a message on Slack or Messenger, a backend system authenticates you, routes your message to the right room, and pushes it to every connected user in real time. That instant delivery relies on WebSockets, database persistence, and secure token-based authentication working together.

In this project, you will build a real-time chat API backend using FastAPI, PostgreSQL, and WebSockets. You'll implement user registration with secure password hashing, JWT authentication, chat room management, and live message delivery.

What You'll Build

A production-style real-time chat API that handles user registration, JWT authentication, chat room creation, and WebSocket-powered live messaging with message history.

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

• A FastAPI application with user registration, login, and JWT-protected endpoints backed by auto-generated Swagger docs.
• A PostgreSQL database with Alembic migrations storing users, rooms, and message history.
• A WebSocket endpoint for real-time messaging with a connection manager that broadcasts messages to all users in a room.
• Secret Mission: Add typing indicators and online presence tracking to the WebSocket layer.

Are there any prerequisites?

You should have a basic understanding of REST APIs and HTTP methods (GET, POST). Basic Python knowledge is also helpful, but you don't need to be an expert.

Before We Start

Before you can build a real-time chat API, you need a database to store users, rooms, and messages. You also need a Python project with the right libraries installed.

In this step, you will install PostgreSQL as your database, create a dedicated database for the project, and set up a Python virtual environment with all the dependencies your chat API needs.

In this step, get ready to:

• Install PostgreSQL and create the chat database.
• Create your project directory with a Python virtual environment.
• Install all project dependencies.

Download and install PostgreSQL

PostgreSQL is an open-source relational database. Your chat API will use it to store user accounts, chat room data, and message history.

• Open your browser and navigate to the EDB PostgreSQL download page:
• https://www.enterprisedb.com/downloads/postgres-postgresql-downloads
• Download the latest Windows installer (the version number will be in the filename).
• Double-click the downloaded installer to launch the setup wizard.
• Click Next through the Welcome and Installation Directory screens, accepting the defaults.
• On the Select Components screen, keep all components selected (PostgreSQL Server, pgAdmin 4, Stack Builder, Command Line Tools). Click Next.
• Accept the default data directory and click Next.
• On the Password screen, enter a password for the postgres superuser. Remember this password. Click Next.
• On the Port screen, keep the default port 5432. Click Next.
• Click Next through the remaining screens and then click Finish to complete the installation.
• Uncheck the Stack Builder option when prompted at the end. You do not need it for this project.

Why do I need to remember the password?

The password you set during installation is for the postgres superuser account. You will use this password every time you connect to the database. You will also include it in your project's connection string.

The PostgreSQL installer does not add its tools to your system PATH by default. You need to add the bin directory manually so you can run psql from any terminal.

• Press the Windows key and type Environment Variables. Open Edit the system environment variables.
• Click Environment Variables.
• Under System variables, find Path and click Edit.
• Click New and add C:\Program Files\PostgreSQL\18\bin (replace 18 with your installed version number if different).
• Click OK on all open dialogs to save.

Now verify that PostgreSQL is installed correctly.

• Open a new terminal window (the PATH changes require a fresh terminal). Press the Windows key, type cmd, and press Enter.
• Verify the installation by running this command:

psql --version

✔️ I see a version number

You should see output similar to psql (PostgreSQL) 18.4. The exact version number may differ. PostgreSQL is installed and ready to use.

ⓧ Command not found

This means the PATH was not updated correctly. Double-check that the path you added points to the correct PostgreSQL version folder. Open a fresh terminal after saving the environment variable changes.

• Open File Explorer and navigate to C:\Program Files\PostgreSQL\ to find the correct version folder.
• Confirm the folder contains a bin subdirectory with psql.exe inside it.
• Update your PATH with the correct path, then open a brand-new terminal and try again.

Create the chat database

PostgreSQL can host multiple databases on a single server. You will create a dedicated database called chatapi to keep your project's data separate.

• Connect to PostgreSQL as the superuser by running this command:

psql -U postgres

What does this command do?

The -U postgres flag tells psql to connect as the postgres superuser. You will be prompted for the password you set during installation.

• Enter the password you chose during the PostgreSQL installation when prompted.
• Create the database by running this SQL command:

CREATE DATABASE chatapi;

• Exit the psql session by running:

\q

You should see your terminal return to the normal command prompt. Your chatapi database is now created and ready to receive tables.

Connection refused or password error?

• If you see a connection refused error, make sure the PostgreSQL service is running. Open the Services app (press the Windows key and search for Services) and confirm that the postgresql service status is Running.
• If you get a password authentication error, try re-entering the password carefully. Passwords are case-sensitive.

Still stuck? Ask for help.

Create the project directory and install dependencies

Now you will create your project folder, set up a Python virtual environment to isolate your dependencies, and install everything your chat API needs.

• Move to your Desktop by running this command:

cd ~/Desktop

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

mkdir fastapi-chat
cd fastapi-chat

• Create a Python virtual environment by running:

python -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay contained to this project instead of affecting your entire system.

• Activate the virtual environment by running:

Windows

venv\Scripts\activate

macOS

source venv/bin/activate

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

• Install all the project dependencies by running:

pip install "fastapi[standard]" "sqlalchemy[asyncio]" asyncpg alembic "pydantic-settings" "pwdlib[argon2]" PyJWT python-multipart

What are these packages?

• fastapi[standard] is your web framework with all the standard extras (includes uvicorn for running the server).
• sqlalchemy[asyncio] is the database toolkit with async support.
• asyncpg is the async PostgreSQL driver.
• alembic handles database migrations (creating and updating tables).
• pydantic-settings loads configuration from environment variables.
• pwdlib[argon2] hashes passwords securely using the Argon2 algorithm.
• PyJWT creates and verifies JSON Web Tokens for authentication.
• python-multipart parses form data (required for the login endpoint).

You should see pip download and install all the packages along with their sub-dependencies. The final output shows a success message.

Seeing installation errors?

• Make sure (venv) appears in your terminal prompt before running pip install. If it does not, re-run the activation command.
• If you see a permissions error, confirm you activated the virtual environment and are not installing globally.
• If a specific package fails to build, make sure you are running Python 3.10 or higher. Run python --version to check.

Still stuck? Help me fix my pip install errors.

• Confirm the installation worked by running:

pip list

You should see fastapi, sqlalchemy, asyncpg, alembic, pydantic-settings, pwdlib, and PyJWT in the list of installed packages.

Your environment is ready. Next up, you will configure the database connection and build the project structure.

Configure the Database and Project Structure

Your environment is ready with all dependencies installed. Now you need a place for your code to live.

Every FastAPI project needs an organized structure so that routers, models, and configuration stay separate. In this step, you'll create the file structure, set up environment-based configuration, and connect to your PostgreSQL database using async SQLAlchemy.

In this step, get ready to:

• Create the project file structure with an app/ directory and a routers/ subdirectory.
• Set up environment-based configuration using pydantic-settings and a .env file.
• Configure the async database engine and session factory.

Create the project file structure

A clean file structure keeps your API organized as it grows. You'll separate concerns into configuration, database, models, schemas, auth, and routers.

• In your terminal (inside the fastapi-chat directory with your virtual environment active), create the directory structure by running these commands:

macOS

mkdir -p app/routers
touch app/__init__.py app/main.py app/config.py app/database.py app/models.py app/schemas.py app/auth.py
touch app/routers/__init__.py app/routers/users.py app/routers/rooms.py app/routers/chat.py

Windows

New-Item -Path app\routers -ItemType Directory
New-Item -ItemType File -Path app\__init__.py, app\main.py, app\config.py, app\database.py, app\models.py, app\schemas.py, app\auth.py
New-Item -ItemType File -Path app\routers\__init__.py, app\routers\users.py, app\routers\rooms.py, app\routers\chat.py

• Confirm the files exist by running:

macOS

ls app/
ls app/routers/

Windows

Get-ChildItem app\
Get-ChildItem app\routers\

You should see __init__.py, main.py, config.py, database.py, models.py, schemas.py, and auth.py in the app/ listing. The routers/ listing should show __init__.py, users.py, rooms.py, and chat.py.

Don't see all the files?

On Windows, make sure you're using PowerShell (not Command Prompt). The New-Item command is PowerShell-specific.

If a file is missing, create it individually with New-Item app\filename.py -ItemType File (Windows) or touch app/filename.py (macOS).

Help me troubleshoot my file structure

Why this structure?

The __init__.py files mark directories as Python packages so your imports work. The routers/ subdirectory keeps each API feature (users, rooms, chat) in its own file instead of cramming everything into one.

Set up environment configuration

Hardcoding database passwords and secret keys in source code is a security risk. Instead, you'll store sensitive values in a .env file that never gets committed to version control.

• Create a .env file in the root of your fastapi-chat folder (not inside app/) and paste in this configuration:

DATABASE_URL=postgresql+asyncpg://postgres:your-password-here@localhost:5432/chatapi
SECRET_KEY=your-secret-key-here-change-in-production
ACCESS_TOKEN_EXPIRE_MINUTES=30

• Replace your-password-here with the password you set for the postgres superuser during PostgreSQL installation.

What does this configuration do?

DATABASE_URL tells SQLAlchemy how to connect to your PostgreSQL database. The postgresql+asyncpg:// scheme specifies the async driver.

SECRET_KEY is used to sign JWT tokens. In production, this would be a long random string. ACCESS_TOKEN_EXPIRE_MINUTES controls how long a login token stays valid.

• Save .env.
• You should see .env appear in your Cursor file explorer alongside the app/ folder.

Don't see the .env file in Cursor?

Dotfiles (files starting with .) are hidden by default on some systems. In Cursor's Explorer sidebar, the file should still appear. Make sure you saved it in the fastapi-chat root, not inside the app/ directory.

Help me find my .env file

Now you need a Python class that reads these environment variables and makes them available throughout your app.

• Open app/config.py in Cursor and paste in the following code:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Database connection string (loaded from .env)
    DATABASE_URL: str = "postgresql+asyncpg://postgres:password@localhost:5432/chatapi"
    # Secret key for signing JWT tokens
    SECRET_KEY: str = "your-secret-key-here-change-in-production"
    # How long access tokens stay valid (in minutes)
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    # Algorithm used for JWT encoding
    ALGORITHM: str = "HS256"

    class Config:
        env_file = ".env"
        case_sensitive = True


settings = Settings()

What does this code do?

The Settings class inherits from BaseSettings, which automatically loads values from your .env file. Each class attribute defines a configuration variable with a type and default value.

The settings = Settings() line at the bottom creates a single instance that your entire app imports. Any module can do from app.config import settings to access configuration values.

• Save app/config.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved both .env and app/config.py.

ⓧ I'd like to double check the full code

Here is the full .env file:

DATABASE_URL=postgresql+asyncpg://postgres:your-password-here@localhost:5432/chatapi
SECRET_KEY=your-secret-key-here-change-in-production
ACCESS_TOKEN_EXPIRE_MINUTES=30

Here is the full app/config.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://postgres:password@localhost:5432/chatapi"
    SECRET_KEY: str = "your-secret-key-here-change-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ALGORITHM: str = "HS256"

    class Config:
        env_file = ".env"
        case_sensitive = True


settings = Settings()

Configure the async database engine

FastAPI is async-native, meaning it can handle many requests at once without blocking. Your database layer needs to match that capability.

SQLAlchemy 2.0+ provides an async engine that pairs with asyncpg (the async PostgreSQL driver you installed). This lets your API serve other requests while waiting for database queries to return.

• Open app/database.py in Cursor and paste in the following code:

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

from app.config import settings

# Create the async database engine
engine = create_async_engine(
    settings.DATABASE_URL,
    echo=True,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

# Session factory for creating database sessions
async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


# Base class for all SQLAlchemy models
class Base(DeclarativeBase):
    pass


# Dependency that provides a database session to each request
async def get_db():
    async with async_session() as session:
        try:
            yield session
        finally:
            await session.close()

What does this code do?

create_async_engine creates a connection pool to your PostgreSQL database. The echo=True setting logs all SQL queries to your terminal, which is helpful for debugging. pool_pre_ping=True checks that connections are alive before using them.

async_sessionmaker is a factory that creates new database sessions. Each API request gets its own session through the get_db dependency. The yield pattern ensures the session always gets closed when the request finishes.

Base is the parent class for all your database models. Every table you define will inherit from it.

• Save app/database.py.
• You won't see any visible output yet. The database engine connects when the app starts running, which happens in a later step.

Seeing a red underline on the import?

If Cursor shows an import error on from app.config import settings, make sure your virtual environment is selected as the Python interpreter. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type Python: Select Interpreter, and choose the one inside your venv folder.

Help me fix my Python import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/database.py.

ⓧ I'd like to double check the full code

Here is the full app/database.py:

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

from app.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=True,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


async def get_db():
    async with async_session() as session:
        try:
            yield session
        finally:
            await session.close()

Your project structure and database configuration are ready. Next up, you'll define the database models that represent your users, rooms, and messages.

Define Database Models and Run Migrations

Every chat application needs a structured way to store users, rooms, and messages. Without a schema, your data has no shape and your queries have no guarantees.

In this step, you'll define SQLAlchemy models that map Python classes to database tables. Then you'll use Alembic to generate and apply a migration that creates those tables in PostgreSQL.

In this step, get ready to:

• Define User, Room, RoomMember, and Message models with relationships.
• Initialize Alembic with async support and configure it for your database.
• Generate and apply the initial migration to create tables in PostgreSQL.

Define SQLAlchemy models

Your database schema needs four tables: users to store accounts, rooms for chat channels, a join table linking users to rooms, and messages for the actual chat content.

• Open app/models.py in Cursor.
• Add the imports and the User model by pasting the following code:

import uuid
from datetime import datetime, timezone

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

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    messages: Mapped[list["Message"]] = relationship(back_populates="sender")
    memberships: Mapped[list["RoomMember"]] = relationship(back_populates="user")

What does this code do?

Each column is declared with Mapped type hints, which tells SQLAlchemy both the Python type and the database column type in one place.

The id column generates a UUID string as the primary key. The relationship() calls at the bottom define how this model connects to others. A user has many messages and many room memberships.

• Below the User class, add the remaining three models:

class Room(Base):
    __tablename__ = "rooms"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    created_by: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    members: Mapped[list["RoomMember"]] = relationship(back_populates="room")
    messages: Mapped[list["Message"]] = relationship(back_populates="room")


class RoomMember(Base):
    __tablename__ = "room_members"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    room_id: Mapped[str] = mapped_column(String(36), ForeignKey("rooms.id"), nullable=False)
    user_id: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    joined_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    room: Mapped["Room"] = relationship(back_populates="members")
    user: Mapped["User"] = relationship(back_populates="memberships")


class Message(Base):
    __tablename__ = "messages"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    content: Mapped[str] = mapped_column(Text, nullable=False)
    room_id: Mapped[str] = mapped_column(String(36), ForeignKey("rooms.id"), nullable=False)
    sender_id: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    room: Mapped["Room"] = relationship(back_populates="messages")
    sender: Mapped["User"] = relationship(back_populates="messages")

How do these models relate to each other?

Room tracks who created it via created_by (a foreign key to users.id). RoomMember is a join table that links users to rooms, enabling many-to-many membership.

Message stores the actual chat content. Each message belongs to one room and one sender. The back_populates arguments keep both sides of each relationship in sync.

• Save app/models.py.

You won't see a change yet. The next substep uses Alembic to turn these model definitions into actual database tables.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file.

ⓧ I'd like to double check the full code

import uuid
from datetime import datetime, timezone

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

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    messages: Mapped[list["Message"]] = relationship(back_populates="sender")
    memberships: Mapped[list["RoomMember"]] = relationship(back_populates="user")


class Room(Base):
    __tablename__ = "rooms"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    created_by: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    members: Mapped[list["RoomMember"]] = relationship(back_populates="room")
    messages: Mapped[list["Message"]] = relationship(back_populates="room")


class RoomMember(Base):
    __tablename__ = "room_members"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    room_id: Mapped[str] = mapped_column(String(36), ForeignKey("rooms.id"), nullable=False)
    user_id: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    joined_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    room: Mapped["Room"] = relationship(back_populates="members")
    user: Mapped["User"] = relationship(back_populates="memberships")


class Message(Base):
    __tablename__ = "messages"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    content: Mapped[str] = mapped_column(Text, nullable=False)
    room_id: Mapped[str] = mapped_column(String(36), ForeignKey("rooms.id"), nullable=False)
    sender_id: Mapped[str] = mapped_column(String(36), ForeignKey("users.id"), nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))

    room: Mapped["Room"] = relationship(back_populates="messages")
    sender: Mapped["User"] = relationship(back_populates="messages")

Initialize Alembic and configure the migration environment

Alembic is a database migration tool that tracks schema changes over time. It compares your Python models against the current database state and generates SQL to bring them in sync.

• Initialize Alembic with the async template by running this command in your terminal:

alembic init -t async alembic

What does this command do?

The -t async flag tells Alembic to generate an env.py configured for async database operations (which matches your async SQLAlchemy setup). The final alembic argument is the directory name where migration files will live.

• Confirm the files were created by running:

ls alembic

You should see env.py, README, script.py.mako, and a versions/ directory. You should also see alembic.ini in your project root.

Command not found?

Make sure your virtual environment is activated. You installed Alembic in Step 1 as part of your dependencies. Run venv\Scripts\activate (Windows) or source venv/bin/activate (macOS/Linux) and try again.

Help me troubleshoot the alembic command not being found in my terminal.

Now you need to configure Alembic to connect to your database and find your models.

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

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

• Replace it with your actual database URL:

sqlalchemy.url = postgresql+asyncpg://postgres:your-password-here@localhost:5432/chatapi

• Save alembic.ini.

Why set the URL here and in .env?

Alembic reads its database URL from alembic.ini when running migrations. Your FastAPI app reads from .env at runtime. Both need to point at the same database.

Next, you need to update alembic/env.py so Alembic can see your models and generate the correct migration.

• Open alembic/env.py in Cursor.
• Replace the entire contents of the file with the following:

import asyncio
from logging.config import fileConfig

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

from alembic import context

from app.database import Base
from app.models import User, Room, RoomMember, Message

config = context.config

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

target_metadata = Base.metadata


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

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


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

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


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

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

    await connectable.dispose()


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


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

What did you change in env.py?

The two critical additions are the imports of Base and all four model classes. Importing the models registers them with Base.metadata, which is what Alembic inspects to know which tables to create.

Setting target_metadata = Base.metadata tells Alembic to compare this metadata against the live database when autogenerating migrations.

• Save alembic/env.py.

Seeing a ModuleNotFoundError?

If you see an error about app not being found, make sure you are running Alembic commands from the fastapi-chat project root (the folder containing both app/ and alembic/).

Help me fix a ModuleNotFoundError when running Alembic commands.

Generate and apply the initial migration

With models defined and Alembic configured, you can now autogenerate a migration script that creates all four tables.

• Generate the migration by running this command:

alembic revision --autogenerate -m "initial tables"

What does this command do?

The --autogenerate flag compares your model metadata against the current database (which is empty). Alembic generates a Python migration file in alembic/versions/ with the SQL needed to create the users, rooms, room_members, and messages tables.

You should see output indicating a new revision was generated. The filename includes a revision hash.

• Apply the migration to your database by running:

alembic upgrade head

You should see SQL statements being executed, ending with the tables being created.

• Verify the tables exist by connecting to PostgreSQL and listing them:

psql -U postgres -d chatapi -c "\dt"

You should see four tables listed: messages, room_members, rooms, and users. You'll also see an alembic_version table that Alembic uses to track which migration has been applied.

Tables not showing up?

Make sure PostgreSQL is running and that the chatapi database exists (you created it in Step 1). Also verify the password in alembic.ini matches the one you set during PostgreSQL installation.

Help me troubleshoot why my Alembic migration did not create tables in PostgreSQL.

Your database schema is in place. Next up, you'll build user registration and JWT authentication on top of these models.

Build User Registration and JWT Authentication

Your database is ready with empty tables waiting for data. Now you need to give users a way to create accounts and prove who they are when making requests.

In this step, you'll build the authentication layer for your chat API. This includes securely hashing passwords with Argon2, issuing JWT tokens on login, and protecting endpoints so only authenticated users can access them.

In this step, get ready to:

• Create Pydantic schemas for request and response validation.
• Build auth utilities for password hashing and JWT token management.
• Wire up registration, login, and the FastAPI app with CORS.

Create the auth utilities

Before building endpoints, you need the foundation: functions to hash passwords, create tokens, and verify a user's identity from a token.

• In Cursor, open app/schemas.py and add the following Pydantic models that define how request and response data is shaped:

from datetime import datetime

from pydantic import BaseModel, EmailStr


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


class UserResponse(BaseModel):
    id: str
    username: str
    email: str
    created_at: datetime

    class Config:
        from_attributes = True


class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"


class RoomCreate(BaseModel):
    name: str


class RoomResponse(BaseModel):
    id: str
    name: str
    created_by: str
    created_at: datetime

    class Config:
        from_attributes = True


class MessageResponse(BaseModel):
    id: str
    content: str
    room_id: str
    sender_id: str
    created_at: datetime

    class Config:
        from_attributes = True


class WebSocketMessage(BaseModel):
    type: str = "message"
    content: str

What does this code do?

Each class defines the shape of data flowing in or out of the API. UserCreate is what a user sends to register. UserResponse is what the API sends back (notice password is excluded for security).

The from_attributes = True config lets Pydantic read data directly from SQLAlchemy model instances instead of plain dictionaries. You'll use the room and message schemas in later steps.

• Save app/schemas.py.

Now open app/auth.py to build the password hashing and token verification utilities. You'll add this file in three chunks.

• Add the password hashing setup and helper functions to app/auth.py:

from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import get_db
from app.models import User

# Set up Argon2 password hashing (industry standard)
password_hash = PasswordHash.recommended()

# Tell FastAPI where the login endpoint is for the OAuth2 flow
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")


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


def verify_password(plain_password: str, hashed: str) -> bool:
    return password_hash.verify(plain_password, hashed)

What does this code do?

PasswordHash.recommended() creates an Argon2 hasher. Argon2 is the current best practice for password storage because it is resistant to GPU-based brute-force attacks.

OAuth2PasswordBearer tells FastAPI that your login endpoint lives at auth/token. FastAPI uses this to add an "Authorize" button to the Swagger UI so you can test protected endpoints.

• Add the JWT token creation function below verify_password:

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

What does this code do?

This function takes a dictionary of data (like the username), adds an expiration timestamp, and encodes it into a signed JWT string using the HS256 algorithm. The token expires after 30 minutes by default (from your .env settings).

Anyone with the SECRET_KEY can verify the token is legitimate. The expiration prevents stolen tokens from working forever.

• Add the get_current_user dependency below create_access_token:

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception

    result = await db.execute(select(User).where(User.username == username))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

What does this code do?

This is a FastAPI dependency. Any endpoint that includes current_user: User = Depends(get_current_user) will automatically extract and validate the JWT from the request header before running endpoint logic.

It decodes the token, pulls the username from the sub claim, and looks up the user in the database. If anything fails, the user gets a 401 Unauthorized response.

• Save app/auth.py.

Seeing import errors in your editor?

If Cursor underlines pwdlib or jwt with a red squiggle, make sure your virtual environment is activated and selected as the Python interpreter in Cursor. Check the bottom status bar to confirm it shows the venv path.

help me fix import errors for pwdlib or jwt in my auth.py file

✔️ Awesome, I've got everything!

Great. Double check you've saved both app/schemas.py and app/auth.py.

ⓧ I'd like to double check the full code

Here is the complete app/auth.py for comparison:

from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import get_db
from app.models import User

password_hash = PasswordHash.recommended()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")


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


def verify_password(plain_password: str, hashed: str) -> bool:
    return password_hash.verify(plain_password, hashed)


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception

    result = await db.execute(select(User).where(User.username == username))
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

Build the registration and login endpoints

With the auth utilities ready, you can now build the actual endpoints that users will call to create an account and get a token.

• Open app/routers/users.py and add the registration endpoint:

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

from app.auth import hash_password, verify_password, create_access_token
from app.database import get_db
from app.models import User
from app.schemas import UserCreate, UserResponse, Token

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


@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: AsyncSession = Depends(get_db)):
    # Check if username or email already exists
    result = await db.execute(
        select(User).where((User.username == user_data.username) | (User.email == user_data.email))
    )
    existing_user = result.scalar_one_or_none()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )

    # Create new user with hashed password
    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user

What does this code do?

The register endpoint accepts a username, email, and password. It first checks the database for duplicates, then hashes the password with Argon2 before storing anything. The plain-text password is never saved.

The response_model=UserResponse ensures the response only includes safe fields (id, username, email, created_at). FastAPI automatically strips out the hashed_password field.

• Add the login endpoint below the register function:

@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    # Find user by username
    result = await db.execute(select(User).where(User.username == form_data.username))
    user = result.scalar_one_or_none()

    # Verify password matches
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    # Create and return JWT access token
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token)

What does this code do?

The login endpoint uses OAuth2PasswordRequestForm which expects username and password as form fields. This format is required by the OAuth2 spec and enables the Swagger UI "Authorize" button to work.

If credentials are valid, it encodes the username as the sub (subject) claim in the JWT. The client sends this token back in the Authorization: Bearer <token> header for all future requests.

• Save app/routers/users.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved app/routers/users.py.

ⓧ I'd like to double check the full code

Here is the complete app/routers/users.py for comparison:

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

from app.auth import hash_password, verify_password, create_access_token
from app.database import get_db
from app.models import User
from app.schemas import UserCreate, UserResponse, Token

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


@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user_data: UserCreate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(User).where((User.username == user_data.username) | (User.email == user_data.email))
    )
    existing_user = result.scalar_one_or_none()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username or email already registered",
        )

    new_user = User(
        username=user_data.username,
        email=user_data.email,
        hashed_password=hash_password(user_data.password),
    )
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return new_user


@router.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.username == form_data.username))
    user = result.scalar_one_or_none()

    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token)

Wire up the FastAPI app and test

The endpoints exist but FastAPI doesn't know about them yet. You need to register the router and add CORS middleware so browsers can talk to the API.

• Open app/main.py and add the application setup:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.routers import users, rooms, chat


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup logic goes here (e.g. connection pools)
    yield
    # Shutdown logic goes here


app = FastAPI(
    title="Chat API",
    description="A real-time chat application built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

# Allow requests from any origin (fine for development)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(users.router)


@app.get("/")
async def root():
    return {"message": "Chat API is running"}

What does this code do?

The lifespan context manager lets you run code at startup and shutdown. For now it's empty, but it's the modern replacement for FastAPI's deprecated on_startup events.

CORSMiddleware with allow_origins=["*"] means any frontend domain can call your API. This is intentionally permissive for development. The import of rooms and chat modules prepares for later steps. They exist as empty files so the import succeeds.

• Save app/main.py.
• Start the development server by running:

fastapi dev app/main.py

• Open your browser and navigate to http://127.0.0.1:8000/docs.

You should see the Swagger UI with your API documentation, including the POST /auth/register and POST /auth/token endpoints listed under the "auth" tag.

• Test registration by expanding POST /auth/register, clicking Try it out, and entering a JSON body like this:

{
  "username": "testuser",
  "email": "test@example.com",
  "password": "mypassword123"
}

• Click Execute. You should see a 201 response with the new user's details (no password in the response).
• Test login by expanding POST /auth/token, clicking Try it out, entering testuser as username and mypassword123 as password.
• Click Execute. You should see a 200 response containing an access_token string.

Server won't start or endpoints return 500?

Make sure PostgreSQL is running and the DATABASE_URL in your .env file matches your actual PostgreSQL password. Also confirm your virtual environment is activated before running fastapi dev.

my FastAPI server won't start or I'm getting 500 errors on the auth endpoints

✔️ Awesome, I've got everything!

Great. Double check you've saved app/main.py.

ⓧ I'd like to double check the full code

Here is the complete app/main.py for comparison:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.routers import users, rooms, chat


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


app = FastAPI(
    title="Chat API",
    description="A real-time chat application built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(users.router)


@app.get("/")
async def root():
    return {"message": "Chat API is running"}

Your API now handles user registration and login with secure password hashing and JWT tokens. Next up, you'll build the chat room endpoints so authenticated users can create and join conversations.

Build Chat Room Endpoints

Your authentication system is in place. Users can register and log in. Now it's time to give them something to do once they're authenticated.

In this step, you'll build the REST endpoints that let users create chat rooms, join existing rooms, and retrieve message history. Every endpoint will be protected by the JWT authentication you built in the previous step.

In this step, get ready to:

• Create room management endpoints for creating, listing, and joining rooms.
• Add a paginated message history endpoint with membership checks.
• Wire up the rooms router and test everything via Swagger UI.

Create REST endpoints for room management

Chat rooms are the containers where conversations happen. You need endpoints to create a room, see which rooms you belong to, and join rooms that others have created.

• Open app/routers/rooms.py in Cursor.
• Add the imports, router definition, and the create room endpoint by pasting this code:

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

from app.auth import get_current_user
from app.database import get_db
from app.models import User, Room, RoomMember, Message
from app.schemas import RoomCreate, RoomResponse, MessageResponse

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


@router.post("/", response_model=RoomResponse, status_code=status.HTTP_201_CREATED)
async def create_room(
    room_data: RoomCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    new_room = Room(name=room_data.name, created_by=current_user.id)
    db.add(new_room)
    await db.flush()

    membership = RoomMember(room_id=new_room.id, user_id=current_user.id)
    db.add(membership)
    await db.commit()
    await db.refresh(new_room)
    return new_room

What does this code do?

The router is prefixed with /rooms, so all endpoints in this file will start with that path.

The create_room endpoint does two things in one transaction: it creates the room, then automatically adds the creator as a member. The await db.flush() call generates the room's ID without committing, so you can use it immediately for the membership record.

You won't see a change yet. The next chunk adds the list and join endpoints that complete this file.

• Add the endpoint for listing rooms the user belongs to, directly below the create_room function:

@router.get("/", response_model=list[RoomResponse])
async def list_rooms(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Room)
        .join(RoomMember, RoomMember.room_id == Room.id)
        .where(RoomMember.user_id == current_user.id)
    )
    rooms = result.scalars().all()
    return rooms

What does this code do?

This endpoint uses a SQL JOIN to find only rooms where the current user has a membership record. It doesn't show all rooms in the system. It shows only rooms the user has created or explicitly joined.

You won't see a change yet. The next chunk adds the join endpoint.

• Add the join room endpoint directly below the list_rooms function:

@router.post("/{room_id}/join", status_code=status.HTTP_200_OK)
async def join_room(
    room_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(Room).where(Room.id == room_id))
    room = result.scalar_one_or_none()
    if not room:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Room not found")

    result = await db.execute(
        select(RoomMember).where(
            RoomMember.room_id == room_id, RoomMember.user_id == current_user.id
        )
    )
    existing = result.scalar_one_or_none()
    if existing:
        return {"message": "Already a member"}

    membership = RoomMember(room_id=room_id, user_id=current_user.id)
    db.add(membership)
    await db.commit()
    return {"message": "Joined room successfully"}

What does this code do?

The join endpoint has three safety checks. First, it verifies the room actually exists (returns 404 if not). Second, it checks whether the user is already a member (returns a friendly message instead of duplicating the record). Only then does it create the membership.

• Save app/routers/rooms.py.

You won't see a visible change in Swagger yet because the router isn't included in the app. You'll wire it up after adding the message history endpoint.

Seeing a syntax error on save?

• Check that each function has async def (not just def).
• Make sure each code chunk is pasted in order with no overlapping lines.
• Verify the imports at the top match exactly. Query is imported from fastapi even though it's used in the next substep.

Help me debug a syntax error in my rooms.py router file.

Add the message history endpoint

Users need to see past messages when they open a room. This endpoint returns messages in reverse chronological order with pagination so clients can load older history on demand.

• In app/routers/rooms.py, add the message history endpoint directly below the join_room function:

@router.get("/{room_id}/messages", response_model=list[MessageResponse])
async def get_messages(
    room_id: str,
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=50, ge=1, le=100),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Check that the user is a member of this room
    result = await db.execute(
        select(RoomMember).where(
            RoomMember.room_id == room_id, RoomMember.user_id == current_user.id
        )
    )
    if not result.scalar_one_or_none():
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not a member of this room")

    # Fetch messages ordered newest first, with pagination
    result = await db.execute(
        select(Message)
        .where(Message.room_id == room_id)
        .order_by(Message.created_at.desc())
        .offset(skip)
        .limit(limit)
    )
    messages = result.scalars().all()
    return messages

What does this code do?

Before returning any messages, this endpoint checks that the requesting user is actually a member of the room. Non-members get a 403 Forbidden response.

The skip and limit query parameters enable pagination. By default, the endpoint returns the 50 most recent messages. Clients can request older messages by increasing skip.

• Save app/routers/rooms.py.

Seeing a red underline on Query?

Make sure your import line at the top reads from fastapi import APIRouter, Depends, HTTPException, status, Query. The Query import was included in the first chunk for exactly this reason.

Help me fix an import issue with Query in my FastAPI rooms router.

✔️ Awesome, I've got everything!

Great. Double check you've saved app/routers/rooms.py before moving on.

ⓧ I'd like to double check the full code

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

from app.auth import get_current_user
from app.database import get_db
from app.models import User, Room, RoomMember, Message
from app.schemas import RoomCreate, RoomResponse, MessageResponse

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


@router.post("/", response_model=RoomResponse, status_code=status.HTTP_201_CREATED)
async def create_room(
    room_data: RoomCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    new_room = Room(name=room_data.name, created_by=current_user.id)
    db.add(new_room)
    await db.flush()

    membership = RoomMember(room_id=new_room.id, user_id=current_user.id)
    db.add(membership)
    await db.commit()
    await db.refresh(new_room)
    return new_room


@router.get("/", response_model=list[RoomResponse])
async def list_rooms(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Room)
        .join(RoomMember, RoomMember.room_id == Room.id)
        .where(RoomMember.user_id == current_user.id)
    )
    rooms = result.scalars().all()
    return rooms


@router.post("/{room_id}/join", status_code=status.HTTP_200_OK)
async def join_room(
    room_id: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(Room).where(Room.id == room_id))
    room = result.scalar_one_or_none()
    if not room:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Room not found")

    result = await db.execute(
        select(RoomMember).where(
            RoomMember.room_id == room_id, RoomMember.user_id == current_user.id
        )
    )
    existing = result.scalar_one_or_none()
    if existing:
        return {"message": "Already a member"}

    membership = RoomMember(room_id=room_id, user_id=current_user.id)
    db.add(membership)
    await db.commit()
    return {"message": "Joined room successfully"}


@router.get("/{room_id}/messages", response_model=list[MessageResponse])
async def get_messages(
    room_id: str,
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=50, ge=1, le=100),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(RoomMember).where(
            RoomMember.room_id == room_id, RoomMember.user_id == current_user.id
        )
    )
    if not result.scalar_one_or_none():
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not a member of this room")

    result = await db.execute(
        select(Message)
        .where(Message.room_id == room_id)
        .order_by(Message.created_at.desc())
        .offset(skip)
        .limit(limit)
    )
    messages = result.scalars().all()
    return messages

Include the rooms router and test via Swagger UI

The rooms endpoints are defined but the app doesn't know about them yet. You need to register the router in your main application file.

• In app/main.py, find this line:

app.include_router(users.router)

• Add the rooms router directly below it, so the two lines look like this:

app.include_router(users.router)
app.include_router(rooms.router)

• Save app/main.py.

The dev server should auto-reload. You should see new room-related routes in the terminal output.

Server not reloading?

• Check your terminal for import errors. If app/routers/rooms.py has a syntax error, the server will fail to start.
• Make sure rooms is already imported at the top of app/main.py in the line from app.routers import users, rooms, chat.

Help me debug why my FastAPI server won't start after adding the rooms router.

Now test the full flow in Swagger UI.

• Return to your browser at http://127.0.0.1:8000/docs and refresh the page.
• You should see a new rooms section in the docs with four endpoints: POST /rooms/, GET /rooms/, POST /rooms/{room_id}/join, and GET /rooms/{room_id}/messages.
• Click the Authorize button at the top of the page.
• Enter the username and password you registered earlier, then click Authorize.
• Expand POST /rooms/, click Try it out, enter {"name": "general"} as the request body, and click Execute.
• You should see a 201 response with the room details including an id field. Copy that room ID.
• Expand GET /rooms/{room_id}/messages, click Try it out, paste the room ID, and click Execute.
• You should see a 200 response with an empty array []. No messages exist yet, but the endpoint works.

Why does Swagger UI have an Authorize button?

FastAPI auto-detects the OAuth2PasswordBearer dependency in your endpoints and adds a built-in authorization flow to the docs. When you authorize in Swagger, it calls your /auth/token endpoint behind the scenes and attaches the JWT to all subsequent requests.

Getting a 401 Unauthorized response?

• Make sure you clicked Authorize at the top of the docs page and entered valid credentials before trying the rooms endpoints.
• If your token has expired (after 30 minutes), close the authorize dialog and re-authorize with the same credentials.

Help me fix a 401 error when testing my rooms endpoints in Swagger UI.

✔️ Awesome, I've got everything!

Great. Double check you've saved both files before moving on.

ⓧ I'd like to double check the full code

Here is the complete app/main.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.routers import users, rooms, chat


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


app = FastAPI(
    title="Chat API",
    description="A real-time chat application built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(users.router)
app.include_router(rooms.router)


@app.get("/")
async def root():
    return {"message": "Chat API is running"}

Your chat rooms are ready. Users can create rooms, join them, and query message history. Next up, you'll build the WebSocket endpoint that makes real-time messaging possible.

Build the Real-Time WebSocket Chat

Your chat rooms are set up with REST endpoints for creating, joining, and fetching message history. Now it is time to make messages flow in real time.

In this step, you will build a WebSocket endpoint that keeps a persistent connection open between each user and the server. Instead of polling for new messages, the server pushes them instantly to every connected client in the room.

In this step, get ready to:

• Create a ConnectionManager class that tracks active connections per room.
• Build a WebSocket endpoint with JWT authentication and room membership verification.
• Persist messages to PostgreSQL and broadcast them in real time.

Create the ConnectionManager class

A WebSocket server needs to know which connections belong to which room so it can broadcast a message to the right audience. The ConnectionManager class handles this bookkeeping.

• Open app/routers/chat.py in Cursor and replace any placeholder content with the following code:

import json

import jwt
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends
from jwt.exceptions import InvalidTokenError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import async_session
from app.models import User, RoomMember, Message

router = APIRouter(tags=["chat"])


class ConnectionManager:
    def __init__(self):
        # Maps room_id to a list of (websocket, user_id) tuples
        self.active_connections: dict[str, list[tuple[WebSocket, str]]] = {}

    async def connect(self, websocket: WebSocket, room_id: str, user_id: str):
        # Accept the WebSocket handshake and register the connection
        await websocket.accept()
        if room_id not in self.active_connections:
            self.active_connections[room_id] = []
        self.active_connections[room_id].append((websocket, user_id))

    def disconnect(self, websocket: WebSocket, room_id: str):
        # Remove a connection from the room's list
        if room_id in self.active_connections:
            self.active_connections[room_id] = [
                (ws, uid) for ws, uid in self.active_connections[room_id] if ws != websocket
            ]
            # Clean up empty rooms
            if not self.active_connections[room_id]:
                del self.active_connections[room_id]

    async def broadcast(self, room_id: str, message: dict, exclude_websocket: WebSocket | None = None):
        # Send a message to all connections in a room
        if room_id in self.active_connections:
            disconnected = []
            for websocket, user_id in self.active_connections[room_id]:
                if websocket == exclude_websocket:
                    continue
                try:
                    await websocket.send_json(message)
                except Exception:
                    disconnected.append(websocket)
            # Clean up any broken connections
            for ws in disconnected:
                self.disconnect(ws, room_id)


manager = ConnectionManager()

What does this code do?

The ConnectionManager stores a dictionary where each key is a room_id and each value is a list of active WebSocket connections (paired with their user IDs).

The connect() method accepts the upgrade handshake and registers the socket. The broadcast() method loops through all connections in a room and sends them a JSON message, with an option to exclude the sender.

Why not use FastAPI's dependency system for WebSocket auth?

Browsers cannot set custom HTTP headers (like Authorization: Bearer) on WebSocket upgrade requests. The standard workaround is to pass the JWT token as a query parameter instead. This means you cannot reuse the get_current_user dependency directly and need a separate helper.

• Save app/routers/chat.py.

You won't see a change yet. The next chunk adds the WebSocket endpoint that uses this manager.

Build the WebSocket endpoint with JWT authentication

The endpoint needs to authenticate the connecting user, verify they belong to the requested room, then enter a loop that receives messages and broadcasts them.

• In app/routers/chat.py, add the following code below manager = ConnectionManager():

async def get_user_from_token(token: str, db: AsyncSession) -> User | None:
    # Decode the JWT and look up the user in the database
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            return None
    except InvalidTokenError:
        return None

    result = await db.execute(select(User).where(User.username == username))
    return result.scalar_one_or_none()


@router.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    # Extract the token from the query string
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=4001, reason="Missing token")
        return

    # Authenticate the user and check room membership
    async with async_session() as db:
        user = await get_user_from_token(token, db)
        if not user:
            await websocket.close(code=4001, reason="Invalid token")
            return

        result = await db.execute(
            select(RoomMember).where(
                RoomMember.room_id == room_id, RoomMember.user_id == user.id
            )
        )
        if not result.scalar_one_or_none():
            await websocket.close(code=4003, reason="Not a member of this room")
            return

    # Register the connection and notify the room
    await manager.connect(websocket, room_id, user.id)

    await manager.broadcast(
        room_id,
        {"type": "system", "content": f"{user.username} joined the chat"},
        exclude_websocket=websocket,
    )

    # Main receive/broadcast loop with message persistence
    try:
        while True:
            data = await websocket.receive_text()
            message_data = json.loads(data)
            content = message_data.get("content", "")

            if not content.strip():
                continue

            # Persist the message to the database before broadcasting
            async with async_session() as db:
                new_message = Message(
                    content=content,
                    room_id=room_id,
                    sender_id=user.id,
                )
                db.add(new_message)
                await db.commit()
                await db.refresh(new_message)

                broadcast_data = {
                    "type": "message",
                    "content": content,
                    "sender_id": user.id,
                    "sender_username": user.username,
                    "message_id": new_message.id,
                    "created_at": new_message.created_at.isoformat(),
                }

            await manager.broadcast(room_id, broadcast_data)

    except WebSocketDisconnect:
        manager.disconnect(websocket, room_id)
        await manager.broadcast(
            room_id,
            {"type": "system", "content": f"{user.username} left the chat"},
        )

What does this code do?

The get_user_from_token() helper decodes the JWT from the query string and looks up the user. It returns None if the token is invalid or expired.

The websocket_endpoint function runs through three phases: authentication (reject if no token or invalid token), membership check (reject if user is not in the room), and then the main loop. Inside the loop, each incoming message is saved to PostgreSQL first, then broadcast to all room members.

When a client disconnects, the WebSocketDisconnect exception fires. The handler removes the connection and broadcasts a "left the chat" system message.

Why persist before broadcasting?

Saving the message to the database before broadcasting guarantees that the message history endpoint always returns everything that was ever sent. If the server crashes after broadcasting but before saving, the message is lost from history.

• Save app/routers/chat.py.

Seeing a syntax error?

Make sure the get_user_from_token function and the websocket_endpoint function are at the module level (no extra indentation), not nested inside the ConnectionManager class.

help me fix a syntax error in my chat.py WebSocket endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved app/routers/chat.py before moving on.

ⓧ I'd like to double check the full code

import json

import jwt
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends
from jwt.exceptions import InvalidTokenError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import async_session
from app.models import User, RoomMember, Message

router = APIRouter(tags=["chat"])


class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, list[tuple[WebSocket, str]]] = {}

    async def connect(self, websocket: WebSocket, room_id: str, user_id: str):
        await websocket.accept()
        if room_id not in self.active_connections:
            self.active_connections[room_id] = []
        self.active_connections[room_id].append((websocket, user_id))

    def disconnect(self, websocket: WebSocket, room_id: str):
        if room_id in self.active_connections:
            self.active_connections[room_id] = [
                (ws, uid) for ws, uid in self.active_connections[room_id] if ws != websocket
            ]
            if not self.active_connections[room_id]:
                del self.active_connections[room_id]

    async def broadcast(self, room_id: str, message: dict, exclude_websocket: WebSocket | None = None):
        if room_id in self.active_connections:
            disconnected = []
            for websocket, user_id in self.active_connections[room_id]:
                if websocket == exclude_websocket:
                    continue
                try:
                    await websocket.send_json(message)
                except Exception:
                    disconnected.append(websocket)
            for ws in disconnected:
                self.disconnect(ws, room_id)


manager = ConnectionManager()


async def get_user_from_token(token: str, db: AsyncSession) -> User | None:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            return None
    except InvalidTokenError:
        return None

    result = await db.execute(select(User).where(User.username == username))
    return result.scalar_one_or_none()


@router.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: str):
    token = websocket.query_params.get("token")
    if not token:
        await websocket.close(code=4001, reason="Missing token")
        return

    async with async_session() as db:
        user = await get_user_from_token(token, db)
        if not user:
            await websocket.close(code=4001, reason="Invalid token")
            return

        result = await db.execute(
            select(RoomMember).where(
                RoomMember.room_id == room_id, RoomMember.user_id == user.id
            )
        )
        if not result.scalar_one_or_none():
            await websocket.close(code=4003, reason="Not a member of this room")
            return

    await manager.connect(websocket, room_id, user.id)

    await manager.broadcast(
        room_id,
        {"type": "system", "content": f"{user.username} joined the chat"},
        exclude_websocket=websocket,
    )

    try:
        while True:
            data = await websocket.receive_text()
            message_data = json.loads(data)
            content = message_data.get("content", "")

            if not content.strip():
                continue

            async with async_session() as db:
                new_message = Message(
                    content=content,
                    room_id=room_id,
                    sender_id=user.id,
                )
                db.add(new_message)
                await db.commit()
                await db.refresh(new_message)

                broadcast_data = {
                    "type": "message",
                    "content": content,
                    "sender_id": user.id,
                    "sender_username": user.username,
                    "message_id": new_message.id,
                    "created_at": new_message.created_at.isoformat(),
                }

            await manager.broadcast(room_id, broadcast_data)

    except WebSocketDisconnect:
        manager.disconnect(websocket, room_id)
        await manager.broadcast(
            room_id,
            {"type": "system", "content": f"{user.username} left the chat"},
        )

Include the chat router and test real-time messaging

The WebSocket endpoint exists but the FastAPI app does not serve it yet. You need to register the chat router in app/main.py.

• In app/main.py, find this line:

app.include_router(rooms.router)

• Add the following line directly below it:

app.include_router(chat.router)

What does this do?

This registers the /ws/{room_id} WebSocket route on the FastAPI app. Without this line, the endpoint exists in code but is not reachable from clients.

• Save app/main.py. Your dev server restarts automatically.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py before testing.

ⓧ I'd like to double check the full code

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.routers import users, rooms, chat


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


app = FastAPI(
    title="Chat API",
    description="A real-time chat application built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(users.router)
app.include_router(rooms.router)
app.include_router(chat.router)


@app.get("/")
async def root():
    return {"message": "Chat API is running"}

Now test the WebSocket endpoint with two browser tabs to confirm real-time messaging works.

• Open your browser and navigate to http://127.0.0.1:8000/docs.
• Use the POST /auth/register endpoint to create two test users (e.g. alice and bob).
• Use POST /auth/token to log in as alice and copy the access_token value.
• Authorize in Swagger UI with the token, then create a room via POST /rooms and note the room id from the response.
• Log in as bob, copy that token, and use POST /rooms/{room_id}/join to join the room.
• Open two browser tabs and open the developer console in each (F12 or Cmd+Option+J on macOS).
• In tab 1 (alice), paste the following into the console, replacing the room ID and token:

// Connect as alice
const ws = new WebSocket("ws://127.0.0.1:8000/ws/[[ROOM_ID="your room id from earlier"]]
?token=[[ALICE_TOKEN="alice's access token"]]");
ws.onmessage = (e) => console.log(JSON.parse(e.data));

• In tab 2 (bob), paste the same snippet with bob's token.
• In either tab, send a message by running:

ws.send(JSON.stringify({"content": "Hello from the WebSocket!"}));

• Check the other tab's console. You should see the message appear as a JSON object with type, content, sender_username, and created_at fields.
• Verify persistence by calling GET /rooms/{room_id}/messages in Swagger UI. The message you just sent should appear in the history.

WebSocket closes immediately?

Check that you copied the full access_token value (not the surrounding quotes). If the token expired, log in again with POST /auth/token to get a fresh one.

Also confirm the room ID exists and the user has joined it. A close code of 4003 means the user is not a member of that room.

help me debug why my WebSocket connection closes immediately

Your chat API is fully functional. Users can register, log in, create rooms, and exchange messages in real time. Every message is persisted to PostgreSQL so history is never lost. Next up, a secret mission to add typing indicators and online presence tracking.

Secret mission

Add Typing Indicators and Online Presence

Your chat API delivers messages in real time, but real chat apps show more than just text. Users expect to see who's online and when someone is typing. Can you extend the WebSocket layer to broadcast presence and typing events?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, 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:

• PostgreSQL chatapi database (stores users, rooms, and messages).
• fastapi-chat project folder on your Desktop (contains all application code, virtual environment, and Alembic migrations).
• FastAPI development server (running process in your terminal).

Keep everything running

No action needed. Choose this if you are still actively building on this project or want to experiment with the API further.

• Your FastAPI dev server stays running in the terminal. You can stop and restart it anytime with fastapi dev app/main.py.
• The chatapi database persists in PostgreSQL and is available whenever the PostgreSQL service is running.
• PostgreSQL itself stays installed on your machine. It uses minimal resources when idle.

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.

• Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
• Deactivate the Python virtual environment by running:

deactivate

• Your chatapi database and fastapi-chat folder remain on your machine, ready when you come back.
• To resume later, open your terminal, navigate to the fastapi-chat folder, activate the virtual environment, and start the dev server again.

Delete - I don't want to use this again

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

• Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it is running.
• Deactivate the virtual environment by running:

deactivate

• Drop the chatapi database. Open a new terminal and run:

psql -U postgres

Then inside the psql prompt, run:

DROP DATABASE chatapi;

• Exit psql by typing \q and pressing Enter.
• Delete the project folder. On macOS or Linux, run:

rm -rf ~/Desktop/fastapi-chat

• On Windows, run:

rmdir /s /q %USERPROFILE%\Desktop\fastapi-chat

• This removes all application code, the virtual environment, and all installed Python packages in one go.

Nice Work!

Nice Work!

Nice work! You just built a production-style real-time chat API from scratch using FastAPI, PostgreSQL, and WebSockets.

You've learned how to:

• Implemented secure user authentication with Argon2 password hashing and JWT tokens to protect API endpoints.
• Designed and migrated a relational PostgreSQL database schema for users, rooms, and messages using SQLAlchemy and Alembic.
• Built a WebSocket system with a connection manager for real-time message delivery across multiple users in chat rooms.
• Added typing indicators and online presence tracking to the WebSocket layer for a production-level chat experience.

Ready to quiz yourself?