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.

WebSocket Echo Server: Quick Example

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 httpx for Modern HTTP Requests
• How To Handle Rate Limiting in Python API Calls
• How To Deploy a Python App to AWS Lambda

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.

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 Hash Passwords Safely in Python with bcrypt
• How To Use Python httpx for Modern HTTP Requests
• Python Exception Handling Best Practices

Skill Level: Intermediate

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 Asynchronous Programming in Python with asyncio
• Understanding HTTP Status Codes: A Complete Guide
• Building REST APIs with FastAPI and Python
• Web Scraping with Python: Best Practices and Tools
• Error Handling and Logging in Production Python Applications

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

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 Handle Asynchronous Programming in Python with asyncio
• Understanding HTTP Status Codes: A Complete Guide
• Building REST APIs with FastAPI and Python
• Web Scraping with Python: Best Practices and Tools
• Error Handling and Logging in Production Python Applications

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

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 Implement Secure User Authentication with Flask and SQLAlchemy
• Complete Guide to Password Reset Tokens in Python Web Applications
• Implementing Multi-Factor Authentication (MFA) in Python Django
• SQL Injection Prevention: Parameterized Queries and ORM Patterns
• Rate Limiting Login Attempts to Prevent Brute Force Attacks

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