Intermediate

If you have ever built a chat application, a live dashboard, or a multiplayer game, you know that regular HTTP requests fall short when you need real-time, bidirectional communication. Polling the server every few seconds wastes bandwidth and adds latency. WebSockets solve this by keeping a persistent connection open between client and server, allowing both sides to send messages at any time.

Python’s websockets library makes building WebSocket servers and clients remarkably straightforward. It is built on top of asyncio, so it integrates naturally with Python’s async ecosystem. You only need to install one package — pip install websockets — and you are ready to go.

In this tutorial, you will learn how to create a WebSocket server, build a client that connects to it, implement broadcast messaging for chat-style applications, handle connection lifecycle events, and add basic authentication. By the end, you will have a working real-time chat server that you can extend for your own projects.

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

WebSocket Echo Server: Quick Example

Part of the Python Web Frameworks Hub. See the full hub for related Python tutorials.

Here is the simplest possible WebSocket server — it echoes back whatever the client sends:

# echo_server.py
import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        print(f"Received: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        print("Echo server running on ws://localhost:8765")
        await asyncio.Future()  # Run forever

asyncio.run(main())

Output (server):

Echo server running on ws://localhost:8765
Received: Hello WebSocket!
Received: How are you?

The websockets.serve() function starts a server that calls the echo handler for every new connection. The async for message in websocket pattern automatically handles the connection lifecycle — it reads messages until the client disconnects, then exits cleanly. We will build on this pattern throughout the tutorial.

What Are WebSockets and Why Use Them?

WebSockets are a communication protocol that provides full-duplex (two-way) communication channels over a single TCP connection. Unlike HTTP, where the client must initiate every request, WebSockets allow either side to send data at any time after the initial handshake.

The WebSocket protocol starts with an HTTP upgrade request. The client sends a regular HTTP request with an Upgrade: websocket header, and if the server agrees, the connection is upgraded from HTTP to WebSocket. From that point on, both sides communicate using lightweight WebSocket frames instead of full HTTP requests.

Use WebSockets when you need low-latency, real-time updates. Use HTTP when you need stateless request-response patterns (like REST APIs).

Setting Up the websockets Library

Install the library with pip:

# install.sh
pip install websockets

Output:

Successfully installed websockets-13.1

The websockets library requires Python 3.8 or later. It has no dependencies beyond the standard library’s asyncio module, keeping your dependency tree clean.

Building a WebSocket Client

To test our servers, we need a client. Here is a simple interactive client that sends messages and prints responses:

# client.py
import asyncio
import websockets

async def chat_client():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        print("Connected to server. Type messages (Ctrl+C to quit):")
        
        # Send a greeting
        await websocket.send("Hello from Python client!")
        response = await websocket.recv()
        print(f"Server says: {response}")
        
        # Interactive loop
        while True:
            message = input("> ")
            await websocket.send(message)
            response = await websocket.recv()
            print(f"Server says: {response}")

asyncio.run(chat_client())

Output:

Connected to server. Type messages (Ctrl+C to quit):
Server says: Echo: Hello from Python client!
> Testing 123
Server says: Echo: Testing 123

The websockets.connect() context manager handles the connection lifecycle automatically. When you exit the async with block (or the program ends), the connection closes cleanly with a proper WebSocket close handshake.

Building a Broadcast Chat Server

A real chat server needs to broadcast messages from one client to all connected clients. This requires tracking active connections in a set:

# chat_server.py
import asyncio
import websockets
import json
from datetime import datetime

connected_clients = set()

async def broadcast(message, sender=None):
    """Send a message to all connected clients except the sender."""
    disconnected = set()
    for client in connected_clients:
        if client != sender:
            try:
                await client.send(message)
            except websockets.ConnectionClosed:
                disconnected.add(client)
    # Clean up disconnected clients
    connected_clients -= disconnected

async def chat_handler(websocket):
    """Handle a single client connection."""
    # Register client
    connected_clients.add(websocket)
    client_id = f"User-{id(websocket) % 1000}"
    print(f"{client_id} connected. Total clients: {len(connected_clients)}")
    
    # Notify others
    join_msg = json.dumps({
        "type": "system",
        "message": f"{client_id} joined the chat",
        "timestamp": datetime.now().isoformat()
    })
    await broadcast(join_msg, sender=websocket)
    
    try:
        async for message in websocket:
            # Wrap message with metadata
            chat_msg = json.dumps({
                "type": "chat",
                "sender": client_id,
                "message": message,
                "timestamp": datetime.now().isoformat()
            })
            print(f"{client_id}: {message}")
            await broadcast(chat_msg)
    except websockets.ConnectionClosed:
        pass
    finally:
        # Unregister client
        connected_clients.discard(websocket)
        leave_msg = json.dumps({
            "type": "system",
            "message": f"{client_id} left the chat",
            "timestamp": datetime.now().isoformat()
        })
        await broadcast(leave_msg)
        print(f"{client_id} disconnected. Total clients: {len(connected_clients)}")

async def main():
    async with websockets.serve(chat_handler, "localhost", 8765):
        print("Chat server running on ws://localhost:8765")
        await asyncio.Future()

asyncio.run(main())

Output (server with two clients):

Chat server running on ws://localhost:8765
User-142 connected. Total clients: 1
User-857 connected. Total clients: 2
User-142: Hello everyone!
User-857: Hey there!
User-142 disconnected. Total clients: 1

The connected_clients set tracks all active connections. When a client sends a message, broadcast() forwards it to every other connected client. The try/finally block ensures we clean up the client set even if the connection drops unexpectedly.

Handling Connection Errors Gracefully

Real-world WebSocket connections drop unexpectedly due to network issues, client crashes, or timeouts. Robust error handling is essential:

# robust_server.py
import asyncio
import websockets
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("websocket-server")

async def robust_handler(websocket):
    """Handler with comprehensive error management."""
    remote = websocket.remote_address
    logger.info(f"New connection from {remote}")
    
    try:
        # Set a ping interval to detect dead connections
        async for message in websocket:
            if len(message) > 10000:
                await websocket.send("Error: Message too long (max 10000 chars)")
                continue
            
            await websocket.send(f"Processed: {message}")
            
    except websockets.ConnectionClosedError as e:
        logger.warning(f"Connection closed with error: {e.code} {e.reason}")
    except websockets.ConnectionClosedOK:
        logger.info(f"Connection closed normally from {remote}")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
    finally:
        logger.info(f"Cleanup complete for {remote}")

async def main():
    async with websockets.serve(
        robust_handler,
        "localhost",
        8765,
        ping_interval=20,     # Send ping every 20 seconds
        ping_timeout=10,      # Wait 10 seconds for pong
        max_size=2**20,       # Max message size: 1MB
        close_timeout=5       # Wait 5 seconds for close handshake
    ):
        logger.info("Robust server running on ws://localhost:8765")
        await asyncio.Future()

asyncio.run(main())

Output:

INFO:websocket-server:Robust server running on ws://localhost:8765
INFO:websocket-server:New connection from ('127.0.0.1', 54321)
WARNING:websocket-server:Connection closed with error: 1006 
INFO:websocket-server:Cleanup complete for ('127.0.0.1', 54321)

The ping_interval and ping_timeout parameters enable automatic keepalive detection. If a client stops responding to pings within the timeout, the server closes the connection and triggers cleanup. The max_size parameter prevents clients from sending oversized messages that could exhaust server memory.

Adding Basic Authentication

You can authenticate WebSocket connections using query parameters, headers, or an initial authentication message. Here is a token-based approach:

# auth_server.py
import asyncio
import websockets
import json
import secrets

# Valid tokens (in production, use a database)
VALID_TOKENS = {
    "demo-token-abc123": "alice",
    "demo-token-def456": "bob"
}

async def authenticate(websocket):
    """Authenticate the client using the first message."""
    try:
        auth_msg = await asyncio.wait_for(websocket.recv(), timeout=5.0)
        data = json.loads(auth_msg)
        token = data.get("token", "")
        
        for valid_token, username in VALID_TOKENS.items():
            if secrets.compare_digest(token, valid_token):
                await websocket.send(json.dumps({
                    "type": "auth", 
                    "status": "ok",
                    "username": username
                }))
                return username
        
        await websocket.send(json.dumps({
            "type": "auth",
            "status": "error", 
            "message": "Invalid token"
        }))
        return None
        
    except asyncio.TimeoutError:
        await websocket.send(json.dumps({
            "type": "auth",
            "status": "error",
            "message": "Authentication timeout"
        }))
        return None

async def secure_handler(websocket):
    """Handler that requires authentication."""
    username = await authenticate(websocket)
    if not username:
        await websocket.close(1008, "Authentication failed")
        return
    
    print(f"{username} authenticated successfully")
    
    async for message in websocket:
        response = f"[{username}] {message}"
        await websocket.send(response)

async def main():
    async with websockets.serve(secure_handler, "localhost", 8765):
        print("Secure server running on ws://localhost:8765")
        await asyncio.Future()

asyncio.run(main())

Output:

Secure server running on ws://localhost:8765
alice authenticated successfully

The server expects the first message to contain a JSON object with a token field. It uses secrets.compare_digest() for timing-safe comparison and gives the client 5 seconds to authenticate before timing out. If authentication fails, the connection is closed with WebSocket status code 1008 (Policy Violation).

Real-Life Example: Live Notification System

Let us build a notification system where a server pushes real-time alerts to connected clients based on their subscribed topics:

# notification_server.py
import asyncio
import websockets
import json
from datetime import datetime
from collections import defaultdict

class NotificationServer:
    """Real-time notification server with topic subscriptions."""
    
    def __init__(self):
        self.subscribers = defaultdict(set)  # topic -> set of websockets
        self.clients = {}  # websocket -> client info
    
    async def handle_subscribe(self, websocket, topics):
        """Subscribe a client to one or more topics."""
        for topic in topics:
            self.subscribers[topic].add(websocket)
        self.clients[websocket]["topics"] = topics
        await websocket.send(json.dumps({
            "type": "subscribed",
            "topics": topics
        }))
    
    async def notify(self, topic, message):
        """Send a notification to all subscribers of a topic."""
        if topic not in self.subscribers:
            return 0
        
        notification = json.dumps({
            "type": "notification",
            "topic": topic,
            "message": message,
            "timestamp": datetime.now().isoformat()
        })
        
        sent = 0
        disconnected = set()
        for client in self.subscribers[topic]:
            try:
                await client.send(notification)
                sent += 1
            except websockets.ConnectionClosed:
                disconnected.add(client)
        
        # Cleanup
        self.subscribers[topic] -= disconnected
        return sent
    
    async def handler(self, websocket):
        """Main connection handler."""
        self.clients[websocket] = {"topics": [], "connected": datetime.now()}
        
        try:
            async for raw_message in websocket:
                data = json.loads(raw_message)
                action = data.get("action")
                
                if action == "subscribe":
                    await self.handle_subscribe(
                        websocket, data.get("topics", [])
                    )
                elif action == "publish":
                    topic = data.get("topic")
                    message = data.get("message")
                    count = await self.notify(topic, message)
                    await websocket.send(json.dumps({
                        "type": "published",
                        "topic": topic,
                        "recipients": count
                    }))
        except websockets.ConnectionClosed:
            pass
        finally:
            # Remove from all subscriptions
            for topic_subs in self.subscribers.values():
                topic_subs.discard(websocket)
            self.clients.pop(websocket, None)

server = NotificationServer()

async def main():
    async with websockets.serve(server.handler, "localhost", 8765):
        print("Notification server running on ws://localhost:8765")
        
        # Simulate periodic system notifications
        async def system_alerts():
            while True:
                await asyncio.sleep(30)
                await server.notify("system", "Heartbeat check - all systems normal")
        
        await asyncio.gather(
            asyncio.Future(),  # Run forever
            system_alerts()
        )

asyncio.run(main())

Output:

Notification server running on ws://localhost:8765

This server supports topic-based subscriptions. Clients subscribe to topics like “alerts”, “updates”, or “system”, and only receive notifications for their subscribed topics. The notify() method handles broadcasting to subscribers and automatically cleans up disconnected clients. You could extend this with persistent message queues, delivery confirmation, or priority levels.

Frequently Asked Questions

How many concurrent connections can a Python WebSocket server handle?

A single Python process using asyncio can typically handle 10,000-50,000 concurrent WebSocket connections, depending on message frequency and server resources. The websockets library is efficient with memory, using roughly 10KB per connection. For higher scale, use multiple processes behind a load balancer like Nginx.

How do I add TLS/SSL encryption (wss://)?

Pass an ssl context to websockets.serve(): ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER), then load your certificate with ssl_context.load_cert_chain(certfile, keyfile). In production, use a reverse proxy like Nginx to handle TLS termination instead.

How do I handle client reconnection?

The websockets library does not auto-reconnect. On the client side, wrap your connection in a retry loop with exponential backoff. The websockets.connect() context manager closes cleanly, so you can simply reconnect in a while True loop with a try/except block catching ConnectionClosed errors.

Can I send binary data over WebSockets?

Yes. The websockets library automatically detects whether you send str (text frames) or bytes (binary frames). Use await websocket.send(b"\x00\x01\x02") for binary data. This is useful for sending images, audio streams, or protobuf-encoded messages.

Can I use WebSockets with Django or Flask?

Django supports WebSockets through Django Channels, which adds an ASGI layer. Flask does not natively support WebSockets, but you can use Flask-SocketIO or run a separate websockets server alongside Flask. For new projects needing WebSocket support, consider FastAPI, which has native WebSocket support built on Starlette.

Conclusion

You now have a solid foundation for building WebSocket applications in Python. We covered creating echo servers, building broadcast chat systems, handling connection errors gracefully, adding authentication, and implementing topic-based notification systems. The websockets library’s async-first design makes it easy to handle thousands of concurrent connections with clean, readable code.

Try extending the notification server with features like message persistence, delivery acknowledgments, or a web-based admin dashboard. For the complete API reference, see the official websockets documentation.

Bare websockets Library Server

The websockets library is the canonical pure-Python implementation — modern async, type-hinted, RFC-compliant. The minimal echo server is 8 lines:

# pip install websockets

import asyncio
from websockets.asyncio.server import serve

async def echo(ws):
    async for message in ws:
        await ws.send(f"Echo: {message}")

async def main():
    async with serve(echo, "0.0.0.0", 8765) as server:
        await server.serve_forever()

asyncio.run(main())

Connect from anywhere — browser JavaScript (new WebSocket("ws://localhost:8765")), wscat, or another Python client via websockets.connect. The handler runs once per connection; async for message reads incoming frames until the client disconnects.

Broadcasting to Multiple Clients

For chat rooms or live dashboards, track all connected clients in a set, broadcast each message to all of them:

import asyncio
from websockets.asyncio.server import serve
import json

clients = set()

async def handler(ws):
    clients.add(ws)
    try:
        async for raw in ws:
            payload = json.loads(raw)
            outbound = json.dumps({"user": payload["user"], "text": payload["text"]})
            # Send to everyone, ignore disconnected clients
            await asyncio.gather(
                *[c.send(outbound) for c in clients],
                return_exceptions=True,
            )
    finally:
        clients.remove(ws)

async def main():
    async with serve(handler, "0.0.0.0", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

Auth, Subprotocols, and Path Routing

Production WebSocket servers need auth and routing. Use the process_request hook for HTTP-level checks before upgrading to WebSocket:

from websockets.asyncio.server import serve
from http import HTTPStatus

async def auth_handler(connection, request):
    # Reject before upgrade
    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
    if not is_valid(token):
        return connection.respond(HTTPStatus.UNAUTHORIZED, "Bad token\n")
    # Attach user info for the handler
    connection.user_id = get_user_id_from_token(token)

async def handler(ws):
    user_id = ws.user_id
    async for msg in ws:
        # use user_id in messages
        pass

async def main():
    async with serve(handler, "0.0.0.0", 8765, process_request=auth_handler):
        await asyncio.Future()

WebSockets in FastAPI

If you already have a FastAPI app, FastAPI ships its own WebSocket support — no second framework needed:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
active = []

@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    active.append(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for client in active:
                await client.send_text(f"[broadcast] {data}")
    except WebSocketDisconnect:
        active.remove(websocket)

Heartbeats and Reconnection

WebSockets across the public internet drop silently — proxies and NAT timeouts kill idle connections after 30-120 seconds. Send pings to keep them alive:

async with serve(handler, "0.0.0.0", 8765, ping_interval=20, ping_timeout=10):
    ...

For clients, implement exponential-backoff reconnect on the JS side — bare browsers don’t auto-reconnect.

Common Pitfalls

• Blocking inside the handler. A synchronous time.sleep(5) blocks the entire event loop. Use await asyncio.sleep(5). Long CPU work belongs in asyncio.to_thread.
• Forgetting to remove disconnected clients. Without cleanup, broadcasting tries to send on a dead socket — exception, server crashes. Use try/finally with remove.
• Treating each message as JSON without try/except. A malformed message kills the handler. Wrap json.loads in a try/except and send an error message back.
• Holding references in the broadcast list. Long-lived sets of WebSocket objects can leak memory if a handler exits without removing itself. Always remove in finally.
• Skipping wss:// in production. Plain ws:// sends data unencrypted. Always terminate TLS at your reverse proxy and use wss://.

FAQ

Q: websockets library or FastAPI WebSockets?
A: FastAPI if you already have a FastAPI app. The websockets library when you want a dedicated, low-overhead WebSocket-only server.

Q: How many concurrent connections can one server handle?
A: Thousands on a single Python process, tens of thousands with uvloop. Beyond that, scale horizontally with a load balancer that supports sticky sessions or use a pub/sub layer (Redis) for cross-instance broadcasts.

Q: How do I scale broadcasts across multiple server instances?
A: Use Redis pub/sub as the message bus. Each instance subscribes to a channel; broadcasts go through Redis; every connected client gets the message regardless of which instance they’re on.

Q: Polling vs WebSockets vs SSE?
A: WebSockets for bidirectional real-time (chat, multiplayer games). Server-Sent Events for one-way push (dashboards, notifications). Polling when you need a backup. Each has its place.

Q: Behind nginx — what’s the config?
A: Add proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; to your location block. Increase proxy_read_timeout well beyond default (3600s).

Wrapping Up

WebSockets are the right answer for any real-time, bidirectional, low-latency need: chat, multiplayer, live cursors, streaming logs, collaborative editing. The websockets library is the modern Python implementation; FastAPI bundles equivalent functionality if you’re already using it. Pings, auth at upgrade time, and TLS termination at a reverse proxy are the production essentials.

Related Articles

• How To Use Python MCP Server for AI Tool Integration
• How To Build a Python gRPC Service with grpcio
• How To Build REST APIs with Python and Starlette
• How To Build a Telegram Bot with Python
• How To Build a Flask Web Application in Python
• How To Build GraphQL APIs with Python and Strawberry
• How To Build a Simple Chatbot with Python and Ollama

Intermediate

You have probably written code that generates random tokens for password resets, API keys, or session identifiers. If you used the random module for this, your tokens are predictable — an attacker who knows the seed can reproduce every token you generate. This is not a theoretical risk; it has been the root cause of real security breaches.

Python 3.6 introduced the secrets module specifically for generating cryptographically secure random values. It is part of the standard library, so there is nothing to install. It uses the operating system’s best source of randomness (/dev/urandom on Linux, CryptGenRandom on Windows) and produces tokens that are safe for security-sensitive applications.

In this tutorial, you will learn how to generate secure tokens in multiple formats (hex, URL-safe, bytes), create safe passwords, build one-time password reset links, and compare tokens securely. By the end, you will have a complete toolkit for handling secrets in any Python application.

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

Generating a Secure Token: Quick Example

Here is the fastest way to generate a cryptographically secure token in Python:

# quick_token.py
import secrets

# Generate a 32-byte URL-safe token
token = secrets.token_urlsafe(32)
print(f"Token: {token}")
print(f"Length: {len(token)} characters")

Output:

Token: x7Kj2mN9pQrS1tUvWxYz3aB4cD5eF6gH7iJ8kL0mNo
Length: 43 characters

The secrets.token_urlsafe() function generates random bytes and encodes them as a URL-safe base64 string. The 32-byte input produces a 43-character token with enough entropy (256 bits) to resist brute-force attacks. We will explore all the token types and their use cases in the sections below.

What is the Secrets Module and Why Use It?

The secrets module provides functions for generating cryptographically strong random numbers suitable for managing secrets such as account authentication, tokens, and similar. Think of it as the security-focused counterpart to random.

The key difference is the source of randomness. The random module uses a Mersenne Twister algorithm — fast and statistically uniform, but deterministic. If someone discovers the internal state (which requires observing only 624 consecutive outputs), they can predict all future values. The secrets module draws from the OS entropy pool, which is non-deterministic and cannot be reversed.

The rule is simple: if the value protects something, use secrets. If it does not, random is fine.

Generating Tokens in Different Formats

The secrets module offers three token functions, each producing a different encoding of the same underlying random bytes. The format you choose depends on where you plan to use the token.

Hex Tokens with token_hex()

Hex tokens produce a string of hexadecimal characters (0-9, a-f). They are commonly used for database identifiers, session IDs, and anywhere you need a clean alphanumeric string.

# hex_tokens.py
import secrets

# Generate hex tokens of different lengths
token_16 = secrets.token_hex(16)  # 16 bytes = 32 hex chars
token_32 = secrets.token_hex(32)  # 32 bytes = 64 hex chars

print(f"16-byte hex: {token_16}")
print(f"32-byte hex: {token_32}")
print(f"Lengths: {len(token_16)}, {len(token_32)} characters")

Output:

16-byte hex: a3b7c9d1e5f20a4b6c8d0e2f4a6b8c0d
32-byte hex: f1a2b3c4d5e6f7089a0b1c2d3e4f50617a8b9c0d1e2f3a4b5c6d7e8f90a1b2c3
Lengths: 32, 64 characters

Each byte produces two hex characters, so token_hex(16) returns a 32-character string. For most applications, 32 bytes (256 bits) provides sufficient entropy.

URL-Safe Tokens with token_urlsafe()

URL-safe tokens use base64 encoding with - and _ instead of + and /. This makes them safe to include in URLs, query parameters, and HTTP headers without encoding issues.

# urlsafe_tokens.py
import secrets

token = secrets.token_urlsafe(32)
reset_link = f"https://example-app.com/reset?token={token}"

print(f"Token: {token}")
print(f"Reset link: {reset_link}")

Output:

Token: Rk9PQkFSLVRoaXMtaXMtYS1zYWZlLXRva2VuLXlheg
Reset link: https://example-app.com/reset?token=Rk9PQkFSLVRoaXMtaXMtYS1zYWZlLXRva2VuLXlheg

This is the most commonly used format for password reset links, email verification tokens, and API keys because it is compact and URL-compatible.

Raw Bytes with token_bytes()

When you need raw binary data — for encryption keys, HMACs, or feeding into other cryptographic functions — use token_bytes().

# raw_bytes.py
import secrets

raw = secrets.token_bytes(32)
print(f"Bytes: {raw}")
print(f"Length: {len(raw)} bytes")
print(f"As hex: {raw.hex()}")

Output:

Bytes: b'\xd4\x8a\x1f...(32 bytes of binary data)'
Length: 32 bytes
As hex: d48a1f3b7c9e2d...

Raw bytes are not human-readable, but they are the right choice when the token feeds directly into a cryptographic function like hmac.new() or a symmetric encryption library.

Generating Secure Passwords

The secrets module includes secrets.choice(), which selects a random element from a sequence using cryptographically secure randomness. Combined with Python’s string module, you can build password generators that meet any complexity requirement.

# password_gen.py
import secrets
import string

def generate_password(length=16):
    """Generate a secure password with mixed character types."""
    alphabet = string.ascii_letters + string.digits + string.punctuation
    
    # Ensure at least one of each required type
    password = [
        secrets.choice(string.ascii_uppercase),
        secrets.choice(string.ascii_lowercase),
        secrets.choice(string.digits),
        secrets.choice(string.punctuation),
    ]
    
    # Fill remaining length with random choices
    password += [secrets.choice(alphabet) for _ in range(length - 4)]
    
    # Shuffle to avoid predictable positions
    # Use secrets for the shuffle via sorting with random keys
    password.sort(key=lambda _: secrets.randbelow(1000))
    
    return ''.join(password)

# Generate 5 passwords
for i in range(5):
    pwd = generate_password(20)
    print(f"Password {i+1}: {pwd}")

Output:

Password 1: kQ7#mR2$nP9!xW4&bT6@
Password 2: jF3*hL8^vC1!yN5&dK7#
Password 3: wS6@tH4$pB9#mX2!qR8&
Password 4: gD1#zE5$cA3!oI7&uY9@
Password 5: fV8*aJ2^lG6!eM4&nW0#

The key detail is the shuffle step. Without it, the first four characters always follow the same type pattern (uppercase, lowercase, digit, punctuation). The secrets.randbelow() function provides a secure random integer that we use as a sort key to randomize character positions.

Comparing Tokens Securely

When verifying a token (checking if a submitted reset token matches the stored one), you must use constant-time comparison to prevent timing attacks. A timing attack measures how long the comparison takes — a regular == comparison returns False as soon as it finds the first mismatched character, leaking information about how many characters were correct.

# secure_compare.py
import secrets
import hmac

stored_token = "abc123def456"
submitted_token = "abc123def456"

# WRONG: vulnerable to timing attacks
if stored_token == submitted_token:
    print("Match (but insecure comparison)")

# RIGHT: constant-time comparison
if secrets.compare_digest(stored_token, submitted_token):
    print("Match (secure comparison)")

# Also works with bytes
stored_bytes = b"secret_token_value"
submitted_bytes = b"secret_token_value"
if secrets.compare_digest(stored_bytes, submitted_bytes):
    print("Bytes match (secure comparison)")

Output:

Match (but insecure comparison)
Match (secure comparison)
Bytes match (secure comparison)

The secrets.compare_digest() function always examines every character, taking the same amount of time regardless of where (or whether) the strings differ. This is the same function used internally by hmac.compare_digest().

Practical Security Patterns

Building a Password Reset Flow

Here is how to generate a time-limited password reset token with proper security practices:

# reset_flow.py
import secrets
import hashlib
import time

class TokenManager:
    """Manages secure, time-limited tokens."""
    
    def __init__(self, expiry_seconds=3600):
        self.tokens = {}  # In production, use a database
        self.expiry = expiry_seconds
    
    def create_token(self, user_id):
        """Generate a reset token for a user."""
        raw_token = secrets.token_urlsafe(32)
        # Store only the hash -- never store raw tokens
        token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
        self.tokens[token_hash] = {
            'user_id': user_id,
            'created': time.time()
        }
        return raw_token  # Send this to the user via email
    
    def verify_token(self, submitted_token):
        """Verify a submitted token and return user_id if valid."""
        token_hash = hashlib.sha256(submitted_token.encode()).hexdigest()
        
        # Constant-time lookup using compare_digest
        for stored_hash, data in self.tokens.items():
            if secrets.compare_digest(stored_hash, token_hash):
                # Check expiry
                if time.time() - data['created'] > self.expiry:
                    del self.tokens[stored_hash]
                    return None  # Expired
                return data['user_id']
        return None  # Not found

# Usage
manager = TokenManager(expiry_seconds=3600)
token = manager.create_token("user_42")
print(f"Reset token: {token}")
print(f"Verified: {manager.verify_token(token)}")
print(f"Invalid token: {manager.verify_token('wrong-token')}")

Output:

Reset token: aB3cD4eF5gH6iJ7kL8mN9oP0qR1sT2uV3wX4yZ5
Verified: user_42
Invalid token: None

The critical detail is that we store only the hash of the token, never the raw token itself. If the database is breached, attackers get useless hashes. The raw token only exists in the email sent to the user and in memory during verification.

Generating API Keys

API keys typically need a prefix for identification and a random body for security:

# api_keys.py
import secrets

def generate_api_key(prefix="pk"):
    """Generate a prefixed API key."""
    random_part = secrets.token_urlsafe(24)
    return f"{prefix}_{random_part}"

# Generate different key types
public_key = generate_api_key("pk")
secret_key = generate_api_key("sk")
test_key = generate_api_key("tk")

print(f"Public key:  {public_key}")
print(f"Secret key:  {secret_key}")
print(f"Test key:    {test_key}")

Output:

Public key:  pk_Rk9PQkFSLVRoaXMtaXMtYQ
Secret key:  sk_c2VjcmV0LWtleS12YWx1ZQ
Test key:    tk_dGVzdC1rZXktdmFsdWUtaGVyZQ

The prefix makes it easy to identify key types in logs and configuration files without revealing the secret portion. This is the same pattern used by services like Stripe (sk_live_, pk_test_).

Real-Life Example: Secure Invitation System

Let us build a complete invitation system that generates secure invite codes, tracks their usage, and enforces expiration and single-use constraints.

# invitation_system.py
import secrets
import hashlib
import time
from datetime import datetime

class InvitationSystem:
    """Secure invitation code manager with expiry and usage tracking."""
    
    def __init__(self):
        self.invitations = {}
    
    def create_invite(self, creator, role="member", max_uses=1, 
                      expiry_hours=48):
        """Create a new invitation with constraints."""
        code = secrets.token_urlsafe(16)
        code_hash = hashlib.sha256(code.encode()).hexdigest()
        
        self.invitations[code_hash] = {
            'creator': creator,
            'role': role,
            'max_uses': max_uses,
            'current_uses': 0,
            'created': time.time(),
            'expiry': time.time() + (expiry_hours * 3600),
            'used_by': []
        }
        return code
    
    def redeem_invite(self, code, user_name):
        """Attempt to redeem an invitation code."""
        code_hash = hashlib.sha256(code.encode()).hexdigest()
        
        for stored_hash, invite in self.invitations.items():
            if not secrets.compare_digest(stored_hash, code_hash):
                continue
            
            if time.time() > invite['expiry']:
                return {"success": False, "error": "Invitation expired"}
            
            if invite['current_uses'] >= invite['max_uses']:
                return {"success": False, "error": "Invitation fully used"}
            
            invite['current_uses'] += 1
            invite['used_by'].append(user_name)
            return {
                "success": True,
                "role": invite['role'],
                "message": f"Welcome {user_name}! Role: {invite['role']}"
            }
        
        return {"success": False, "error": "Invalid invitation code"}
    
    def get_stats(self):
        """Get invitation usage statistics."""
        active = sum(1 for inv in self.invitations.values() 
                     if time.time() < inv['expiry'] 
                     and inv['current_uses'] < inv['max_uses'])
        expired = sum(1 for inv in self.invitations.values() 
                      if time.time() >= inv['expiry'])
        return {"active": active, "expired": expired, 
                "total": len(self.invitations)}

# Demo
system = InvitationSystem()

# Create invitations
admin_code = system.create_invite("alice", role="admin", max_uses=1)
team_code = system.create_invite("bob", role="editor", max_uses=5, 
                                  expiry_hours=72)

print(f"Admin invite: {admin_code}")
print(f"Team invite:  {team_code}")
print()

# Redeem invitations
print(system.redeem_invite(admin_code, "charlie"))
print(system.redeem_invite(admin_code, "dave"))  # Should fail
print(system.redeem_invite(team_code, "eve"))
print(system.redeem_invite("invalid-code", "mallory"))
print()

print(f"Stats: {system.get_stats()}")

Output:

Admin invite: xK7mN2pQ9rS1tUvW
Team invite:  aB3cD4eF5gH6iJ7k

{'success': True, 'role': 'admin', 'message': 'Welcome charlie! Role: admin'}
{'success': False, 'error': 'Invitation fully used'}
{'success': True, 'role': 'editor', 'message': 'Welcome eve! Role: editor'}
{'success': False, 'error': 'Invalid invitation code'}

Stats: {'active': 1, 'expired': 0, 'total': 2}

This system demonstrates several security best practices: hashed storage of codes, constant-time comparison, usage limits, and time-based expiration. In production, you would replace the in-memory dictionary with a database and add rate limiting to prevent brute-force code guessing.

Frequently Asked Questions

Can I just use random.SystemRandom() instead of secrets?

Yes, random.SystemRandom() also uses the OS entropy pool and is cryptographically secure. However, secrets is the recommended module since Python 3.6 because it provides a cleaner API specifically designed for security tasks. It also includes compare_digest() and token_urlsafe(), which SystemRandom does not.

How many bytes should my tokens be?

The Python documentation recommends at least 32 bytes (256 bits) for tokens used in security contexts. This provides sufficient entropy that brute-force guessing is infeasible even with massive computing resources. For lower-stakes use cases like email verification, 16 bytes (128 bits) is still very strong.

What happens if I call token_hex() with no arguments?

If you omit the byte count, secrets uses a default that is “reasonable for most use cases” — currently 32 bytes. However, it is better to specify the length explicitly so your code is self-documenting and immune to future default changes.

Is secrets slower than random?

Yes, but the difference is negligible for token generation. Generating 10,000 tokens with secrets takes about 50 milliseconds compared to 15 milliseconds with random. Since you typically generate tokens one at a time (not in bulk), the performance difference is irrelevant.

Should I use uuid4() or secrets for unique identifiers?

uuid.uuid4() generates random UUIDs using os.urandom(), so it is cryptographically secure. Use UUIDs when you need a standardized format (like database primary keys). Use secrets when you need a security token with a specific format (URL-safe, hex) or when you need compare_digest() for safe verification.

Conclusion

The secrets module gives you everything you need to generate cryptographically secure tokens, passwords, and random values in Python. We covered token_hex() for hexadecimal strings, token_urlsafe() for URL-compatible tokens, token_bytes() for raw binary data, and compare_digest() for timing-attack-resistant comparisons. We also built practical systems for password resets, API key generation, and invitation codes.

Try extending the invitation system with features like rate limiting, audit logging, or integration with a real database using SQLite or PostgreSQL. The secrets module is small but foundational — once you understand it, you can build secure authentication flows for any Python application.

For the full API reference, see the official Python secrets documentation.

Related Articles

• How To Use PyJWT for JSON Web Tokens in Python
• How To Use Python MCP Server for AI Tool Integration
• How To Use Python Litestar for Async Web APIs
• How To Use Python litellm for Multi-Model LLM Integration
• How To Use Python instructor for Structured LLM Outputs
• How To Use Python LlamaIndex for Document Q&A
• How To Read and Update Smartsheets With Python 3

Skill Level: Intermediate

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

Introduction to Modern HTTP Requests

For years, the Python requests library has been the go-to solution for making HTTP requests. While requests is powerful and user-friendly, it has limitations that modern Python developers encounter daily. It doesn’t support async operations natively, lacks HTTP/2 support, and can feel sluggish when you’re handling dozens of concurrent requests. If you’ve found yourself wrestling with requests’ synchronous nature or spinning up ThreadPoolExecutor just to manage multiple requests, you’re not alone. Many developers hit a wall where they need something more capable.

Enter httpx — a modern HTTP client that drops in as a replacement for requests while adding powerful features like native async/await support, HTTP/2 capabilities, and streaming responses. The best part? If you already know requests, you’ll feel right at home with httpx. The API is remarkably similar, which means you can start using it immediately without relearning everything. We’ll cover installation, basic usage, async patterns, and real-world examples that show why httpx is becoming the preferred choice for new Python projects.

In this guide, we’ll walk through everything you need to know about httpx. We’ll start with a quick example to get you up and running, explore what makes httpx special compared to other HTTP libraries, and then dive into practical patterns you can use in your own projects. Whether you’re building a simple API client or managing complex async workflows, httpx has the tools you need. Let’s get started.

Quick Example: GET Request in 5 Lines

# quick_get.py
import httpx

response = httpx.get("https://jsonplaceholder.typicode.com/posts/1")
print(response.status_code)
print(response.json())

That’s it. If you’ve used requests before, you already know httpx. The synchronous API is nearly identical, but under the hood, httpx brings modern features to the table. Now let’s explore what makes httpx more powerful than its predecessor.

When the 429 wave hits, read the headers

What is httpx and Why Use It?

httpx is a modern HTTP client library for Python that combines the simplicity of requests with advanced features like async support, HTTP/2, and more sophisticated timeout handling. Created by Tom Christie (the same developer behind Starlette and FastAPI), httpx is built on a foundation that understands modern Python development patterns. It’s not just a requests replacement — it’s a redesign based on everything we’ve learned about making HTTP libraries in the 2020s.

The key differences matter when you’re building real applications. Unlike requests, httpx supports both synchronous and asynchronous code from the same library. You don’t need to install separate packages or maintain multiple code paths. It supports HTTP/2 by default, which means better performance for services that support it. It has built-in connection pooling, proper async context managers, and a cleaner API that feels more Pythonic.

Let’s compare httpx to other HTTP libraries in the Python ecosystem:

httpx shines when you need synchronous and asynchronous code in the same project. Unlike aiohttp, which requires AsyncIO from the start, httpx lets you start simple and scale to async when you need it. Unlike requests, httpx doesn’t force you into threading patterns when you want to handle multiple requests concurrently. It’s the bridge between requests’ simplicity and aiohttp’s power.

Installing httpx

Installation is straightforward. httpx is available on PyPI and installs cleanly without forcing dependencies on you. For basic functionality, you only need one command.

# install_httpx.sh
pip install httpx

If you want HTTP/2 support with better performance, you can install the optional dependencies. The h2 library handles HTTP/2 protocol details, while httpcore provides the underlying transport layer.

# install_httpx_with_http2.sh
pip install httpx[http2]

That’s all you need. Unlike some HTTP libraries, httpx doesn’t require compiling C extensions or installing system dependencies. It’s pure Python with optional performance enhancements.

Headers are the bread crumbs out of the rate limit forest

Making GET Requests

GET requests are the foundation of HTTP. They retrieve data without side effects, and httpx makes them effortless. The basic pattern is identical to requests, but httpx adds subtle improvements like better error handling and automatic timeout management.

# get_requests.py
import httpx

# Simple GET request
response = httpx.get("https://jsonplaceholder.typicode.com/users")
print(f"Status: {response.status_code}")
print(f"Content-Type: {response.headers['content-type']}")
print(f"First user: {response.json()[0]['name']}")

# GET with query parameters
params = {"userId": 1}
response = httpx.get("https://jsonplaceholder.typicode.com/posts", params=params)
print(f"Posts for user 1: {len(response.json())}")

# GET with custom headers
headers = {"User-Agent": "MyApp/1.0"}
response = httpx.get("https://httpbin.org/headers", headers=headers)
print(response.json())

Notice how httpx handles query parameters naturally through the `params` dictionary. You don’t manually construct query strings or worry about URL encoding — httpx handles that behind the scenes. Headers work the same way, accepting a dictionary that httpx merges with the default headers. This consistent API means you can focus on your application logic instead of HTTP bookkeeping.

Making POST Requests

POST requests send data to the server. httpx supports multiple ways to send data: form-encoded, JSON, raw bytes, or streaming. Let’s explore the most common patterns.

# post_requests.py
import httpx

# POST with JSON data
client = httpx.Client()
data = {
    "title": "New Post",
    "body": "This is a test post",
    "userId": 1
}
response = client.post(
    "https://jsonplaceholder.typicode.com/posts",
    json=data
)
print(f"Created post with ID: {response.json()['id']}")

# POST with form data
form_data = {"username": "john", "password": "secret123"}
response = client.post(
    "https://httpbin.org/post",
    data=form_data
)
print(f"Form post status: {response.status_code}")

# POST with custom timeout
try:
    response = client.post(
        "https://httpbin.org/delay/10",
        timeout=5.0
    )
except httpx.TimeoutException:
    print("Request timed out after 5 seconds")

client.close()

When making POST requests, use the `json` parameter for JSON data and the `data` parameter for form-encoded data. httpx automatically sets the correct Content-Type header for you. The `Client()` context manager maintains connection pooling across multiple requests, which is more efficient than using module-level functions for repeated requests. Timeouts are crucial for production code — they prevent your application from hanging if a server stops responding.

Exponential backoff: be patient, then exponentially more patient

Using Async with httpx

This is where httpx truly shines. Async support is built in from the ground up, not bolted on as an afterthought. When you need to handle multiple concurrent requests, async/await patterns let you handle hundreds of concurrent connections with minimal memory overhead — something that would require threading or multiprocessing with requests.

# async_requests.py
import asyncio
import httpx

async def fetch_posts(user_id):
    """Fetch posts for a specific user"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://jsonplaceholder.typicode.com/posts?userId={user_id}"
        )
        return response.json()

async def fetch_multiple_users():
    """Fetch posts for multiple users concurrently"""
    tasks = [
        fetch_posts(user_id)
        for user_id in range(1, 6)
    ]
    results = await asyncio.gather(*tasks)

    for i, posts in enumerate(results, 1):
        print(f"User {i}: {len(posts)} posts")

# Run the async function
asyncio.run(fetch_multiple_users())

The `AsyncClient()` context manager handles resource cleanup automatically. Using `asyncio.gather()`, we fetch posts for five users concurrently in roughly the time it takes to fetch one. This pattern scales to thousands of concurrent requests without the overhead of creating threads. The key difference from synchronous code is minimal — just add `async` and `await` keywords.

# async_with_timeout.py
import asyncio
import httpx

async def fetch_with_timeout():
    """Fetch with explicit timeout configuration"""
    timeout = httpx.Timeout(10.0)  # 10 second timeout for all operations

    async with httpx.AsyncClient(timeout=timeout) as client:
        try:
            response = await client.get(
                "https://jsonplaceholder.typicode.com/posts/1"
            )
            print(f"Success: {response.status_code}")
        except httpx.TimeoutException:
            print("Request timed out")
        except httpx.RequestError as e:
            print(f"Network error: {e}")

asyncio.run(fetch_with_timeout())

Timeout handling in async contexts is critical. The `Timeout` object lets you set different timeouts for connection, read, write, and pool operations. This fine-grained control prevents your async application from hanging on unresponsive servers.

HTTP/2 Support

HTTP/2 is faster than HTTP/1.1 because it multiplexes multiple requests over a single connection and compresses headers. With httpx, HTTP/2 support is automatic for servers that support it. You don’t need to change your code — just have the h2 library installed.

# http2_support.py
import httpx

# HTTP/2 is automatic when available
response = httpx.get("https://httpbin.org/get")
print(f"HTTP Version: {response.http_version}")

# Force HTTP/1.1 if needed
client = httpx.Client(http2=False)
response = client.get("https://httpbin.org/get")
print(f"HTTP Version (forced 1.1): {response.http_version}")
client.close()

# Create client with explicit HTTP/2 support
client = httpx.Client(http2=True)
response = client.get("https://httpbin.org/get")
print(f"HTTP Version (with HTTP/2): {response.http_version}")
client.close()

The performance difference is subtle for single requests but becomes dramatic with concurrent requests over HTTP/2. Since HTTP/2 multiplexes requests on a single connection, you avoid the overhead of establishing multiple TCP connections. For APIs that support it, this can mean 20-50% faster performance in real-world scenarios.

Decorators beat manual retry loops every time

Timeouts and Error Handling

Production code needs robust error handling. httpx provides clear exceptions for different failure scenarios, making it easy to distinguish between network problems, timeouts, and server errors.

# error_handling.py
import httpx

def fetch_with_fallback(url, fallback_url):
    """Fetch from primary URL with fallback"""
    try:
        response = httpx.get(url, timeout=5.0)
        response.raise_for_status()  # Raise exception for bad status codes
        return response.json()
    except httpx.TimeoutException:
        print(f"Timeout on {url}, trying fallback")
        return httpx.get(fallback_url).json()
    except httpx.HTTPStatusError as e:
        print(f"HTTP error: {e.response.status_code}")
        raise
    except httpx.RequestError as e:
        print(f"Request error: {e}")
        raise

try:
    data = fetch_with_fallback(
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2"
    )
    print(f"Fetched post: {data['title']}")
except Exception as e:
    print(f"All attempts failed: {e}")

httpx organizes exceptions in a clear hierarchy. `RequestError` is the base class for all request-related errors. `TimeoutException` indicates a timeout (connection, read, write, or pool). `HTTPStatusError` means the server responded with an error status code (4xx or 5xx). Using `raise_for_status()` automatically raises an exception for bad status codes, similar to requests.

# advanced_timeouts.py
import httpx

# Granular timeout control
timeout = httpx.Timeout(
    timeout=10.0,  # Default timeout
    connect=5.0,   # Connection timeout
    read=10.0,     # Read timeout
    write=10.0,    # Write timeout
    pool=10.0      # Connection pool timeout
)

client = httpx.Client(timeout=timeout)

# Override timeout for specific request
try:
    response = client.get(
        "https://httpbin.org/delay/2",
        timeout=2.0  # Override client timeout
    )
    print(f"Response: {response.status_code}")
except httpx.TimeoutException:
    print("Custom timeout exceeded")
finally:
    client.close()

Different operations need different timeout values. Connection timeouts should be shorter (5-10 seconds), while read timeouts depend on the expected response size. For large downloads, you might need 30+ second read timeouts. httpx lets you configure each separately.

Streaming Responses

When dealing with large files or streaming APIs, loading the entire response into memory is inefficient. httpx supports streaming, letting you process responses chunk by chunk.

# streaming_responses.py
import httpx

# Stream a large response
with httpx.stream("GET", "https://httpbin.org/bytes/1024") as response:
    print(f"Status: {response.status_code}")
    print(f"Content-Length: {response.headers.get('content-length')}")

    # Process response in chunks
    for chunk in response.iter_bytes(chunk_size=256):
        print(f"Received {len(chunk)} bytes")

# Stream with iterator
with httpx.stream("GET", "https://httpbin.org/json") as response:
    for line in response.iter_lines():
        if line:
            print(f"Line: {line[:50]}...")

# Async streaming
import asyncio

async def async_stream():
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", "https://httpbin.org/bytes/512") as response:
            async for chunk in response.aiter_bytes(chunk_size=128):
                print(f"Async received {len(chunk)} bytes")

asyncio.run(async_stream())

Streaming is essential for production applications that download large files or handle streaming APIs. The `iter_bytes()` method gives you raw bytes, while `iter_lines()` automatically splits on newlines — useful for newline-delimited JSON APIs. Async streaming with `aiter_bytes()` and `aiter_lines()` works the same way in async contexts.

Know your limits, respect the headers, fetch responsibly

Real-Life Example: Async API Data Aggregator

Let’s build a practical example that combines async requests, error handling, and structured data processing. Imagine you’re aggregating data from multiple APIs and want to do it efficiently.

# api_aggregator.py
import asyncio
import httpx
from datetime import datetime

class APIAggregator:
    """Aggregates data from multiple APIs concurrently"""

    def __init__(self, max_concurrent=5):
        self.max_concurrent = max_concurrent
        self.timeout = httpx.Timeout(10.0)

    async def fetch_post(self, client, post_id):
        """Fetch a single post"""
        try:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/posts/{post_id}",
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except httpx.RequestError as e:
            print(f"Error fetching post {post_id}: {e}")
            return None

    async def fetch_user_comments(self, client, user_id):
        """Fetch comments for a user"""
        try:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/comments?email=*@{user_id}.com",
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except httpx.RequestError as e:
            print(f"Error fetching comments: {e}")
            return []

    async def aggregate(self):
        """Aggregate data from multiple endpoints"""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            # Fetch posts concurrently
            post_tasks = [
                self.fetch_post(client, i)
                for i in range(1, 6)
            ]
            posts = await asyncio.gather(*post_tasks)

            # Fetch comments concurrently
            comment_tasks = [
                self.fetch_user_comments(client, i)
                for i in range(1, 3)
            ]
            comments = await asyncio.gather(*comment_tasks)

            return {
                "timestamp": datetime.now().isoformat(),
                "posts_fetched": len([p for p in posts if p]),
                "total_comments": sum(len(c) for c in comments if c),
                "sample_post": posts[0] if posts else None
            }

# Run the aggregator
async def main():
    aggregator = APIAggregator()
    results = await aggregator.aggregate()

    print(f"Aggregation completed at {results['timestamp']}")
    print(f"Posts fetched: {results['posts_fetched']}")
    print(f"Total comments: {results['total_comments']}")
    print(f"First post title: {results['sample_post']['title']}")

asyncio.run(main())

This example demonstrates several httpx patterns: creating an async client, managing concurrent requests with proper error handling, using consistent timeouts across all requests, and returning structured data. The `APIAggregator` class is reusable and extensible — you could add caching, retry logic, or progress tracking. In production, you’d likely add logging and more sophisticated error recovery.

Frequently Asked Questions

Q: Is httpx a drop-in replacement for requests?
A: Mostly yes. The synchronous API is nearly identical, so most requests code works with httpx unchanged. However, httpx is stricter about some behaviors (like automatic redirects) and has some API differences. Test thoroughly before migrating production code.

Q: Do I need to use async?
A: No. httpx works great synchronously, and you only need async when handling many concurrent requests. Start with synchronous code and migrate to async if profiling shows it’s beneficial.

Q: What’s the performance difference between httpx and requests?
A: For single requests, performance is similar. For concurrent requests, httpx with async is dramatically faster because it avoids thread overhead. HTTP/2 support also improves performance for compatible servers.

Q: How do I handle cookies and sessions?
A: httpx clients maintain cookies automatically. Use a persistent client for multiple requests to the same host to keep cookies and connection pooling active across requests.

Q: Can I use httpx with Django or Flask?
A: Yes, but use the synchronous API in request handlers since WSGI is synchronous. Use async httpx in ASGI applications like FastAPI or Django Async Views.

Q: How do I set up authentication?
A: httpx supports multiple auth methods. For basic auth: `httpx.get(url, auth=(“user”, “pass”))`. For bearer tokens: `headers={“Authorization”: “Bearer token”}`. For custom auth, create a subclass of `httpx.Auth`.

Conclusion

httpx represents the future of HTTP requests in Python. It combines the simplicity of requests with modern features like async/await, HTTP/2, and better timeout handling. Whether you’re building a simple API client or managing complex concurrent request workflows, httpx has the tools you need without unnecessary complexity.

Start by installing httpx and replacing requests in a non-critical project. You’ll quickly discover why developers are switching. For more details, comprehensive documentation is available at httpx.readthedocs.io.

Related Articles

• How To Handle API Rate Limits and Retry Logic in Python 3
• How To Mock API Calls in Python
• How To Build a REST API with FastAPI in Python
• How To Use the OpenAI API with Python
• How To Create a Simple REST API With Python Flask in 2026
• How to Build a Twitter Bot with Python and Twitter API v2
• How To Read and Update Smartsheets With Python 3

Related Python Tutorials

• How To Handle API Rate Limits and Retry Logic in Python 3
• How To Mock API Calls in Python
• How To Use the OpenAI API with Python

Skill Level: Intermediate

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

Introduction to Modern HTTP Requests

For years, the Python requests library has been the go-to solution for making HTTP requests. While requests is powerful and user-friendly, it has limitations that modern Python developers encounter daily. It doesn’t support async operations natively, lacks HTTP/2 support, and can feel sluggish when you’re handling dozens of concurrent requests. If you’ve found yourself wrestling with requests’ synchronous nature or spinning up ThreadPoolExecutor just to manage multiple requests, you’re not alone. Many developers hit a wall where they need something more capable.

Enter httpx — a modern HTTP client that drops in as a replacement for requests while adding powerful features like native async/await support, HTTP/2 capabilities, and streaming responses. The best part? If you already know requests, you’ll feel right at home with httpx. The API is remarkably similar, which means you can start using it immediately without relearning everything. We’ll cover installation, basic usage, async patterns, and real-world examples that show why httpx is becoming the preferred choice for new Python projects.

In this guide, we’ll walk through everything you need to know about httpx. We’ll start with a quick example to get you up and running, explore what makes httpx special compared to other HTTP libraries, and then dive into practical patterns you can use in your own projects. Whether you’re building a simple API client or managing complex async workflows, httpx has the tools you need. Let’s get started.

Quick Example: GET Request in 5 Lines

# quick_get.py
import httpx

response = httpx.get("https://jsonplaceholder.typicode.com/posts/1")
print(response.status_code)
print(response.json())

That’s it. If you’ve used requests before, you already know httpx. The synchronous API is nearly identical, but under the hood, httpx brings modern features to the table. Now let’s explore what makes httpx more powerful than its predecessor.

httpx handles synchronous and asynchronous requests from a single library

What is httpx and Why Use It?

httpx is a modern HTTP client library for Python that combines the simplicity of requests with advanced features like async support, HTTP/2, and more sophisticated timeout handling. Created by Tom Christie (the same developer behind Starlette and FastAPI), httpx is built on a foundation that understands modern Python development patterns. It’s not just a requests replacement — it’s a redesign based on everything we’ve learned about making HTTP libraries in the 2020s.

The key differences matter when you’re building real applications. Unlike requests, httpx supports both synchronous and asynchronous code from the same library. You don’t need to install separate packages or maintain multiple code paths. It supports HTTP/2 by default, which means better performance for services that support it. It has built-in connection pooling, proper async context managers, and a cleaner API that feels more Pythonic.

Let’s compare httpx to other HTTP libraries in the Python ecosystem:

httpx shines when you need synchronous and asynchronous code in the same project. Unlike aiohttp, which requires AsyncIO from the start, httpx lets you start simple and scale to async when you need it. Unlike requests, httpx doesn’t force you into threading patterns when you want to handle multiple requests concurrently. It’s the bridge between requests’ simplicity and aiohttp’s power.

Installing httpx

Installation is straightforward. httpx is available on PyPI and installs cleanly without forcing dependencies on you. For basic functionality, you only need one command.

# install_httpx.sh
pip install httpx

If you want HTTP/2 support with better performance, you can install the optional dependencies. The h2 library handles HTTP/2 protocol details, while httpcore provides the underlying transport layer.

# install_httpx_with_http2.sh
pip install httpx[http2]

That’s all you need. Unlike some HTTP libraries, httpx doesn’t require compiling C extensions or installing system dependencies. It’s pure Python with optional performance enhancements.

POST, GET, PUT, DELETE — all methods work seamlessly with httpx

Making GET Requests

GET requests are the foundation of HTTP. They retrieve data without side effects, and httpx makes them effortless. The basic pattern is identical to requests, but httpx adds subtle improvements like better error handling and automatic timeout management.

# get_requests.py
import httpx

# Simple GET request
response = httpx.get("https://jsonplaceholder.typicode.com/users")
print(f"Status: {response.status_code}")
print(f"Content-Type: {response.headers['content-type']}")
print(f"First user: {response.json()[0]['name']}")

# GET with query parameters
params = {"userId": 1}
response = httpx.get("https://jsonplaceholder.typicode.com/posts", params=params)
print(f"Posts for user 1: {len(response.json())}")

# GET with custom headers
headers = {"User-Agent": "MyApp/1.0"}
response = httpx.get("https://httpbin.org/headers", headers=headers)
print(response.json())

Notice how httpx handles query parameters naturally through the `params` dictionary. You don’t manually construct query strings or worry about URL encoding — httpx handles that behind the scenes. Headers work the same way, accepting a dictionary that httpx merges with the default headers. This consistent API means you can focus on your application logic instead of HTTP bookkeeping.

Making POST Requests

POST requests send data to the server. httpx supports multiple ways to send data: form-encoded, JSON, raw bytes, or streaming. Let’s explore the most common patterns.

# post_requests.py
import httpx

# POST with JSON data
client = httpx.Client()
data = {
    "title": "New Post",
    "body": "This is a test post",
    "userId": 1
}
response = client.post(
    "https://jsonplaceholder.typicode.com/posts",
    json=data
)
print(f"Created post with ID: {response.json()['id']}")

# POST with form data
form_data = {"username": "john", "password": "secret123"}
response = client.post(
    "https://httpbin.org/post",
    data=form_data
)
print(f"Form post status: {response.status_code}")

# POST with custom timeout
try:
    response = client.post(
        "https://httpbin.org/delay/10",
        timeout=5.0
    )
except httpx.TimeoutException:
    print("Request timed out after 5 seconds")

client.close()

When making POST requests, use the `json` parameter for JSON data and the `data` parameter for form-encoded data. httpx automatically sets the correct Content-Type header for you. The `Client()` context manager maintains connection pooling across multiple requests, which is more efficient than using module-level functions for repeated requests. Timeouts are crucial for production code — they prevent your application from hanging if a server stops responding.

Connection pooling keeps your HTTP clients fast and memory-efficient

Using Async with httpx

This is where httpx truly shines. Async support is built in from the ground up, not bolted on as an afterthought. When you need to handle multiple concurrent requests, async/await patterns let you handle hundreds of concurrent connections with minimal memory overhead — something that would require threading or multiprocessing with requests.

# async_requests.py
import asyncio
import httpx

async def fetch_posts(user_id):
    """Fetch posts for a specific user"""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://jsonplaceholder.typicode.com/posts?userId={user_id}"
        )
        return response.json()

async def fetch_multiple_users():
    """Fetch posts for multiple users concurrently"""
    tasks = [
        fetch_posts(user_id)
        for user_id in range(1, 6)
    ]
    results = await asyncio.gather(*tasks)

    for i, posts in enumerate(results, 1):
        print(f"User {i}: {len(posts)} posts")

# Run the async function
asyncio.run(fetch_multiple_users())

The `AsyncClient()` context manager handles resource cleanup automatically. Using `asyncio.gather()`, we fetch posts for five users concurrently in roughly the time it takes to fetch one. This pattern scales to thousands of concurrent requests without the overhead of creating threads. The key difference from synchronous code is minimal — just add `async` and `await` keywords.

# async_with_timeout.py
import asyncio
import httpx

async def fetch_with_timeout():
    """Fetch with explicit timeout configuration"""
    timeout = httpx.Timeout(10.0)  # 10 second timeout for all operations

    async with httpx.AsyncClient(timeout=timeout) as client:
        try:
            response = await client.get(
                "https://jsonplaceholder.typicode.com/posts/1"
            )
            print(f"Success: {response.status_code}")
        except httpx.TimeoutException:
            print("Request timed out")
        except httpx.RequestError as e:
            print(f"Network error: {e}")

asyncio.run(fetch_with_timeout())

Timeout handling in async contexts is critical. The `Timeout` object lets you set different timeouts for connection, read, write, and pool operations. This fine-grained control prevents your async application from hanging on unresponsive servers.

HTTP/2 Support

HTTP/2 is faster than HTTP/1.1 because it multiplexes multiple requests over a single connection and compresses headers. With httpx, HTTP/2 support is automatic for servers that support it. You don’t need to change your code — just have the h2 library installed.

# http2_support.py
import httpx

# HTTP/2 is automatic when available
response = httpx.get("https://httpbin.org/get")
print(f"HTTP Version: {response.http_version}")

# Force HTTP/1.1 if needed
client = httpx.Client(http2=False)
response = client.get("https://httpbin.org/get")
print(f"HTTP Version (forced 1.1): {response.http_version}")
client.close()

# Create client with explicit HTTP/2 support
client = httpx.Client(http2=True)
response = client.get("https://httpbin.org/get")
print(f"HTTP Version (with HTTP/2): {response.http_version}")
client.close()

The performance difference is subtle for single requests but becomes dramatic with concurrent requests over HTTP/2. Since HTTP/2 multiplexes requests on a single connection, you avoid the overhead of establishing multiple TCP connections. For APIs that support it, this can mean 20-50% faster performance in real-world scenarios.

HTTP/2 multiplexing lets httpx handle multiple requests on one connection

Timeouts and Error Handling

Production code needs robust error handling. httpx provides clear exceptions for different failure scenarios, making it easy to distinguish between network problems, timeouts, and server errors.

# error_handling.py
import httpx

def fetch_with_fallback(url, fallback_url):
    """Fetch from primary URL with fallback"""
    try:
        response = httpx.get(url, timeout=5.0)
        response.raise_for_status()  # Raise exception for bad status codes
        return response.json()
    except httpx.TimeoutException:
        print(f"Timeout on {url}, trying fallback")
        return httpx.get(fallback_url).json()
    except httpx.HTTPStatusError as e:
        print(f"HTTP error: {e.response.status_code}")
        raise
    except httpx.RequestError as e:
        print(f"Request error: {e}")
        raise

try:
    data = fetch_with_fallback(
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2"
    )
    print(f"Fetched post: {data['title']}")
except Exception as e:
    print(f"All attempts failed: {e}")

httpx organizes exceptions in a clear hierarchy. `RequestError` is the base class for all request-related errors. `TimeoutException` indicates a timeout (connection, read, write, or pool). `HTTPStatusError` means the server responded with an error status code (4xx or 5xx). Using `raise_for_status()` automatically raises an exception for bad status codes, similar to requests.

# advanced_timeouts.py
import httpx

# Granular timeout control
timeout = httpx.Timeout(
    timeout=10.0,  # Default timeout
    connect=5.0,   # Connection timeout
    read=10.0,     # Read timeout
    write=10.0,    # Write timeout
    pool=10.0      # Connection pool timeout
)

client = httpx.Client(timeout=timeout)

# Override timeout for specific request
try:
    response = client.get(
        "https://httpbin.org/delay/2",
        timeout=2.0  # Override client timeout
    )
    print(f"Response: {response.status_code}")
except httpx.TimeoutException:
    print("Custom timeout exceeded")
finally:
    client.close()

Different operations need different timeout values. Connection timeouts should be shorter (5-10 seconds), while read timeouts depend on the expected response size. For large downloads, you might need 30+ second read timeouts. httpx lets you configure each separately.

Streaming Responses

When dealing with large files or streaming APIs, loading the entire response into memory is inefficient. httpx supports streaming, letting you process responses chunk by chunk.

# streaming_responses.py
import httpx

# Stream a large response
with httpx.stream("GET", "https://httpbin.org/bytes/1024") as response:
    print(f"Status: {response.status_code}")
    print(f"Content-Length: {response.headers.get('content-length')}")

    # Process response in chunks
    for chunk in response.iter_bytes(chunk_size=256):
        print(f"Received {len(chunk)} bytes")

# Stream with iterator
with httpx.stream("GET", "https://httpbin.org/json") as response:
    for line in response.iter_lines():
        if line:
            print(f"Line: {line[:50]}...")

# Async streaming
import asyncio

async def async_stream():
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", "https://httpbin.org/bytes/512") as response:
            async for chunk in response.aiter_bytes(chunk_size=128):
                print(f"Async received {len(chunk)} bytes")

asyncio.run(async_stream())

Streaming is essential for production applications that download large files or handle streaming APIs. The `iter_bytes()` method gives you raw bytes, while `iter_lines()` automatically splits on newlines — useful for newline-delimited JSON APIs. Async streaming with `aiter_bytes()` and `aiter_lines()` works the same way in async contexts.

Streaming responses keep memory usage constant regardless of file size

Real-Life Example: Async API Data Aggregator

Let’s build a practical example that combines async requests, error handling, and structured data processing. Imagine you’re aggregating data from multiple APIs and want to do it efficiently.

# api_aggregator.py
import asyncio
import httpx
from datetime import datetime

class APIAggregator:
    """Aggregates data from multiple APIs concurrently"""

    def __init__(self, max_concurrent=5):
        self.max_concurrent = max_concurrent
        self.timeout = httpx.Timeout(10.0)

    async def fetch_post(self, client, post_id):
        """Fetch a single post"""
        try:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/posts/{post_id}",
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except httpx.RequestError as e:
            print(f"Error fetching post {post_id}: {e}")
            return None

    async def fetch_user_comments(self, client, user_id):
        """Fetch comments for a user"""
        try:
            response = await client.get(
                f"https://jsonplaceholder.typicode.com/comments?email=*@{user_id}.com",
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except httpx.RequestError as e:
            print(f"Error fetching comments: {e}")
            return []

    async def aggregate(self):
        """Aggregate data from multiple endpoints"""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            # Fetch posts concurrently
            post_tasks = [
                self.fetch_post(client, i)
                for i in range(1, 6)
            ]
            posts = await asyncio.gather(*post_tasks)

            # Fetch comments concurrently
            comment_tasks = [
                self.fetch_user_comments(client, i)
                for i in range(1, 3)
            ]
            comments = await asyncio.gather(*comment_tasks)

            return {
                "timestamp": datetime.now().isoformat(),
                "posts_fetched": len([p for p in posts if p]),
                "total_comments": sum(len(c) for c in comments if c),
                "sample_post": posts[0] if posts else None
            }

# Run the aggregator
async def main():
    aggregator = APIAggregator()
    results = await aggregator.aggregate()

    print(f"Aggregation completed at {results['timestamp']}")
    print(f"Posts fetched: {results['posts_fetched']}")
    print(f"Total comments: {results['total_comments']}")
    print(f"First post title: {results['sample_post']['title']}")

asyncio.run(main())

This example demonstrates several httpx patterns: creating an async client, managing concurrent requests with proper error handling, using consistent timeouts across all requests, and returning structured data. The `APIAggregator` class is reusable and extensible — you could add caching, retry logic, or progress tracking. In production, you’d likely add logging and more sophisticated error recovery.

Frequently Asked Questions

Q: Is httpx a drop-in replacement for requests?
A: Mostly yes. The synchronous API is nearly identical, so most requests code works with httpx unchanged. However, httpx is stricter about some behaviors (like automatic redirects) and has some API differences. Test thoroughly before migrating production code.

Q: Do I need to use async?
A: No. httpx works great synchronously, and you only need async when handling many concurrent requests. Start with synchronous code and migrate to async if profiling shows it’s beneficial.

Q: What’s the performance difference between httpx and requests?
A: For single requests, performance is similar. For concurrent requests, httpx with async is dramatically faster because it avoids thread overhead. HTTP/2 support also improves performance for compatible servers.

Q: How do I handle cookies and sessions?
A: httpx clients maintain cookies automatically. Use a persistent client for multiple requests to the same host to keep cookies and connection pooling active across requests.

Q: Can I use httpx with Django or Flask?
A: Yes, but use the synchronous API in request handlers since WSGI is synchronous. Use async httpx in ASGI applications like FastAPI or Django Async Views.

Q: How do I set up authentication?
A: httpx supports multiple auth methods. For basic auth: `httpx.get(url, auth=(“user”, “pass”))`. For bearer tokens: `headers={“Authorization”: “Bearer token”}`. For custom auth, create a subclass of `httpx.Auth`.

Conclusion

httpx represents the future of HTTP requests in Python. It combines the simplicity of requests with modern features like async/await, HTTP/2, and better timeout handling. Whether you’re building a simple API client or managing complex concurrent request workflows, httpx has the tools you need without unnecessary complexity.

Start by installing httpx and replacing requests in a non-critical project. You’ll quickly discover why developers are switching. For more details, comprehensive documentation is available at httpx.readthedocs.io.

Related Articles

• How To Make Concurrent HTTP Requests with aiohttp in Python
• How To Use Python requests: Sessions, Retry, and Timeout
• How To Use Python Requests To Call REST APIs With Authentication
• How To Use Python MCP Server for AI Tool Integration
• How To Use Python litellm for Multi-Model LLM Integration
• How To Use Python instructor for Structured LLM Outputs
• Getting Historical Stock Data Using Python 3

Related Python Tutorials

• How To Make Concurrent HTTP Requests with aiohttp in Python
• How To Use Python responses for Mocking HTTP Requests in Tests
• How To Use Python requests: Sessions, Retry, and Timeout

Skill Level: Intermediate

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

Why Password Security Matters in Your Applications

Every year, millions of passwords are exposed in data breaches. When companies store passwords in plain text or with weak encryption methods, attackers who gain access to the database can immediately use those credentials against users. A single security oversight can compromise not just one user, but your entire application’s integrity and your reputation. The good news? Protecting passwords isn’t complicated — bcrypt makes it remarkably straightforward to implement enterprise-grade password security in your Python applications today.

Bcrypt is a purpose-built password hashing library that automatically handles the complexity of secure password storage. Unlike generic hashing functions, bcrypt incorporates salt generation, adaptive work factors, and built-in protections against common attacks. Whether you’re building a small project or scaling to millions of users, bcrypt gives you the same security guarantees with just a few lines of code.

In this guide, we’ll explore how bcrypt works from first principles, then move through practical implementations including user registration, password verification, and real-world patterns you can use immediately. By the end, you’ll understand why bcrypt is the industry standard and how to integrate it into your projects with confidence.

Quick Example: Hash and Verify in 10 Lines

Before diving into the theory, let’s see bcrypt in action. This example demonstrates the complete workflow — hashing a password and then verifying a user’s input against that stored hash. It’s this simple and this secure.

# quick_hash.py
import bcrypt

password = "MySecurePassword123!"
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
print("Hashed:", hashed)

user_input = "MySecurePassword123!"
is_valid = bcrypt.checkpw(user_input.encode(), hashed)
print("Password match:", is_valid)

Hashed: b'$2b$12$abcdefghijklmnopqrstuvwxyzAbCdEfGhIjKlMnOpQrStUvWxYz'
Password match: True

That’s the essence of password hashing with bcrypt. The hashpw() function handles salt generation and hashing internally, while checkpw() verifies inputs without needing to decrypt anything. This simplicity masks sophisticated security mechanisms working behind the scenes.

What is Password Hashing and Why Use bcrypt?

Password hashing is a one-way cryptographic function that transforms a plaintext password into an irreversible string of characters. The database stores only the hash, not the original password. When a user logs in, the application hashes their input and compares it to the stored hash — if they match, the password is correct. If attackers compromise the database, they get hashes, not passwords.

Not all hashing functions are suitable for passwords. General-purpose algorithms like MD5, SHA1, and even SHA256 are too fast — an attacker with access to your password hashes can run billions of guesses per second using GPU clusters. Bcrypt solves this by intentionally being slow and computationally expensive, making brute-force attacks impractical. Here’s how different approaches compare:

Bcrypt’s adaptive cost factor is crucial — as hardware becomes faster, you can increase the work factor to keep password cracking slow. A hash that takes 0.2 seconds to compute today will still take 0.2 seconds tomorrow, even as computers improve. This forward-looking design is why bcrypt remains relevant decades after its creation.

Installing bcrypt

Bcrypt requires a simple pip installation. It’s a compiled C extension for performance, and the package handles all dependency management across operating systems. Most Python environments will install it without issues, though some systems may need development headers for the C compiler.

# install_bcrypt.py
import subprocess
import sys

# Install bcrypt
subprocess.check_call([sys.executable, "-m", "pip", "install", "bcrypt"])

# Verify installation
import bcrypt
print("Bcrypt version:", bcrypt.__version__)
print("Installation successful!")

Collecting bcrypt
  Downloading bcrypt-4.1.2-cp312-cp312-linux_x86_64.whl
Installing collected packages: bcrypt
Successfully installed bcrypt-4.1.2
Bcrypt version: 4.1.2
Installation successful!

Once installed, bcrypt is immediately available for import. We recommend adding bcrypt to your project’s requirements.txt or pyproject.toml file so it’s installed automatically for anyone who clones your repository. The package is maintained actively and released regularly with security updates, so keep it current with pip install --upgrade bcrypt periodically.

Investigating hash functions like a crime scene — following the trail that only goes one way

Hashing a Password

The hashing process in bcrypt is deceptively simple from a user perspective, but implements several sophisticated techniques internally. When you call hashpw(), bcrypt generates a cryptographically random salt, applies the Blowfish cipher multiple times based on the cost factor, and returns a single string containing all the information needed to verify passwords later.

Here’s the standard pattern for creating password hashes when users register or change their password:

# hash_password_demo.py
import bcrypt

def hash_password(password: str) -> str:
    """
    Hash a plaintext password using bcrypt.

    Args:
        password: The user's plaintext password

    Returns:
        The bcrypt hash as a string (can be stored in database)
    """
    # Encode string to bytes (bcrypt requires bytes)
    password_bytes = password.encode('utf-8')

    # Generate salt and hash in one operation
    # Default cost factor is 12
    hashed = bcrypt.hashpw(password_bytes, bcrypt.gensalt())

    # Return as string for database storage
    return hashed.decode('utf-8')

# Example usage
user_password = "SecurePass123!"
stored_hash = hash_password(user_password)
print(f"Original: {user_password}")
print(f"Stored:   {stored_hash}")
print(f"Length:   {len(stored_hash)} characters")

Original: SecurePass123!
Stored:   $2b$12$QaNu.rBvNp8zXrLbSKc.MOaVfVL6qfKfhIU3SYvKzGdpLp4I8TS6O
Length:   60 characters

Every hash is exactly 60 characters long, making it database-friendly to store in a VARCHAR(60) column. The hash contains embedded information: the algorithm identifier ($2b$), the cost factor ($12$), and the salt. You never need to store these separately — everything is self-contained in that single string.

A critical consideration: passwords must be encoded to bytes before hashing. Python 3 strings are Unicode by default, but bcrypt operates on bytes. Always use .encode('utf-8') or specify the encoding explicitly. Different encodings will produce different hashes, which is why consistency matters — the verification step must use the same encoding.

Verifying a Password

When a user logs in, you don’t decrypt the stored hash or compare strings directly. Instead, you hash the incoming password and let bcrypt compare the hashes internally using timing-safe comparison. This prevents timing attacks where attackers measure response times to gain information about the hash.

Here’s how to verify passwords securely:

# verify_password_demo.py
import bcrypt

def verify_password(provided_password: str, stored_hash: str) -> bool:
    """
    Verify a plaintext password against a bcrypt hash.

    Args:
        provided_password: Password the user provided at login
        stored_hash: The bcrypt hash from the database

    Returns:
        True if the password matches, False otherwise
    """
    # Encode both to bytes
    provided_bytes = provided_password.encode('utf-8')
    stored_bytes = stored_hash.encode('utf-8')

    # checkpw handles timing-safe comparison internally
    return bcrypt.checkpw(provided_bytes, stored_bytes)

# Example: Simulating a login attempt
stored_hash = "$2b$12$QaNu.rBvNp8zXrLbSKc.MOaVfVL6qfKfhIU3SYvKzGdpLp4I8TS6O"

# Correct password
attempt1 = "SecurePass123!"
print(f"Attempt 1 ('{attempt1}'): {verify_password(attempt1, stored_hash)}")

# Wrong password
attempt2 = "WrongPassword123"
print(f"Attempt 2 ('{attempt2}'): {verify_password(attempt2, stored_hash)}")

# Close but not exact
attempt3 = "SecurePass123"
print(f"Attempt 3 ('{attempt3}'): {verify_password(attempt3, stored_hash)}")

Attempt 1 ('SecurePass123!'): True
Attempt 2 ('WrongPassword123'): False
Attempt 3 ('SecurePass123'): False

Notice that even a single character difference results in verification failure. Bcrypt’s comparison is exact — there’s no “close enough” or partial matches. This is intentional: password verification must be binary.

The checkpw() function uses constant-time comparison, meaning it takes the same amount of time regardless of where a mismatch occurs. A naive string comparison might return immediately upon finding the first different character, but bcrypt compares the entire hash every time. This prevents attackers from using timing measurements to guess parts of the hash.

Following the principle of least privilege — storing only what you absolutely need

Understanding Salt and Work Factor

Bcrypt’s security relies on two mechanisms working together: salt and cost factor. The salt is a random string added to each password before hashing, ensuring identical passwords produce different hashes. Without salt, an attacker could create a rainbow table — a precomputed mapping of common passwords to their hashes — and look up compromised passwords instantly.

The cost factor (work factor) determines how many rounds of hashing bcrypt performs. A higher cost factor makes hashing slower, which also makes brute-force attacks slower. Each increment doubles the computational cost. Let’s explore these concepts:

# salt_cost_demo.py
import bcrypt
import time

password = "TestPassword123"
password_bytes = password.encode('utf-8')

# Demonstrate different cost factors
print("Cost Factor Performance Analysis:")
print("-" * 50)

for cost in [10, 12, 14]:
    start = time.time()
    hashed = bcrypt.hashpw(password_bytes, bcrypt.gensalt(rounds=cost))
    elapsed = time.time() - start

    print(f"Cost {cost}: {elapsed:.3f}s - Hash: {hashed.decode()[:30]}...")

print("\nDifferent salts, same password:")
print("-" * 50)

# Same password hashed multiple times produces different results
for i in range(3):
    hashed = bcrypt.hashpw(password_bytes, bcrypt.gensalt())
    print(f"Hash {i+1}: {hashed.decode()}")

Cost Factor Performance Analysis:
--------------------------------------------------
Cost 10: 0.089s - Hash: $2b$10$abc123defghijklmnopqrst...
Cost 12: 0.324s - Hash: $2b$12$def456ghijklmnopqrstuv...
Cost 14: 1.274s - Hash: $2b$14$ghi789jklmnopqrstuvwx...

Different salts, same password:
--------------------------------------------------
Hash 1: $2b$12$aBcDeFgHiJkLmNoPqRsT.uVwXyZ123456789012345678901a
Hash 2: $2b$12$BcDeFgHiJkLmNoPqRsT.uVwXyZ123456789012345678901aB
Hash 3: $2b$12$CdEfGhIjKlMnOpQrStUv.wXyZ123456789012345678901aBc

Notice how cost factor 10 is fast but cost factor 14 takes over a second. In production, you need to balance security with user experience — a login taking 5 seconds would frustrate users. The default cost of 12 (0.2-0.3 seconds) is a proven sweet spot for most applications. As hardware improves over time, you can increase the cost factor in your code to maintain the same hashing duration.

Each hash includes its salt and cost factor as part of the output string. The format is: $2b$COST$SALT_AND_HASH. When you verify a password with checkpw(), bcrypt extracts this information from the stored hash automatically, ensuring verification uses the same parameters as the original hashing. You never need to manage salt or cost separately.

Adjusting the Cost Factor

While the default cost of 12 works well, production applications often adjust this based on their specific needs. Services with high login volume might use cost 10 to keep latency low, while security-critical applications might use cost 13 or 14. The key is measuring and balancing performance against security.

Here’s how to implement configurable cost factors and test performance in your application:

# configurable_cost.py
import bcrypt
import time
from typing import Optional

class PasswordManager:
    """Manages password hashing with configurable cost factor."""

    def __init__(self, cost_factor: int = 12):
        """
        Initialize with a specific cost factor.

        Args:
            cost_factor: Bcrypt cost factor (10-12 recommended, max 31)
        """
        if not 4 <= cost_factor <= 31:
            raise ValueError("Cost factor must be between 4 and 31")
        self.cost_factor = cost_factor

    def hash_password(self, password: str) -> str:
        """Hash a password with the configured cost factor."""
        password_bytes = password.encode('utf-8')
        salt = bcrypt.gensalt(rounds=self.cost_factor)
        hashed = bcrypt.hashpw(password_bytes, salt)
        return hashed.decode('utf-8')

    def verify_password(self, password: str, stored_hash: str) -> bool:
        """Verify a password against a stored hash."""
        try:
            password_bytes = password.encode('utf-8')
            stored_bytes = stored_hash.encode('utf-8')
            return bcrypt.checkpw(password_bytes, stored_bytes)
        except ValueError:
            # Invalid hash format
            return False

    def benchmark(self, test_password: str = "TestPassword123") -> float:
        """Measure how long hashing takes with current cost factor."""
        start = time.time()
        self.hash_password(test_password)
        return time.time() - start

# Test different cost factors
print("Finding optimal cost factor:")
print("-" * 50)

for cost in [10, 11, 12, 13]:
    manager = PasswordManager(cost_factor=cost)
    elapsed = manager.benchmark()
    print(f"Cost {cost}: {elapsed:.3f} seconds")

# Use production setting
print("\nProduction PasswordManager:")
print("-" * 50)
pm = PasswordManager(cost_factor=12)
pwd = "MySecurePassword!"
hashed = pm.hash_password(pwd)
print(f"Hash: {hashed}")
print(f"Verify: {pm.verify_password(pwd, hashed)}")

Finding optimal cost factor:
--------------------------------------------------
Cost 10: 0.087 seconds
Cost 11: 0.175 seconds
Cost 12: 0.334 seconds
Cost 13: 0.672 seconds

Production PasswordManager:
--------------------------------------------------
Hash: $2b$12$XyZ789aBcDeFgHiJkLmNo.pQrStUvWxYzAbCdEfGhIjKlMnOpQrSt
Verify: True

This PasswordManager class wraps bcrypt operations and makes cost factor configurable without changing application logic everywhere. You can update the default cost factor in one place, and all password operations use the new value. For applications with mixed cost factors in the database (from previous versions), bcrypt’s automatic parameter extraction handles this — older hashes with cost 10 verify correctly even if your application now uses cost 12.

Measuring twice, hashing once — balancing security with real-world performance constraints

Real-Life Example: User Registration CLI

Let’s build a practical user registration system that demonstrates password hashing in context. This example includes input validation, database simulation, and proper error handling — everything you’d need to adapt for a real application:

# user_registration_system.py
import bcrypt
import json
import os
from typing import Optional, Dict, Any

class UserDatabase:
    """Simulates a user database with file-based storage."""

    def __init__(self, db_file: str = "users.json"):
        self.db_file = db_file
        self._load_database()

    def _load_database(self) -> None:
        """Load users from file or create new database."""
        if os.path.exists(self.db_file):
            with open(self.db_file, 'r') as f:
                self.users = json.load(f)
        else:
            self.users = {}

    def _save_database(self) -> None:
        """Persist users to file."""
        with open(self.db_file, 'w') as f:
            json.dump(self.users, f, indent=2)

    def user_exists(self, username: str) -> bool:
        """Check if username already exists."""
        return username in self.users

    def register_user(self, username: str, email: str,
                     password: str) -> Dict[str, Any]:
        """
        Register a new user with password hashing.

        Returns:
            Dict with status and message
        """
        # Validation
        if self.user_exists(username):
            return {"success": False, "message": "Username already taken"}

        if len(password) < 8:
            return {"success": False,
                   "message": "Password must be 8+ characters"}

        if not email or '@' not in email:
            return {"success": False, "message": "Invalid email address"}

        # Hash password with cost factor 12
        password_bytes = password.encode('utf-8')
        salt = bcrypt.gensalt(rounds=12)
        password_hash = bcrypt.hashpw(password_bytes, salt).decode('utf-8')

        # Store user (never store plaintext password)
        self.users[username] = {
            "email": email,
            "password_hash": password_hash,
            "created": "2026-04-09"
        }
        self._save_database()

        return {"success": True,
               "message": f"User {username} registered successfully"}

    def authenticate_user(self, username: str,
                         password: str) -> Dict[str, Any]:
        """
        Verify username and password.

        Returns:
            Dict with authentication result
        """
        if not self.user_exists(username):
            return {"success": False, "message": "Invalid credentials"}

        user = self.users[username]
        password_bytes = password.encode('utf-8')
        stored_hash = user["password_hash"].encode('utf-8')

        # Use bcrypt to verify
        if bcrypt.checkpw(password_bytes, stored_hash):
            return {"success": True,
                   "message": f"Welcome back, {username}!"}
        else:
            return {"success": False, "message": "Invalid credentials"}

# Interactive CLI registration demo
if __name__ == "__main__":
    db = UserDatabase("demo_users.json")

    print("User Registration System")
    print("=" * 50)

    # Register a new user
    result = db.register_user(
        username="alice_dev",
        email="alice@example.com",
        password="AliceSecure123!"
    )
    print(f"Register: {result['message']}")

    # Attempt duplicate registration
    result = db.register_user(
        username="alice_dev",
        email="alice2@example.com",
        password="DifferentPass123"
    )
    print(f"Duplicate attempt: {result['message']}")

    # Correct login
    result = db.authenticate_user("alice_dev", "AliceSecure123!")
    print(f"Login (correct): {result['message']}")

    # Incorrect login
    result = db.authenticate_user("alice_dev", "WrongPassword123")
    print(f"Login (wrong): {result['message']}")

    # Cleanup
    if os.path.exists("demo_users.json"):
        os.remove("demo_users.json")

User Registration System
==================================================
Register: User alice_dev registered successfully
Duplicate attempt: Username already taken
Login (correct): Welcome back, alice_dev!
Login (wrong): Invalid credentials

This registration system demonstrates critical security patterns: passwords are never stored, only their bcrypt hashes are persisted. The validation ensures reasonable password strength before hashing. The authentication method uses bcrypt’s comparison, so timing attacks won’t reveal information about the hash. You can adapt this structure directly into web frameworks like Flask, Django, or FastAPI by replacing the file-based database with a proper database connection.

Catching attackers in the act — timing-safe comparisons leave no fingerprints

Frequently Asked Questions

Q: Can I decrypt a bcrypt hash to see the original password?

No, bcrypt is one-way cryptography by design. There’s no decryption function. This is a feature, not a limitation — if a password is forgotten, the user must reset it, not retrieve it. If someone claims to have recovered your original password from a bcrypt hash, they’re either lying or the system wasn’t actually using bcrypt.

Q: How long does a bcrypt hash stay valid?

The hash itself never expires. A hash created in 2015 will still verify passwords correctly in 2026. However, you can re-hash passwords when users log in with a higher cost factor. For example, migrate users to cost 13 only when they authenticate, avoiding the need to rehash all passwords at once.

Q: Is cost factor 12 always the right choice?

Cost 12 is a solid default that takes about 0.2-0.3 seconds on modern hardware. However, measure your specific use case: if you have thousands of concurrent logins, cost 10 might be necessary; if you’re a high-security application with low login volume, cost 13 or 14 is justified. The OWASP recommendation is to adjust cost so hashing takes 0.5-2 seconds on your target hardware.

Q: What if my bcrypt hash contains non-ASCII characters?

Bcrypt hashes only contain ASCII characters in the format $2a$, $2b$, or $2y$ followed by base-64 encoded characters. If you’re seeing non-ASCII, something went wrong during encoding. Always store bcrypt hashes as UTF-8 strings, and encode/decode carefully at boundaries with your database and external systems.

Q: Can I use the same salt for multiple passwords?

No, and bcrypt prevents this automatically. Every call to bcrypt.gensalt() generates a cryptographically unique salt. Using the same salt for multiple passwords would allow attackers to detect if two users have the same password. Let bcrypt generate a fresh salt for each password.

Q: Should I add my own salt on top of bcrypt’s salt?

No, that’s unnecessary and unlikely to improve security. Bcrypt’s salt is cryptographically sound. Adding additional layers of salting adds complexity without meaningful security benefit, and it could introduce vulnerabilities. Trust bcrypt’s design and focus on protecting your source code and deployment infrastructure instead.

Conclusion

Password hashing with bcrypt is one of the few areas where security and simplicity align perfectly. In just a few lines of code, you get enterprise-grade protection against the most common password attacks. The key patterns — always use bcrypt.hashpw() for storage and bcrypt.checkpw() for verification — should become second nature in any application handling user accounts.

Start with a cost factor of 12 and monitor your application’s performance. As hardware improves in the coming years, you can increase the cost factor to maintain security without redesigning your system. Test your implementation with common passwords and verify that both hashing and verification complete in acceptable timeframes for your users.

For deeper technical details and the latest version of bcrypt, visit the official bcrypt repository on GitHub. The PyCA project maintains bcrypt as part of Python’s cryptographic toolkit, with regular security audits and updates.

Related Articles

• How To Use Python MCP Server for AI Tool Integration
• How To Use Python litellm for Multi-Model LLM Integration
• How To Use Python instructor for Structured LLM Outputs
• How To Use Python LlamaIndex for Document Q&A
• How To Use PyJWT for JSON Web Tokens in Python
• How To Use Python cryptography for Encryption and Decryption
• Getting Historical Stock Data Using Python 3

Related Python Tutorials

• How To Use Python subprocess To Run Shell Commands Safely
• How To Write Unit Tests with pytest in Python
• How To Format Python Code with Black and isort

Intermediate

You have built a FastAPI application with clean endpoints and Pydantic validation. Everything works perfectly — until you realize anyone on the internet can call your API. No login required, no identity checks, no permission controls. This is the exact moment every API developer reaches: your endpoints need authentication, and you need it done properly without building a security framework from scratch.

FastAPI has built-in support for OAuth2 with Password flow and integrates cleanly with JSON Web Tokens (JWT). You will need three packages beyond FastAPI itself: python-jose[cryptography] for creating and verifying JWT tokens, passlib[bcrypt] for secure password hashing, and python-multipart for handling form data in the login endpoint. Install them with pip install python-jose[cryptography] passlib[bcrypt] python-multipart.

This tutorial walks you through the complete authentication flow: hashing and verifying passwords, creating JWT access tokens, protecting endpoints with dependency injection, extracting the current user from tokens, and implementing role-based access control. By the end, you will have a reusable authentication system that you can drop into any FastAPI project.

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

JWT Authentication in 30 Seconds

Part of the Python Web Frameworks Hub. See the full hub for related Python tutorials.

Part of the Modern Python AI Stack series. See the full tutorial hub for all 23 tutorials on LangGraph, MCP, Pydantic AI, Polars, FastAPI, Litestar, Typer, and more.

Here is the simplest possible protected endpoint in FastAPI. This gives you the core pattern before we build out the full system.

# quick_auth.py
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

@app.get("/protected")
def protected_route(token: str = Depends(oauth2_scheme)):
    # In a real app, you would decode and verify the JWT here
    if token != "secret-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"message": "You have access!", "token": token}

Output (without token):

{"detail": "Not authenticated"}

Output (with valid token in Authorization header):

{"message": "You have access!", "token": "secret-token"}

The OAuth2PasswordBearer dependency automatically extracts the token from the Authorization: Bearer <token> header. If the header is missing, FastAPI returns a 401 response before your function even runs. The tokenUrl parameter tells Swagger UI where to send login requests.

How JWT Authentication Works

Before writing the full implementation, it helps to understand the flow. JWT (JSON Web Token) authentication works in four steps: the client sends credentials (username and password), the server verifies them and returns a signed token, the client includes that token in every subsequent request, and the server verifies the token signature on each protected endpoint.

The JWT itself contains encoded JSON with the user’s identity (called “claims”), an expiration time, and a cryptographic signature. The server signs the token with a secret key, so any tampering is detectable without a database lookup. This makes JWT ideal for stateless APIs where you do not want to store session data on the server.

Password Hashing with Passlib

Never store passwords in plain text. Use passlib with the bcrypt algorithm to hash passwords before storing them and verify them during login.

# password_utils.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash a plain text password for storage."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Check a plain text password against a stored hash."""
    return pwd_context.verify(plain_password, hashed_password)

# Example usage
hashed = hash_password("my_secure_password")
print(f"Hashed: {hashed}")
print(f"Verify correct: {verify_password('my_secure_password', hashed)}")
print(f"Verify wrong: {verify_password('wrong_password', hashed)}")

Output:

Hashed: $2b$12$LJ3m4ys3Lg...Kz8dHJKe (unique each time)
Verify correct: True
Verify wrong: False

The CryptContext handles algorithm selection, salt generation, and hash verification. The deprecated="auto" setting means passlib will automatically upgrade old hashes to the current scheme when passwords are verified. Each hash is unique even for the same password because bcrypt generates a random salt internally.

Creating and Decoding JWT Tokens

The python-jose library creates and verifies JWT tokens. Each token contains a payload (claims) signed with your secret key.

# jwt_utils.py
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET_KEY = "your-secret-key-keep-this-safe-and-long"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """Create a JWT access token with an expiration time."""
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_access_token(token: str) -> dict:
    """Decode and verify a JWT token. Raises JWTError if invalid."""
    return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

# Example
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")

payload = decode_access_token(token)
print(f"Payload: {payload}")

Output:

Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzd...
Payload: {'sub': 'alice', 'role': 'admin', 'exp': 1712534400}

The sub (subject) claim is a JWT standard for identifying the user. The exp claim sets when the token expires — after this time, decode_access_token raises a JWTError automatically. In production, store SECRET_KEY in an environment variable, not in your source code.

Building the Complete Authentication System

Now let us combine password hashing and JWT tokens into a complete FastAPI authentication system with a user database, login endpoint, and protected routes.

# auth_app.py
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel

# Configuration
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app = FastAPI(title="Auth Demo")

# Simulated user database
fake_users_db = {
    "alice": {
        "username": "alice",
        "hashed_password": pwd_context.hash("alice123"),
        "role": "admin",
    },
    "bob": {
        "username": "bob",
        "hashed_password": pwd_context.hash("bob456"),
        "role": "user",
    },
}

# Pydantic models
class Token(BaseModel):
    access_token: str
    token_type: str

class User(BaseModel):
    username: str
    role: str

# Helper functions
def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or not pwd_context.verify(password, user["hashed_password"]):
        return None
    return user

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

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

# Endpoints
@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(data={"sub": user["username"], "role": user["role"]})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me", response_model=User)
async def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user

@app.get("/admin")
async def admin_only(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return {"message": f"Welcome admin {current_user.username}!"}

@app.get("/public")
async def public_endpoint():
    return {"message": "This endpoint is open to everyone"}

Testing the login flow:

# Step 1: Login to get a token
curl -X POST http://127.0.0.1:8000/login \
  -d "username=alice&password=alice123"

# Response:
{"access_token": "eyJhbGciOiJIUzI1NiI...", "token_type": "bearer"}

# Step 2: Access protected endpoint with the token
curl http://127.0.0.1:8000/me \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiI..."

# Response:
{"username": "alice", "role": "admin"}

# Step 3: Access admin-only endpoint
curl http://127.0.0.1:8000/admin \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiI..."

# Response:
{"message": "Welcome admin alice!"}

The get_current_user dependency is the heart of this system. It extracts the token from the header, decodes it, looks up the user, and returns a User object — all before your endpoint function runs. Any endpoint that includes current_user: User = Depends(get_current_user) is automatically protected.

Role-Based Access Control

The admin endpoint above uses a simple role check, but you can make this reusable with a dependency factory that creates role-checking dependencies on the fly.

# role_checker.py
from fastapi import Depends, HTTPException, status

def require_role(required_role: str):
    """Create a dependency that checks if the user has the required role."""
    async def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Role '{required_role}' required. You have '{current_user.role}'.",
            )
        return current_user
    return role_checker

# Usage in endpoints
@app.get("/admin/dashboard")
async def admin_dashboard(admin: User = Depends(require_role("admin"))):
    return {"message": "Admin dashboard", "user": admin.username}

@app.get("/editor/publish")
async def editor_publish(editor: User = Depends(require_role("editor"))):
    return {"message": "Editor publish page", "user": editor.username}

Output (bob tries /admin/dashboard):

{"detail": "Role 'admin' required. You have 'user'."}

The require_role function returns a new dependency for each role. This pattern scales cleanly — add as many roles as your application needs without duplicating validation logic in every endpoint.

Token Refresh Strategy

Access tokens should be short-lived (15-30 minutes) for security. But you do not want users logging in every 30 minutes. The standard solution is a refresh token with a longer lifespan that can generate new access tokens.

# token_refresh.py
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire, "type": "refresh"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/login", response_model=dict)
async def login_with_refresh(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    access_token = create_access_token({"sub": user["username"], "role": user["role"]})
    refresh_token = create_refresh_token({"sub": user["username"]})
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }

@app.post("/refresh", response_model=Token)
async def refresh_access_token(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=401, detail="Invalid token type")
        username = payload.get("sub")
        new_token = create_access_token({"sub": username, "role": fake_users_db[username]["role"]})
        return {"access_token": new_token, "token_type": "bearer"}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

The refresh token has a type claim set to "refresh" so it cannot be used as an access token. When the access token expires, the client sends the refresh token to /refresh to get a new access token without re-entering credentials.

Security Best Practices

Authentication code is one place where shortcuts cause real damage. Here are the practices that matter most for a production FastAPI application.

Real-Life Example: Protected Notes API

Let us build a practical application: a notes API where each user can only see and modify their own notes. This demonstrates how authentication integrates with real CRUD operations.

# notes_api.py
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel
from typing import Optional

SECRET_KEY = "notes-app-secret-change-me"
ALGORITHM = "HS256"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app = FastAPI(title="Protected Notes API")

# Databases
users_db = {
    "alice": {"username": "alice", "hashed_password": pwd_context.hash("alice123")},
    "bob": {"username": "bob", "hashed_password": pwd_context.hash("bob456")},
}
notes_db = {}
next_note_id = 1

# Models
class Token(BaseModel):
    access_token: str
    token_type: str

class NoteCreate(BaseModel):
    title: str
    content: str

class NoteResponse(BaseModel):
    id: int
    title: str
    content: str
    owner: str
    created_at: str

# Auth helpers
def create_token(username: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=30)
    return jwt.encode({"sub": username, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None or username not in users_db:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Endpoints
@app.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users_db.get(form_data.username)
    if not user or not pwd_context.verify(form_data.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return {"access_token": create_token(form_data.username), "token_type": "bearer"}

@app.post("/notes", response_model=NoteResponse, status_code=201)
async def create_note(note: NoteCreate, username: str = Depends(get_current_user)):
    global next_note_id
    note_data = {
        "id": next_note_id,
        "title": note.title,
        "content": note.content,
        "owner": username,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    notes_db[next_note_id] = note_data
    next_note_id += 1
    return note_data

@app.get("/notes", response_model=list[NoteResponse])
async def list_my_notes(username: str = Depends(get_current_user)):
    return [n for n in notes_db.values() if n["owner"] == username]

@app.delete("/notes/{note_id}", status_code=204)
async def delete_note(note_id: int, username: str = Depends(get_current_user)):
    note = notes_db.get(note_id)
    if not note:
        raise HTTPException(status_code=404, detail="Note not found")
    if note["owner"] != username:
        raise HTTPException(status_code=403, detail="Not your note")
    del notes_db[note_id]

Testing the flow:

# Login as Alice
curl -X POST http://127.0.0.1:8000/login -d "username=alice&password=alice123"
# {"access_token": "eyJ...", "token_type": "bearer"}

# Create a note (use Alice's token)
curl -X POST http://127.0.0.1:8000/notes \
  -H "Authorization: Bearer eyJ..." \
  -H "Content-Type: application/json" \
  -d '{"title": "Shopping List", "content": "Milk, eggs, bread"}'
# {"id": 1, "title": "Shopping List", "content": "Milk, eggs, bread", "owner": "alice", ...}

# Bob cannot see Alice's notes
curl http://127.0.0.1:8000/notes -H "Authorization: Bearer BOB_TOKEN"
# [] (empty list -- Bob has no notes)

Each note is tagged with its owner, and the list_my_notes endpoint filters by the authenticated user. The delete endpoint checks ownership before allowing deletion. This pattern of tying data to the authenticated user is fundamental to building multi-tenant APIs.

Frequently Asked Questions

Should I use sessions or JWT for my API?

Use JWT for stateless APIs (especially mobile or SPA clients) where you do not want to maintain server-side session storage. Use sessions for traditional server-rendered web applications where the backend controls the full page lifecycle. JWT works well for microservices because any service can verify the token independently without a shared session store.

How should I generate and store the SECRET_KEY?

Generate a random key with openssl rand -hex 32 in your terminal. Store it in an environment variable and read it with os.environ["SECRET_KEY"]. Never commit it to version control. In production, use a secrets manager like AWS Secrets Manager, HashiCorp Vault, or your platform’s built-in secrets.

How do I implement password reset?

Create a /forgot-password endpoint that generates a short-lived token (10-15 minutes) and sends it to the user’s email. Create a /reset-password endpoint that accepts the token and the new password. Use the same JWT mechanism but with a different token type claim so reset tokens cannot be used for API access.

Can I add Google or GitHub login?

Yes. Use the authlib or httpx-oauth library to implement OAuth2 authorization code flow. FastAPI’s dependency injection makes it straightforward to add multiple authentication methods. The user logs in through the provider’s OAuth flow, and your server exchanges the authorization code for user profile data.

How do I test authenticated endpoints?

Use FastAPI’s TestClient with the headers parameter. Create a test user, generate a token in your test setup, and include it in requests: client.get("/me", headers={"Authorization": f"Bearer {token}"}). For dependency overrides, use app.dependency_overrides[get_current_user] = lambda: test_user.

Conclusion

FastAPI’s dependency injection system makes OAuth2 and JWT authentication clean and reusable. The get_current_user dependency is the single point of authentication that you inject into any endpoint that needs protection. Combined with passlib for secure password hashing and python-jose for token management, you have a production-ready auth system in under 100 lines of code.

Start with the Protected Notes API example and extend it with a real database, email verification, and OAuth2 social login. The official FastAPI security documentation at fastapi.tiangolo.com/tutorial/security covers advanced patterns including scopes and multiple authentication schemes.

Related Articles

• How To Use Python SQLModel for Database Models with FastAPI
• FastAPI vs Flask: Which Python Framework Should You Choose?
• How To Build a REST API with FastAPI in Python
• How To Use Python Requests To Call REST APIs With Authentication
• How To Use Python litellm for Multi-Model LLM Integration
• How To Use Python instructor for Structured LLM Outputs
• Getting Historical Stock Data Using Python 3

Intermediate

You have a new Python project that needs a web API. You open Google, type “best Python web framework,” and immediately drown in opinions. Flask has been the go-to choice for over a decade. FastAPI showed up in 2018 and climbed to 75,000+ GitHub stars faster than almost any Python project in history. Both can build APIs. Both are lightweight. So which one should you actually pick for your next project?

The answer depends on what you are building. Flask gives you maximum flexibility and a massive ecosystem of extensions. FastAPI gives you automatic data validation, async support out of the box, and self-documenting endpoints. Neither is universally better — they solve different problems in different ways, and understanding those differences saves you from rewriting code six months later.

In this article, we compare Flask and FastAPI across every dimension that matters: setup and routing, data validation, async support, performance, automatic documentation, ecosystem maturity, and real-world project structure. Every comparison includes runnable code so you can see the differences yourself. By the end, you will have a clear decision framework for choosing the right tool.

Written by Pubs

Python developer and educator with 15+ years building production systems across data engineering, web APIs, and AI tooling. Founder of Python How To Program — 270+ in-depth tutorials covering the modern Python stack.

View all tutorials by Pubs →

FastAPI vs Flask: Quick Comparison

Part of the Python Web Frameworks Hub. See the full hub for related Python tutorials.

Part of the Modern Python AI Stack series. See the full tutorial hub for all 23 tutorials on LangGraph, MCP, Pydantic AI, Polars, FastAPI, Litestar, Typer, and more.

Before we dig into code, here is a high-level comparison table covering the key differences between Flask and FastAPI.

Now let us see how these differences play out in actual code.

Hello World: Setup and First Route

The fastest way to feel the difference between Flask and FastAPI is to build the simplest possible endpoint in each. Both frameworks let you go from zero to running server in under 10 lines.

Flask Hello World

Flask uses the @app.route decorator and returns plain strings or dictionaries. Install it with pip install flask and run with the built-in development server.

# flask_hello.py
from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/hello")
def hello():
    return jsonify({"message": "Hello from Flask!"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

Output (when you visit http://localhost:5000/hello):

{"message": "Hello from Flask!"}

FastAPI Hello World

FastAPI uses the same decorator pattern but returns dictionaries directly — no need for jsonify. Install it with pip install fastapi uvicorn and run with Uvicorn.

# fastapi_hello.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/hello")
async def hello():
    return {"message": "Hello from FastAPI!"}

# Run with: uvicorn fastapi_hello:app --reload --port 8000

Output (when you visit http://localhost:8000/hello):

{"message": "Hello from FastAPI!"}

The syntax is nearly identical. The key differences: FastAPI uses HTTP method decorators (@app.get instead of @app.route), supports async def natively, and returns dicts without a wrapper function. Flask requires jsonify() to return proper JSON responses.

Data Validation: Where FastAPI Pulls Ahead

Data validation is where the two frameworks diverge most sharply. Flask leaves validation entirely up to you. FastAPI makes it automatic through Python type hints and Pydantic models. This single difference changes how much boilerplate you write for every endpoint.

Flask: Manual Validation

In Flask, you parse the request body yourself, check each field manually, and return error responses when something is wrong. Here is a typical pattern for creating a user.

# flask_validation.py
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()

    # Manual validation -- you write every check
    if not data:
        return jsonify({"error": "Request body required"}), 400
    if "name" not in data or not isinstance(data["name"], str):
        return jsonify({"error": "name must be a string"}), 400
    if "email" not in data or "@" not in data["email"]:
        return jsonify({"error": "Valid email required"}), 400
    if "age" in data and not isinstance(data["age"], int):
        return jsonify({"error": "age must be an integer"}), 400

    return jsonify({"status": "created", "user": data}), 201

if __name__ == "__main__":
    app.run(debug=True, port=5000)

Output (POST with valid data):

{"status": "created", "user": {"name": "Alice", "email": "alice@example.com", "age": 30}}

FastAPI: Pydantic Does the Work

FastAPI validates request data automatically using Pydantic models. You define a model with type hints, and FastAPI rejects invalid requests before your function even runs.

# fastapi_validation.py
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class User(BaseModel):
    name: str
    email: EmailStr
    age: int | None = None

@app.post("/users", status_code=201)
async def create_user(user: User):
    return {"status": "created", "user": user.model_dump()}

# Run with: uvicorn fastapi_validation:app --reload --port 8000

Output (POST with invalid email):

{
  "detail": [
    {
      "type": "value_error",
      "loc": ["body", "email"],
      "msg": "value is not a valid email address"
    }
  ]
}

The FastAPI version is half the code and catches more errors. Pydantic validates types, checks email format, handles optional fields with defaults, and returns structured error responses automatically. In Flask, you would need to add a library like Marshmallow or Cerberus to get similar functionality, and even then it requires more setup.

Async Support: Native vs Retrofitted

Modern APIs often need to handle many concurrent connections — waiting on database queries, calling external services, or streaming responses. Async programming lets your server handle thousands of these waiting operations without blocking. This is where the architectural difference between Flask and FastAPI matters most.

FastAPI was built on ASGI (Asynchronous Server Gateway Interface) from the start. Every route handler can be an async def function, and the framework coordinates concurrency through Python’s asyncio event loop. Flask was built on WSGI (Web Server Gateway Interface), which is synchronous by design. Flask 2.0 added async def support, but it runs each async view in a separate thread rather than using a true event loop.

FastAPI Async Example

# fastapi_async.py
import asyncio
from fastapi import FastAPI

app = FastAPI()

async def fetch_from_database():
    """Simulate a slow database query."""
    await asyncio.sleep(1)
    return {"users": ["Alice", "Bob", "Charlie"]}

async def fetch_from_cache():
    """Simulate a cache lookup."""
    await asyncio.sleep(0.5)
    return {"cached": True}

@app.get("/dashboard")
async def dashboard():
    # Both calls run concurrently -- total time ~1 second, not 1.5
    db_task = asyncio.create_task(fetch_from_database())
    cache_task = asyncio.create_task(fetch_from_cache())
    db_result = await db_task
    cache_result = await cache_task
    return {**db_result, **cache_result}

# Run with: uvicorn fastapi_async:app --reload --port 8000

Output:

{"users": ["Alice", "Bob", "Charlie"], "cached": true}

The two async calls run concurrently using asyncio.create_task(), so the total response time is about 1 second instead of 1.5 seconds. This pattern scales beautifully when your API calls multiple microservices or databases per request.

Flask Async (Limited)

# flask_async.py
import asyncio
from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/dashboard")
async def dashboard():
    # Flask 2.0+ supports async views, but runs them in threads
    await asyncio.sleep(1)
    return jsonify({"users": ["Alice", "Bob", "Charlie"], "note": "async works but limited"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

Flask’s async support works for simple cases, but it does not give you the same concurrency benefits as FastAPI’s native ASGI approach. For I/O-heavy workloads with many concurrent connections, FastAPI’s architecture has a clear advantage.

Automatic API Documentation

One of FastAPI’s most impressive features is automatic interactive documentation. The moment you define an endpoint with type hints, FastAPI generates Swagger UI and ReDoc pages at /docs and /redoc respectively. No configuration needed.

Flask has no built-in documentation generator. You can add it with extensions like Flask-RESTX (which includes Swagger) or Flasgger, but they require additional decorators and configuration. Here is what you need in Flask to get what FastAPI gives you for free.

This is a significant productivity win for teams building APIs. Frontend developers, QA testers, and external consumers can explore your API interactively without reading source code or maintaining a separate Postman collection.

Project Structure and Scalability

Both frameworks support clean project organization through modular patterns, but they use different terminology. Flask uses Blueprints to split a large application into reusable modules. FastAPI uses APIRouter for the same purpose. The concepts are nearly identical.

Flask Blueprints

# flask_blueprint.py
from flask import Flask, Blueprint, jsonify

# Define a blueprint for user routes
users_bp = Blueprint("users", __name__, url_prefix="/users")

@users_bp.route("/")
def list_users():
    return jsonify({"users": ["Alice", "Bob"]})

@users_bp.route("/<int:user_id>")
def get_user(user_id):
    return jsonify({"user_id": user_id, "name": "Alice"})

# Main app registers the blueprint
app = Flask(__name__)
app.register_blueprint(users_bp)

if __name__ == "__main__":
    app.run(debug=True, port=5000)

Output (GET /users/):

{"users": ["Alice", "Bob"]}

FastAPI APIRouter

# fastapi_router.py
from fastapi import FastAPI, APIRouter

# Define a router for user routes
users_router = APIRouter(prefix="/users", tags=["users"])

@users_router.get("/")
async def list_users():
    return {"users": ["Alice", "Bob"]}

@users_router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "Alice"}

# Main app includes the router
app = FastAPI()
app.include_router(users_router)

# Run with: uvicorn fastapi_router:app --reload --port 8000

Output (GET /users/):

{"users": ["Alice", "Bob"]}

The patterns are almost identical. The main difference is that FastAPI’s router automatically includes the routes in the generated documentation under the specified tag, while Flask Blueprints need additional configuration for documentation.

Ecosystem and Community

Flask has been around since 2010, which gives it a massive head start in ecosystem size. If you need a feature, chances are someone built a Flask extension for it: Flask-Login for authentication, Flask-SQLAlchemy for ORM integration, Flask-Mail for email, Flask-CORS for cross-origin requests, Flask-Migrate for database migrations, and hundreds more.

FastAPI’s ecosystem is smaller but growing rapidly. It leans on the broader Python ecosystem rather than framework-specific extensions. For databases, you use SQLAlchemy or Tortoise-ORM directly. For authentication, you use python-jose and passlib. For CORS, FastAPI has built-in middleware. This approach means fewer “FastAPI-specific” packages but more flexibility in choosing your tools.

When to Use Each Framework

After comparing code, features, and ecosystems, here is a practical decision framework. Neither framework is objectively better — the right choice depends on what you are building and who is building it.

Choose Flask When

Flask is the better choice when you are building server-rendered web applications with HTML templates (Jinja2 is deeply integrated), when your team is new to web development and benefits from Flask’s minimal learning curve, when you need a specific Flask extension that has no equivalent in FastAPI’s ecosystem, when you are prototyping quickly and do not need type-enforced validation, or when you are maintaining an existing Flask codebase and migration is not justified.

Choose FastAPI When

FastAPI is the better choice when you are building a pure REST API or microservice (no server-rendered HTML), when you need automatic request/response validation and do not want to write it yourself, when your API handles many concurrent I/O operations (database calls, external API requests), when you want auto-generated interactive documentation for your team or API consumers, or when you are starting a new project and your team is comfortable with Python type hints.

Real-Life Example: Todo API in Both Frameworks

To make the comparison concrete, here is the same Todo API built in both frameworks. This gives you a side-by-side view of how the same requirements translate into code.

Flask Version

# flask_todo.py
from flask import Flask, request, jsonify

app = Flask(__name__)
todos = []
next_id = 1

@app.route("/todos", methods=["GET"])
def list_todos():
    return jsonify(todos)

@app.route("/todos", methods=["POST"])
def create_todo():
    global next_id
    data = request.get_json()
    if not data or "title" not in data:
        return jsonify({"error": "title is required"}), 400
    todo = {
        "id": next_id,
        "title": data["title"],
        "done": data.get("done", False)
    }
    next_id += 1
    todos.append(todo)
    return jsonify(todo), 201

@app.route("/todos/<int:todo_id>", methods=["PUT"])
def update_todo(todo_id):
    data = request.get_json()
    for todo in todos:
        if todo["id"] == todo_id:
            todo["title"] = data.get("title", todo["title"])
            todo["done"] = data.get("done", todo["done"])
            return jsonify(todo)
    return jsonify({"error": "Not found"}), 404

@app.route("/todos/<int:todo_id>", methods=["DELETE"])
def delete_todo(todo_id):
    global todos
    todos = [t for t in todos if t["id"] != todo_id]
    return jsonify({"status": "deleted"})

if __name__ == "__main__":
    app.run(debug=True, port=5000)

FastAPI Version

# fastapi_todo.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
todos = []
next_id = 1

class TodoCreate(BaseModel):
    title: str
    done: bool = False

class TodoResponse(BaseModel):
    id: int
    title: str
    done: bool

@app.get("/todos", response_model=list[TodoResponse])
async def list_todos():
    return todos

@app.post("/todos", response_model=TodoResponse, status_code=201)
async def create_todo(todo: TodoCreate):
    global next_id
    new_todo = {"id": next_id, "title": todo.title, "done": todo.done}
    next_id += 1
    todos.append(new_todo)
    return new_todo

@app.put("/todos/{todo_id}", response_model=TodoResponse)
async def update_todo(todo_id: int, todo: TodoCreate):
    for t in todos:
        if t["id"] == todo_id:
            t["title"] = todo.title
            t["done"] = todo.done
            return t
    raise HTTPException(status_code=404, detail="Not found")

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    global todos
    todos = [t for t in todos if t["id"] != todo_id]
    return {"status": "deleted"}

# Run with: uvicorn fastapi_todo:app --reload --port 8000

Both versions do the same thing, but the FastAPI version gives you automatic validation on every POST and PUT request, typed response models that document exactly what each endpoint returns, and interactive Swagger docs at /docs — all without a single extra line of configuration. The Flask version requires you to validate manually and document separately.

Frequently Asked Questions

Is Flask dead now that FastAPI exists?

Not at all. Flask remains one of the most popular Python web frameworks with active development, regular releases, and a massive ecosystem. Flask 3.0 introduced further improvements and the framework continues to evolve. Many production applications run Flask successfully, and it remains an excellent choice for server-rendered web applications, prototypes, and projects that benefit from its extensive extension library.

Can I migrate from Flask to FastAPI incrementally?

Yes, but it is not a simple find-and-replace. The routing syntax is similar, but data validation patterns, middleware, and extension usage differ significantly. The most practical approach is to build new endpoints in FastAPI while keeping existing Flask endpoints running, then migrate route by route. Libraries like a2wsgi can help run both frameworks during the transition period.

Is FastAPI really faster than Flask?

In benchmarks, FastAPI running on Uvicorn typically handles 2-3x more requests per second than Flask running on Gunicorn for async workloads. For synchronous, CPU-bound tasks, the difference is smaller. The real performance gain comes from FastAPI’s native async support, which lets a single worker handle many concurrent I/O-bound requests without blocking.

Which should I learn first as a beginner?

Flask is often recommended as a first framework because it has fewer concepts to learn upfront. You can build a working web app without understanding type hints, Pydantic models, or async/await. Once you are comfortable with Flask and HTTP concepts, learning FastAPI becomes straightforward because you already understand routing, request handling, and middleware.

What about Django? How does it compare?

Django is a “batteries-included” full-stack framework with an ORM, admin panel, authentication system, and template engine built in. Flask and FastAPI are both “micro” frameworks that let you choose your own components. If you need a full web application with user management, an admin dashboard, and server-rendered pages, Django is worth considering. For pure REST APIs and microservices, Flask or FastAPI are typically more appropriate.

Conclusion

Flask and FastAPI are both excellent Python web frameworks that serve different needs. Flask gives you simplicity, flexibility, and the largest extension ecosystem in the Python web world. FastAPI gives you automatic validation, native async support, and self-documenting APIs with zero extra configuration. The code examples in this article show that the syntax is similar enough to switch between them comfortably.

For your next project, start by asking: “Am I building a REST API or a full web application?” If you are building a pure API with typed data flowing in and out, FastAPI will save you hours of boilerplate validation and documentation. If you are building a traditional web app with HTML templates, or you need a specific Flask extension, Flask remains the proven choice.

You can explore the official documentation for both frameworks to go deeper: Flask documentation and FastAPI documentation.

Related Articles

• How To Use Python SQLModel for Database Models with FastAPI
• How To Build a Flask Web Application in Python
• How To Add Authentication to FastAPI with OAuth2 and JWT
• How To Build a REST API with FastAPI in Python
• How To Create a Simple REST API With Python Flask in 2026
• How To Use Python litellm for Multi-Model LLM Integration
• How To Detect If A Date Is A Public Holiday In Python 3