Build a Real-Time Chat App with FastAPI and WebSockets

tags: python, fastapi, websocket, tutorial

---

Building a Real-Time Chat App with FastAPI and WebSockets: A Step-by-Step Guide

As developers, we've all been there - we see a demo of a real-time chat app and think, "That's exactly what I need for my project." But, we all know that building such an app from scratch can be a daunting task. You need to worry about scalability, performance, and most importantly, real-time updates. In this post, we'll show you how to build a real-time chat app using FastAPI and WebSockets.

Setting Up the Project

Before we dive into the nitty-gritty of building a chat app, let's set up the project structure. FastAPI is a modern Python web framework that allows us to build high-performance APIs quickly. We'll create a new project using the following command:

mkdir real-time-chat-app
cd real-time-chat-app
mkdir app models routes utils

Our project structure will look like this:

real-time-chat-app/
app/
__init__.py
main.py
models/
__init__.py
user.py
message.py
routes/
__init__.py
auth.py
chat.py
utils/
__init__.py
websocket.py
requirements.txt

Next, let's install the required dependencies:

pip install fastapi uvicorn aiomysql

Defining the Data Models

In our chat app, we'll have two main data models: User and Message. We'll define these models in the models directory:

# app/models/user.py
from pydantic import BaseModel

class User(BaseModel):
    id: int
    username: str
    password: str

# app/models/message.py
from pydantic import BaseModel
from datetime import datetime

class Message(BaseModel):
    id: int
    content: str
    sender_id: int
    receiver_id: int
    timestamp: datetime

Setting Up the Database

For this example, we'll use a MySQL database. You can use any other database engine if you prefer. We'll use the aiomysql library to connect to the database.

First, let's create the database schema:

CREATE TABLE users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    username VARCHAR(255),
    password VARCHAR(255)
);

CREATE TABLE messages (
    id INT PRIMARY KEY AUTO_INCREMENT,
    content VARCHAR(255),
    sender_id INT,
    receiver_id INT,
    timestamp DATETIME,
    FOREIGN KEY (sender_id) REFERENCES users(id),
    FOREIGN KEY (receiver_id) REFERENCES users(id)
);

Next, let's create a database connection using aiomysql:

# app/utils/db.py
import aiomysql

async def get_db():
    db = await aiomysql.connect(
        host="localhost",
        user="root",
        password="password",
        db="real_time_chat"
    )
    return db

Building the Chat API

Now that we have our data models and database connection set up, let's build the chat API. We'll create a new route for sending messages and another for retrieving messages.

First, let's create a message sending route:

# app/routes/chat.py
from fastapi import APIRouter, HTTPException
from app.models.message import Message
from app.utils.websocket import send_message_to_all

router = APIRouter()

@router.post("/messages/")
async def create_message(message: Message):
    db = await get_db()
    async with db.cursor() as cur:
        await cur.execute("INSERT INTO messages (content, sender_id, receiver_id) VALUES (%s, %s, %s)", (message.content, message.sender_id, message.receiver_id))
        await db.commit()
    await send_message_to_all(message)
    return {"message": "Message sent successfully"}

Next, let's create a route for retrieving messages:

# app/routes/chat.py
from fastapi import APIRouter
from app.models.message import Message

router = APIRouter()

@router.get("/messages/")
async def get_messages():
    db = await get_db()
    async with db.cursor() as cur:
        await cur.execute("SELECT * FROM messages")
        messages = await cur.fetchall()
    return [{"id": message[0], "content": message[1], "sender_id": message[2], "receiver_id": message[3], "timestamp": message[4]} for message in messages]

Setting Up WebSockets

Now that we have our chat API set up, let's set up WebSockets. WebSockets allow us to establish a persistent connection between the client and server, enabling real-time updates.

First, let's create a new file for handling WebSocket connections:

# app/utils/websocket.py
import asyncio
from fastapi import WebSocket

async def send_message_to_all(message: Message):
    await asyncio.wait([ws.send_json({"type": "message", "content": message.content}) for ws in await WebSocket.get_running_connections()])

async def handle_connection(websocket: WebSocket):
    await websocket.accept()
    while True:
        message = await websocket.receive_json()
        if message["type"] == "message":
            await send_message_to_all(message["content"])
        elif message["type"] == "disconnect":
            break

Finally, let's create a new route for handling WebSocket connections:

# app/routes/chat.py
from fastapi import APIRouter, WebSocket

router = APIRouter()

@router.get("/ws/")
async def get_websocket():
    return {"type": "websocket"}

Running the App

Now that we have our chat app set up, let's run it:

uvicorn app.main:app --host 0.0.0.0 --port 8000

Conclusion

In this post, we showed you how to build a real-time chat app using FastAPI and WebSockets. With this guide, you should be able to build a chat app that supports real-time updates and persistent connections.

Remember to replace the database connection settings and WebSocket connection settings with your own. With this app, you can create a real-time chat experience for your users.

What's next? You can extend this app to support file uploads, user authentication, and more. The possibilities are endless. Happy coding!

---

If you found this helpful, consider buying me a coffee ☕ — it keeps these articles coming!

Also check out my AI tools collection: AI 次元世界 — free AI tools for developers.