WebSocket authentication in FastAPI - Mini Project: Build & Apply

WebSocket Authentication with FastAPI

📖 Scenario: You are building a chat application where users connect through WebSocket. To keep the chat secure, you want to check if users are authenticated before allowing them to join the chat.

🎯 Goal: Build a FastAPI WebSocket endpoint that authenticates users by checking a token before accepting the connection.

📋 What You'll Learn

Create a dictionary called valid_tokens with exact token-user pairs

Create a variable called auth_token to hold the token from the WebSocket query

Use a for loop with variables token and user to check if auth_token is valid

Accept the WebSocket connection only if the token is valid, otherwise close it

💡 Why This Matters

🌍 Real World

WebSocket authentication is essential for real-time apps like chat, live notifications, or games to ensure only authorized users connect.

💼 Career

Understanding WebSocket authentication is important for backend developers working with real-time communication and security.

1

Create valid tokens dictionary

Create a dictionary called valid_tokens with these exact entries: 'token123': 'alice', 'token456': 'bob', 'token789': 'carol'.

FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

# Create valid_tokens dictionary here

Need a hint?

Use curly braces {} to create the dictionary with the exact keys and values.

2

Extract token from WebSocket query

Inside the websocket_endpoint function, create a variable called auth_token that gets the token from websocket.query_params.get('token').

FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

valid_tokens = {'token123': 'alice', 'token456': 'bob', 'token789': 'carol'}

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    # Extract auth_token from query params
    # Your code here

Need a hint?

Use websocket.query_params.get('token') to get the token from the URL query.

3

Check if token is valid

Use a for loop with variables token and user to iterate over valid_tokens.items(). Inside the loop, check if auth_token == token. If yes, accept the WebSocket connection with await websocket.accept() and break the loop.

FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

valid_tokens = {'token123': 'alice', 'token456': 'bob', 'token789': 'carol'}

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    auth_token = websocket.query_params.get('token')
    # Check token validity and accept connection
    # Your code here

Need a hint?

Use for token, user in valid_tokens.items(): to loop through tokens and users.

4

Close connection if token invalid

After the for loop, add an else clause that closes the WebSocket connection with await websocket.close() if no valid token was found.

FastAPI

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

valid_tokens = {'token123': 'alice', 'token456': 'bob', 'token789': 'carol'}

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
    auth_token = websocket.query_params.get('token')
    for token, user in valid_tokens.items():
        if auth_token == token:
            await websocket.accept()
            break
    # Close connection if token invalid
    # Your code here

Need a hint?

Use the else clause on the for loop to close the connection if no token matches.