Setting Up a WebSocket Server with FastAPI

Overview

In this chapter, we will set up a basic WebSocket server using FastAPI, a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. FastAPI is easy to use and provides a straightforward way to create a WebSocket endpoint.

Installing FastAPI and Uvicorn

To get started, you need to install FastAPI and Uvicorn. Uvicorn is a lightning-fast ASGI server implementation, using `uvloop` and `httptools`.

Step 1: Install FastAPI and Uvicorn

pip install fastapi uvicorn

This command will install both FastAPI and Uvicorn, which we will use to run our WebSocket server.

Creating the WebSocket Server

Now that we have FastAPI and Uvicorn installed, let's create a basic WebSocket server.

Step 2: Create the Project Structure

Create a new directory for your project and navigate into it. Inside this directory, create a file named main.py:

mkdir websocket-server
cd websocket-server
touch main.py

Step 3: Implement the WebSocket Endpoint

Open main.py and add the following code to create a simple WebSocket endpoint:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

Let's break down this code:

• We import FastAPI and WebSocket from the fastapi module.
• We create an instance of the FastAPI class.
• We define a WebSocket endpoint at the path /ws using the @app.websocket decorator.
• Within the endpoint function, we accept the WebSocket connection using await websocket.accept().
• We enter a while loop to continuously receive and send messages. When a message is received using await websocket.receive_text(), we send back a response with the received message using await websocket.send_text().

Step 4: Run the WebSocket Server

To run the server, use the following command:

uvicorn main:app --reload

This command tells Uvicorn to run the app instance defined in the main module. The --reload flag enables auto-reloading, so the server restarts whenever you make changes to the code.

Testing the WebSocket Server

With the WebSocket server running, you can test it using a WebSocket client. You can use a WebSocket client library in Python or a browser-based client like WebSocket Echo Test.

Using a Browser-Based Client

Navigate to WebSocket Echo Test and connect to your server by entering ws://localhost:8000/ws in the "Location" field and clicking "Connect". You can then send messages and see the responses from the server.

Using a Python Client

Create a new Python script named client.py and add the following code:

import asyncio
import websockets

async def hello():
    uri = "ws://localhost:8000/ws"
    async with websockets.connect(uri) as websocket:
        await websocket.send("Hello, WebSocket!")
        response = await websocket.recv()
        print(f"< {response}")

asyncio.get_event_loop().run_until_complete(hello())

This script uses the websockets library to connect to the WebSocket server, send a message, and print the response.

Conclusion

In this chapter, we have set up a basic WebSocket server using FastAPI and tested it with both a browser-based client and a Python client. This server can now receive and send messages in real-time, providing a foundation for building more complex WebSocket applications.

In the next chapter, we will create a WebSocket client using JavaScript and integrate it with our FastAPI server.

When you're ready, say "Next" to proceed to Chapter 3.