Intermediate

Any code that calls an external API creates a testing headache. Running real HTTP requests in tests means your tests depend on network availability, API rate limits, third-party uptime, and credentials — factors entirely outside your control. Tests that pass on your laptop fail in CI. Tests that run in 200ms suddenly take 30 seconds waiting for a slow endpoint. Error handling code is nearly impossible to test because you cannot force a real API to return a 500 error on demand. The result: developers skip testing HTTP-calling code, and bugs hide there until production finds them.

The responses library intercepts calls to the requests library before they hit the network and returns pre-configured fake responses instead. Your production code calls requests.get() normally; during tests, responses catches that call and returns whatever JSON, status code, or error you specify. The real network is never touched. Install it with pip install responses. It requires only requests as a dependency, which you already have if you are making HTTP calls.

This tutorial covers the full responses workflow: registering mock responses as a decorator and context manager, testing error conditions and network failures, using callback functions for dynamic responses, verifying that your code makes the correct API calls, and integrating with pytest. By the end, you will be able to test every HTTP code path — success, error, timeout, redirect — with fast, deterministic tests.

Written by Pubs

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

View all tutorials by Pubs →

responses Quick Example

Part of the Python Testing & Quality Hub. See the full hub for related Python tutorials.

The pattern is simple: activate responses, register a URL with a fake response, then run your code. Any call to that URL via requests returns your fake data instead of hitting the network.

# test_responses_quick.py
import responses
import requests

def get_user(user_id: int) -> dict:
    """Fetch a user from a REST API."""
    url = f"https://jsonplaceholder.typicode.com/users/{user_id}"
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.json()

@responses.activate
def test_get_user_success():
    # Register the fake response BEFORE calling your code
    responses.add(
        responses.GET,
        "https://jsonplaceholder.typicode.com/users/1",
        json={"id": 1, "name": "Alice Smith", "email": "alice@example.com"},
        status=200
    )
    user = get_user(1)
    assert user["name"] == "Alice Smith"
    assert user["email"] == "alice@example.com"
    print(f"Fetched user: {user}")
    print(f"API calls made: {len(responses.calls)}")

test_get_user_success()

Output:

Fetched user: {'id': 1, 'name': 'Alice Smith', 'email': 'alice@example.com'}
API calls made: 1

The @responses.activate decorator intercepts all requests calls within that function. Without a registered response for a URL, responses raises a ConnectionError — this is intentional. It prevents tests from accidentally hitting real APIs if a URL is unregistered, which would cause flaky tests and unexpected API usage.

What Is the responses Library?

The responses library works by patching the underlying transport layer of the requests library. When activated, it replaces the real HTTP adapter with a mock adapter that checks whether an incoming request URL matches any registered mock. If it matches, the mock response is returned. If not, a ConnectionError is raised (by default).

The library supports all HTTP methods (GET, POST, PUT, PATCH, DELETE), regex URL matching, streaming responses, and even simulating connection errors and timeouts. It is compatible with the requests.Session object and all requests-based libraries like httpx equivalents built on requests.

Registering GET and POST Mocks

The most common pattern: register responses for GET and POST endpoints, then test the code paths that call them. Here is a realistic example testing a service that creates and fetches resources:

# test_responses_crud.py
import responses
import requests
import json

BASE_URL = "https://jsonplaceholder.typicode.com"

def create_post(title: str, body: str, user_id: int) -> dict:
    resp = requests.post(
        f"{BASE_URL}/posts",
        json={"title": title, "body": body, "userId": user_id},
        headers={"Content-Type": "application/json"},
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

def get_posts_by_user(user_id: int) -> list:
    resp = requests.get(f"{BASE_URL}/posts?userId={user_id}", timeout=10)
    resp.raise_for_status()
    return resp.json()

@responses.activate
def test_create_and_fetch():
    # Mock the POST endpoint
    responses.add(
        responses.POST,
        f"{BASE_URL}/posts",
        json={"id": 101, "title": "Python Tips", "body": "Use f-strings!", "userId": 5},
        status=201
    )
    # Mock the GET endpoint (with query params as part of URL)
    responses.add(
        responses.GET,
        f"{BASE_URL}/posts",
        json=[
            {"id": 101, "title": "Python Tips", "userId": 5},
            {"id": 102, "title": "Async Python", "userId": 5}
        ],
        status=200
    )

    new_post = create_post("Python Tips", "Use f-strings!", user_id=5)
    assert new_post["id"] == 101
    assert new_post["title"] == "Python Tips"

    user_posts = get_posts_by_user(5)
    assert len(user_posts) == 2
    assert user_posts[0]["userId"] == 5

    # Verify request bodies were sent correctly
    post_call = responses.calls[0]
    sent_body = json.loads(post_call.request.body)
    assert sent_body["userId"] == 5
    print(f"POST body sent: {sent_body}")
    print(f"GET returned {len(user_posts)} posts")
    print(f"Total API calls made: {len(responses.calls)}")

test_create_and_fetch()

Output:

POST body sent: {'title': 'Python Tips', 'body': 'Use f-strings!', 'userId': 5}
GET returned 2 posts
Total API calls made: 2

The responses.calls list is a powerful testing tool — it records every request made during the test, including URL, method, headers, and body. You can assert on these to verify your code sends the right data to the API, not just that it handles the response correctly.

Testing Error Conditions

Error handling is the most undertested part of any HTTP-calling codebase. Without responses, you cannot force a real API to return a 500 error or refuse to connect. With responses, every error scenario becomes straightforward to test.

# test_responses_errors.py
import responses
import requests
from requests.exceptions import ConnectionError, Timeout

def safe_fetch(url: str) -> dict:
    """Fetch with error handling -- returns None on failure."""
    try:
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.HTTPError as e:
        print(f"HTTP error: {e.response.status_code}")
        return None
    except (ConnectionError, Timeout) as e:
        print(f"Connection error: {type(e).__name__}")
        return None

BASE = "https://api.example-test.com"

@responses.activate
def test_404_returns_none():
    responses.add(responses.GET, f"{BASE}/resource/999", status=404)
    result = safe_fetch(f"{BASE}/resource/999")
    assert result is None
    print("404 test: returned None correctly")

@responses.activate
def test_500_returns_none():
    responses.add(responses.GET, f"{BASE}/resource/1", status=500,
                  json={"error": "Internal Server Error"})
    result = safe_fetch(f"{BASE}/resource/1")
    assert result is None
    print("500 test: returned None correctly")

@responses.activate
def test_connection_error():
    responses.add(responses.GET, f"{BASE}/resource/2",
                  body=ConnectionError("Network unreachable"))
    result = safe_fetch(f"{BASE}/resource/2")
    assert result is None
    print("ConnectionError test: handled gracefully")

@responses.activate
def test_timeout():
    responses.add(responses.GET, f"{BASE}/resource/3",
                  body=Timeout("Request timed out"))
    result = safe_fetch(f"{BASE}/resource/3")
    assert result is None
    print("Timeout test: handled gracefully")

test_404_returns_none()
test_500_returns_none()
test_connection_error()
test_timeout()

Output:

HTTP error: 404
404 test: returned None correctly
HTTP error: 500
500 test: returned None correctly
Connection error: ConnectionError
ConnectionError test: handled gracefully
Connection error: Timeout
Timeout test: handled gracefully

Passing an exception as body= to responses.add() causes the mock to raise that exception instead of returning a response. This directly tests your except blocks — the code paths that are most often skipped in testing because they are hard to trigger against a real API.

Dynamic Responses with add_callback

Sometimes a static mock response is not enough — you need the response to depend on what the request contains. add_callback lets you provide a function that receives the request object and returns the response dynamically.

# test_responses_callback.py
import responses
import requests
import json

def handle_user_request(request):
    """Return different users based on the URL path."""
    user_id = int(request.url.split("/")[-1])
    if user_id > 100:
        return (404, {}, json.dumps({"error": "User not found"}))
    return (200, {"Content-Type": "application/json"},
            json.dumps({"id": user_id, "name": f"User {user_id}", "active": True}))

def handle_post_request(request):
    """Echo back the posted data with a generated ID."""
    body = json.loads(request.body)
    body["id"] = 42
    body["created"] = "2025-05-04"
    return (201, {"Content-Type": "application/json"}, json.dumps(body))

@responses.activate
def test_dynamic_responses():
    responses.add_callback(
        responses.GET,
        "https://api.example-test.com/users/",
        callback=handle_user_request,
        match_querystring=False
    )
    responses.add_callback(
        responses.POST,
        "https://api.example-test.com/posts",
        callback=handle_post_request
    )

    # Valid user
    r = requests.get("https://api.example-test.com/users/7")
    assert r.status_code == 200
    assert r.json()["name"] == "User 7"
    print(f"User 7: {r.json()}")

    # Invalid user
    r = requests.get("https://api.example-test.com/users/999")
    assert r.status_code == 404
    print(f"User 999: {r.json()}")

    # Post with echoed body
    r = requests.post("https://api.example-test.com/posts",
                      json={"title": "Test", "author": "Alice"})
    assert r.status_code == 201
    assert r.json()["id"] == 42
    print(f"Created post: {r.json()}")

test_dynamic_responses()

Output:

User 7: {'id': 7, 'name': 'User 7', 'active': True}
User 999: {'error': 'User not found'}
Created post: {'title': 'Test', 'author': 'Alice', 'id': 42, 'created': '2025-05-04'}

Callbacks unlock complex scenarios: responses that change after N calls, responses that validate request authentication headers and return 401 if the token is wrong, or responses that simulate paginated APIs where each call returns the next page. The callback receives the full PreparedRequest object, so you have access to headers, body, URL, and query parameters.

Real-Life Example: Testing a GitHub API Client

# test_github_client.py
import responses
import requests

class GitHubClient:
    BASE_URL = "https://api.github.com"

    def __init__(self, token: str):
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"token {token}",
            "Accept": "application/vnd.github.v3+json"
        })

    def get_repo(self, owner: str, repo: str) -> dict:
        resp = self.session.get(f"{self.BASE_URL}/repos/{owner}/{repo}")
        resp.raise_for_status()
        return resp.json()

    def list_issues(self, owner: str, repo: str, state: str = "open") -> list:
        resp = self.session.get(
            f"{self.BASE_URL}/repos/{owner}/{repo}/issues",
            params={"state": state}
        )
        resp.raise_for_status()
        return resp.json()

    def create_issue(self, owner: str, repo: str, title: str, body: str = "") -> dict:
        resp = self.session.post(
            f"{self.BASE_URL}/repos/{owner}/{repo}/issues",
            json={"title": title, "body": body}
        )
        resp.raise_for_status()
        return resp.json()

@responses.activate
def test_github_client_full():
    BASE = "https://api.github.com"
    client = GitHubClient(token="test-token-abc")

    # Mock: get repo
    responses.add(responses.GET, f"{BASE}/repos/alice/my-project",
                  json={"id": 999, "name": "my-project", "stargazers_count": 42, "open_issues": 3},
                  status=200)

    # Mock: list open issues
    responses.add(responses.GET, f"{BASE}/repos/alice/my-project/issues",
                  json=[
                      {"id": 1, "title": "Fix login bug", "state": "open"},
                      {"id": 2, "title": "Add dark mode", "state": "open"},
                  ], status=200)

    # Mock: create issue
    responses.add(responses.POST, f"{BASE}/repos/alice/my-project/issues",
                  json={"id": 101, "title": "Performance regression in v2.1", "number": 101},
                  status=201)

    # Run the client methods
    repo = client.get_repo("alice", "my-project")
    assert repo["stargazers_count"] == 42
    print(f"Repo: {repo['name']} ({repo['stargazers_count']} stars)")

    issues = client.list_issues("alice", "my-project")
    assert len(issues) == 2
    print(f"Open issues: {[i['title'] for i in issues]}")

    new_issue = client.create_issue("alice", "my-project",
                                    "Performance regression in v2.1",
                                    "Noticed in benchmarks...")
    assert new_issue["number"] == 101
    print(f"Created issue #{new_issue['number']}: {new_issue['title']}")

    # Verify Authorization header was sent with every request
    for call in responses.calls:
        assert "Authorization" in call.request.headers
        assert call.request.headers["Authorization"] == "token test-token-abc"
    print(f"All {len(responses.calls)} calls included auth header: OK")

test_github_client_full()

Output:

Repo: my-project (42 stars)
Open issues: ['Fix login bug', 'Add dark mode']
Created issue #101: Performance regression in v2.1
All 3 calls included auth header: OK

The final assertion — checking that every call included the Authorization header — is a pattern worth using in every API client test. It catches a common bug: authentication headers being dropped when the session is configured incorrectly or when a redirect is followed. responses.calls makes this verification trivial.

Frequently Asked Questions

How does responses compare to httpretty and VCR.py?

responses is requests-specific and lightweight — it only patches requests, which is exactly what you want when your codebase uses requests. httpretty works at the socket level and intercepts any HTTP library (including urllib and http.client), making it more powerful but also more invasive and harder to debug. VCR.py records real API responses to YAML cassette files and replays them, which is useful for complex integration tests but adds file management overhead. For most requests-based testing, responses is the right balance of simplicity and control.

What happens if I call requests without activating responses?

Without @responses.activate or the context manager, responses is inactive and all requests calls go to the real network as normal. This is by design — responses never interferes with production code or non-test code. Inside an activated context, any requests call to an unregistered URL raises a ConnectionError with a clear message indicating no matching mock was found.

Does responses work with requests.Session?

Yes — responses patches the underlying transport adapter used by all requests objects, including Session instances. Any Session().get(), Session().post(), or other session-based call is intercepted exactly like a bare requests.get() call. This means you can test code that uses sessions for cookie handling, connection pooling, or persistent headers without any special configuration.

How do I assert a URL was called exactly N times?

Use responses.assert_call_count(url, count) after your test code runs. For example, responses.assert_call_count("https://api.example.com/users", 3) raises AssertionError if the URL was not called exactly 3 times. Alternatively, filter responses.calls manually: len([c for c in responses.calls if c.request.url == url]). This is useful for testing retry logic — confirm your code retries exactly 3 times before giving up.

Can I match URLs with regex patterns?

Yes — pass a compiled regex pattern to responses.add() instead of a string URL. For example: responses.add(responses.GET, re.compile(r"https://api.example.com/users/\d+"), json={...}) matches any numeric user ID. This is useful when the exact URL varies but follows a pattern, like paginated endpoints with varying page numbers or resource IDs generated at runtime.

Conclusion

The responses library makes HTTP-dependent code fully testable by intercepting requests calls before they reach the network. You have covered registering static mocks with responses.add(), simulating errors and network failures, using add_callback for dynamic responses that depend on request content, and inspecting responses.calls to verify request correctness. The GitHub client example showed a complete test suite for a real-world API client that tests success paths, error handling, and authentication — all without a single real network call.

Extend the GitHub client tests by adding retry logic — use a callback that returns 429 on the first two calls and 200 on the third, then verify your retry decorator handles it correctly. For the complete API reference, see the responses documentation on GitHub.

Related Articles

• How To Use Python freezegun for Mocking Time in Tests
• How To Use Python unittest.mock for Mocking Dependencies
• How To Use Python questionary for Interactive CLI Prompts
• How To Use Python memray for Memory Profiling
• How To Use Python beartype for Runtime Type Checking
• How To Use Python nox for Automated Test Sessions
• How To Implement Retry Logic in Python with Tenacity

Intermediate

Testing code that depends on the current date and time is notoriously painful. Functions that call datetime.now() return different values every second, which makes assertions fragile. Code that checks “is this subscription expired?” or “has this token timed out?” cannot be tested reliably without controlling what time the code thinks it is. Most developers resort to awkward monkeypatching, injecting mock datetime objects, or skipping time-sensitive tests altogether — all of which leave production bugs hiding in the gaps.

freezegun solves this problem cleanly. It intercepts calls to Python’s datetime, date, time, and time.time and returns a frozen or controlled time value that you specify. It works as a decorator, a context manager, or a standalone object — and it patches all modules that import datetime at test time, not just the one you are targeting. Install it with pip install freezegun. No other dependencies required.

This tutorial covers everything you need: freezing time with freeze_time as a decorator and context manager, moving time forward in increments with tick, testing expiry logic and time-zone-aware code, and integrating freezegun with pytest fixtures for clean test organization. By the end, you will be able to test any time-dependent code with complete determinism.

Written by Pubs

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

View all tutorials by Pubs →

freezegun Quick Example

Part of the Python Testing & Quality Hub. See the full hub for related Python tutorials.

Here is the core pattern: wrap a test with @freeze_time and every call to datetime.now() inside that test returns the frozen date you specified.

# test_freezegun_quick.py
from freezegun import freeze_time
from datetime import datetime

def get_greeting():
    """Returns a greeting based on the current hour."""
    hour = datetime.now().hour
    if hour < 12:
        return "Good morning"
    elif hour < 18:
        return "Good afternoon"
    else:
        return "Good evening"

@freeze_time("2025-05-15 09:30:00")  # 9:30 AM
def test_morning_greeting():
    assert get_greeting() == "Good morning"
    print(f"Time during test: {datetime.now()}")

@freeze_time("2025-05-15 14:00:00")  # 2:00 PM
def test_afternoon_greeting():
    assert get_greeting() == "Good afternoon"
    print(f"Time during test: {datetime.now()}")

test_morning_greeting()
test_afternoon_greeting()
print(f"Time after tests: {datetime.now().strftime('%H:%M:%S')} (back to real time)")

Output:

Time during test: 2025-05-15 09:30:00
Time during test: 2025-05-15 14:00:00
Time after tests: 14:22:07 (back to real time)

The frozen time is scoped to the decorated function -- as soon as test_morning_greeting returns, datetime.now() reverts to real system time. No cleanup needed. The get_greeting() function is unaware it is being tested; it just calls datetime.now() normally, which is the whole point.

What Is freezegun and How Does It Work?

Freezegun works by patching the datetime module at the import level. When you activate freeze_time, it replaces the real datetime.datetime, datetime.date, datetime.datetime.now, and time.time with fake versions that return your specified time. Crucially, it patches all modules that have already imported datetime -- not just the module where you apply the decorator.

The key advantage of freezegun over unittest.mock.patch is that you do not need to know which module imported datetime. If your service layer imports from datetime import datetime and your model layer imports import datetime, freezegun patches both automatically.

Using freeze_time as a Decorator and Context Manager

You can use freeze_time in three ways: as a class/function decorator, as a context manager inside a test, or as a manually started/stopped object. Each has its place.

# test_freezegun_modes.py
from freezegun import freeze_time
from datetime import datetime, date

# --- MODE 1: Decorator on a function ---
@freeze_time("2025-01-01")
def test_new_year():
    assert datetime.now().year == 2025
    assert date.today() == date(2025, 1, 1)
    print(f"Decorator mode: {datetime.now()}")

test_new_year()

# --- MODE 2: Context manager (freeze only part of a test) ---
def test_partial_freeze():
    real_before = datetime.now().year
    with freeze_time("2030-07-04 12:00:00"):
        frozen = datetime.now()
        assert frozen.year == 2030
        assert frozen.month == 7
    real_after = datetime.now().year
    assert real_before == real_after  # time unfrozen after 'with' block
    print(f"Context manager: frozen={frozen}, before/after real={real_before}")

test_partial_freeze()

# --- MODE 3: Manual start/stop (useful for class-level setUp/tearDown) ---
def test_manual():
    freezer = freeze_time("2024-06-15 08:00:00")
    freezer.start()
    print(f"Manual start: {datetime.now()}")
    freezer.stop()
    print(f"Manual stop: back to real time (year {datetime.now().year})")

test_manual()

Output:

Decorator mode: 2025-01-01 00:00:00
Context manager: frozen=2030-07-04 12:00:00, before/after real=2025
Manual start: 2024-06-15 08:00:00
Manual stop: back to real time (year 2025)

Use the decorator when the entire test function needs frozen time. Use the context manager when you need to test that code behaves differently before and after a specific moment -- for example, that a discount code is valid on day 1 but expired on day 8. Use manual start/stop in unittest.TestCase classes where you need to freeze time in setUp and restore it in tearDown.

Moving Time Forward with tick=True

By default, frozen time does not advance -- every call to datetime.now() returns the same value. Pass tick=True to start time at the frozen point but let it advance at real-world speed. This is useful when testing code that measures elapsed time with datetime.now().

# test_freezegun_tick.py
from freezegun import freeze_time
from datetime import datetime
import time

def measure_operation():
    """Returns elapsed time in seconds for an operation."""
    start = datetime.now()
    time.sleep(0.1)  # simulate 100ms work
    end = datetime.now()
    return (end - start).total_seconds()

# Without tick: start==end because time is frozen
@freeze_time("2025-03-01 10:00:00")
def test_no_tick():
    elapsed = measure_operation()
    print(f"No tick elapsed: {elapsed}s")  # 0.0 -- time never moves

# With tick=True: time advances from the frozen start point
@freeze_time("2025-03-01 10:00:00", tick=True)
def test_with_tick():
    start_now = datetime.now()
    elapsed = measure_operation()
    end_now = datetime.now()
    print(f"Tick elapsed: {elapsed:.3f}s (started at {start_now})")
    print(f"End time: {end_now}")

test_no_tick()
test_with_tick()

Output:

No tick elapsed: 0.0s
Tick elapsed: 0.101s (started at 2025-03-01 10:00:00)
End time: 2025-03-01 10:00:00.101232

With tick=True, time starts from 2025-03-01 10:00:00 and advances at the real wall-clock rate. This is the right mode when you want a deterministic start time but still need elapsed-time measurements to reflect reality -- for example, testing that a timeout fires after 30 seconds without actually waiting 30 seconds (use tick=True and advance time manually with freezer.move_to() instead).

Testing Expiry and Scheduling Logic

The most valuable use of freezegun is testing business logic that depends on dates -- subscription expiry, token timeouts, scheduled jobs that only run on certain days. These tests are impossible to write reliably without time control.

# test_expiry_logic.py
from freezegun import freeze_time
from datetime import datetime, timedelta

class Subscription:
    def __init__(self, start_date: datetime, duration_days: int):
        self.start_date = start_date
        self.expiry_date = start_date + timedelta(days=duration_days)

    def is_active(self) -> bool:
        return datetime.now() < self.expiry_date

    def days_remaining(self) -> int:
        delta = self.expiry_date - datetime.now()
        return max(0, delta.days)

    def renew(self, extra_days: int):
        self.expiry_date += timedelta(days=extra_days)

# --- Tests ---

@freeze_time("2025-06-01 00:00:00")
def test_subscription_active():
    sub = Subscription(datetime(2025, 5, 1), duration_days=90)
    assert sub.is_active() is True
    assert sub.days_remaining() == 59
    print(f"Active test: {sub.days_remaining()} days remaining")

@freeze_time("2025-08-05 00:00:00")  # after 90 days
def test_subscription_expired():
    sub = Subscription(datetime(2025, 5, 1), duration_days=90)
    assert sub.is_active() is False
    assert sub.days_remaining() == 0
    print("Expired test: subscription correctly expired")

@freeze_time("2025-07-25 00:00:00")  # 5 days before expiry
def test_subscription_renewal():
    sub = Subscription(datetime(2025, 5, 1), duration_days=90)
    assert sub.days_remaining() == 4
    sub.renew(30)
    assert sub.days_remaining() == 34
    print(f"Renewal test: {sub.days_remaining()} days after renewal")

test_subscription_active()
test_subscription_expired()
test_subscription_renewal()

Output:

Active test: 59 days remaining
Expired test: subscription correctly expired
Renewal test: 34 days after renewal

Notice that Subscription.__init__ takes an explicit start_date parameter rather than calling datetime.now() internally -- this is good design that makes the class easier to test. The is_active() and days_remaining() methods call datetime.now() to get the current time, which freezegun intercepts. Each test is completely deterministic regardless of when you run it.

Integrating freezegun with pytest Fixtures

For project-wide time control in a pytest suite, define a reusable fixture instead of applying @freeze_time to every test individually.

# conftest.py (put this in your project root)
import pytest
from freezegun import freeze_time

@pytest.fixture
def frozen_time():
    """Freeze time at a known reference point for all tests that need it."""
    freezer = freeze_time("2025-09-01 09:00:00")
    frozen = freezer.start()
    yield frozen
    freezer.stop()

@pytest.fixture
def future_time():
    """Fast-forward to 1 year in the future."""
    freezer = freeze_time("2026-09-01 09:00:00")
    frozen = freezer.start()
    yield frozen
    freezer.stop()

# test_with_fixtures.py
from datetime import datetime

def test_uses_frozen_time(frozen_time):
    now = datetime.now()
    assert now.year == 2025
    assert now.month == 9
    print(f"Fixture frozen: {now}")

def test_uses_future_time(future_time):
    now = datetime.now()
    assert now.year == 2026
    print(f"Future fixture: {now}")

Output (when run with pytest):

PASSED test_uses_frozen_time -- Fixture frozen: 2025-09-01 09:00:00
PASSED test_uses_future_time -- Future fixture: 2026-09-01 09:00:00

The yield pattern in the fixture ensures freezer.stop() always runs even if the test raises an exception -- equivalent to a finally block. This is the safest pattern for freezegun in pytest, as it avoids leaving time frozen if a test fails unexpectedly.

Real-Life Example: Testing a Token Expiry System

# test_token_system.py
from freezegun import freeze_time
from datetime import datetime, timedelta
import hashlib, secrets

class AuthToken:
    TTL_SECONDS = 3600  # 1 hour

    def __init__(self, user_id: str):
        self.user_id = user_id
        self.token = secrets.token_hex(16)
        self.created_at = datetime.now()
        self.expires_at = self.created_at + timedelta(seconds=self.TTL_SECONDS)

    def is_valid(self) -> bool:
        return datetime.now() < self.expires_at

    def time_remaining(self) -> int:
        """Returns seconds remaining, or 0 if expired."""
        delta = self.expires_at - datetime.now()
        return max(0, int(delta.total_seconds()))

    def refresh(self):
        """Extend token lifetime from now."""
        self.expires_at = datetime.now() + timedelta(seconds=self.TTL_SECONDS)

# Test: token is valid immediately after creation
@freeze_time("2025-10-01 12:00:00")
def test_token_valid_at_creation():
    token = AuthToken("user123")
    assert token.is_valid() is True
    assert token.time_remaining() == 3600
    print(f"Created: {token.created_at}, expires: {token.expires_at}")

# Test: token is expired 1 hour and 1 second later
def test_token_expires():
    with freeze_time("2025-10-01 12:00:00") as frozen:
        token = AuthToken("user123")
        assert token.is_valid() is True

    with freeze_time("2025-10-01 13:00:01"):
        assert token.is_valid() is False
        assert token.time_remaining() == 0
    print("Expiry test: token correctly expired after TTL")

# Test: refreshing extends lifetime
def test_token_refresh():
    with freeze_time("2025-10-01 12:00:00"):
        token = AuthToken("user123")

    with freeze_time("2025-10-01 12:59:00"):  # 1 min before expiry
        assert token.time_remaining() == 60
        token.refresh()

    with freeze_time("2025-10-01 13:58:00"):  # 1h after refresh
        assert token.is_valid() is True
        print(f"Refreshed: {token.time_remaining()}s remaining")

test_token_valid_at_creation()
test_token_expires()
test_token_refresh()

Output:

Created: 2025-10-01 12:00:00, expires: 2025-10-01 13:00:00
Expiry test: token correctly expired after TTL
Refreshed: 120s remaining

The test_token_refresh test uses three separate freeze_time context managers to simulate creating a token, approaching expiry, refreshing it, and confirming the new expiry -- all without time.sleep(). This test runs in milliseconds and is completely deterministic.

Frequently Asked Questions

Does freezegun patch all modules that import datetime?

Yes -- this is freezegun's key feature over unittest.mock.patch. When you activate freeze_time, it patches datetime.datetime, datetime.date, datetime.datetime.now, datetime.datetime.utcnow, datetime.date.today, and time.time across all loaded modules. However, it does not patch C extensions that call the system clock directly (e.g., some database drivers or asyncio event loops). In those cases you may still need additional mocking.

Does freezegun work with timezone-aware datetimes?

Yes. Pass a timezone-aware string like freeze_time("2025-06-01 12:00:00+10:00") or use freeze_time(datetime(2025, 6, 1, 12, tzinfo=timezone.utc)). Freezegun will return timezone-aware datetimes from datetime.now(tz) when called with a timezone argument. Without a timezone argument, datetime.now() returns naive datetimes at the frozen local time, same as the real function.

Does freezegun work with async code?

Yes -- freezegun patches the underlying datetime module which is called by both synchronous and asynchronous code. However, freezegun does not mock asyncio.sleep or event loop time. If your async code relies on the event loop clock (common in timeout implementations using asyncio.wait_for), you will need pytest-anyio or time_machine (a faster alternative to freezegun for async scenarios) to control event loop time.

Can I freeze to dates in the past or far future?

Yes -- any valid date string or datetime object works. freeze_time("1985-10-26 01:21:00") is perfectly valid. Dates before the Unix epoch (1970-01-01) also work. One practical caveat: some systems and libraries behave unexpectedly with dates far outside their intended range -- for example, SSL certificate validation or JWT expiry checks that use hard-coded year limits.

When should I use time-machine instead of freezegun?

time-machine is a newer alternative that works at the C level, making it faster and more reliable for async code and C extensions. If you are starting a new project with Python 3.8+, time-machine is worth considering. But if you already use freezegun and it covers your needs, there is no pressing reason to migrate -- freezegun is stable, widely used, and covers 95% of time-mocking scenarios.

Conclusion

Freezegun turns time-dependent tests from flaky guesswork into deterministic, maintainable specs. You have learned the three usage modes -- decorator, context manager, and manual start/stop -- plus tick=True for advancing time at real speed, and the pytest fixture pattern for reusable time control across a test suite. The real-life token expiry example demonstrated how to test multi-step time progressions without any time.sleep() calls.

Extend the token system by testing edge cases: a token created one second before midnight, a refresh that should fail because the token is already expired, or a batch of tokens created at different times and checked simultaneously. Freezegun makes all of these easy to write and fast to run.

For the full API reference and advanced configuration options, see the freezegun GitHub repository. For async-heavy projects, also look at time-machine as a comparison.

Related Articles

• How To Use Python responses for Mocking HTTP Requests in Tests
• How To Use Python unittest.mock for Mocking Dependencies
• How To Use Python pendulum for Better Date and Time Handling
• How To Use Python memray for Memory Profiling
• How To Use Python beartype for Runtime Type Checking
• How To Use Python nox for Automated Test Sessions
• How To Parse Dates and Times with Python dateutil

Intermediate

Python’s built-in sorting handles strings lexicographically — which means “file10.txt” sorts before “file2.txt” because “1” comes before “2” in ASCII. This is fine for computers, but deeply frustrating for anyone who has to look at a sorted file listing. Natural sort order — the order humans expect when they see numbers inside strings — is what natsort provides. It is the difference between a file manager that makes sense and one that scrambles your numbered files.

The natsort library is a pure-Python solution that detects numeric substrings inside strings and compares them as numbers, not as character sequences. It handles filenames, version strings, IP addresses, biological sequences, and locale-aware sorting with a clean, minimal API. No need to write custom sort key functions for every new case.

In this article, you will learn how natsort works, how to use natsorted() for common cases, sort paths and version numbers, use natural sort keys with sort() and sorted(), handle case-insensitive and locale-aware sorting, and build a practical file organizer that uses natural sort to group and display files. Install natsort with pip install natsort.

Written by Pubs

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

View all tutorials by Pubs →

Natural Sorting: Quick Example

# natsort_quick.py
from natsort import natsorted

files = ['file10.txt', 'file2.txt', 'file1.txt', 'file20.txt', 'file3.txt']

# Default Python sort -- lexicographic (wrong order for humans)
print("Default sort:", sorted(files))

# Natural sort -- numbers compared as integers
print("Natural sort:", natsorted(files))

Output:

Default sort: ['file1.txt', 'file10.txt', 'file2.txt', 'file20.txt', 'file3.txt']
Natural sort: ['file1.txt', 'file2.txt', 'file3.txt', 'file10.txt', 'file20.txt']

That one import and one function call is all it takes to fix the ordering. natsorted() returns a new list, just like sorted(). The rest of this article covers the full range of what natsort can do for you beyond basic filename sorting.

What Is Natural Sort Order?

Natural sort order is a collation algorithm that treats numeric substrings within strings as actual numbers when comparing. Instead of comparing character by character (‘2’ vs ‘1’ vs ‘0’), it identifies contiguous digit sequences, converts them to integers, and compares those integers. This produces an ordering that matches human intuition.

The key insight: lexicographic sorting is correct for most string data (names, words, identifiers), but breaks down whenever numeric sequences appear within strings. natsort detects these sequences and applies numeric comparison only where it is appropriate, leaving the rest of the string to standard character comparison.

Using natsorted() for Common Cases

The main entry point is natsorted(), which accepts any iterable and optional keyword arguments to control sorting behavior:

# natsort_basics.py
from natsort import natsorted, ns

# Version strings
versions = ['1.10.2', '1.9.0', '2.0.1', '1.2.3', '1.10.0']
print("Versions:", natsorted(versions))

# Mixed strings with numbers and text
items = ['item_5', 'item_12', 'item_3', 'item_10', 'item_1']
print("Items:", natsorted(items))

# Case-insensitive natural sort
mixed_case = ['File10.txt', 'file2.TXT', 'FILE1.txt', 'File20.txt']
print("Case-insensitive:", natsorted(mixed_case, alg=ns.IGNORECASE))

# Reverse natural sort
print("Reversed:", natsorted(versions, reverse=True))

# Sort with a key function (e.g., sort tuples by second element)
data = [('b', 'v10'), ('a', 'v2'), ('c', 'v1'), ('d', 'v20')]
print("By version:", natsorted(data, key=lambda x: x[1]))

Output:

Versions: ['1.2.3', '1.9.0', '1.10.0', '1.10.2', '2.0.1']
Items: ['item_1', 'item_3', 'item_5', 'item_10', 'item_12']
Case-insensitive: ['FILE1.txt', 'file2.TXT', 'File10.txt', 'File20.txt']
Reversed: ['2.0.1', '1.10.2', '1.10.0', '1.9.0', '1.2.3']
By version: [('c', 'v1'), ('a', 'v2'), ('b', 'v10'), ('d', 'v20')]

Sorting File Paths

File path sorting has an extra wrinkle: you usually want to sort directory components separately from filenames, and you want numbers within both components to sort naturally. natsort’s ns.PATH algorithm handles this correctly:

# natsort_paths.py
from natsort import natsorted, ns
from pathlib import Path

# Simulate a file listing with numbered paths
paths = [
    '/data/project10/output2.csv',
    '/data/project2/output10.csv',
    '/data/project1/output1.csv',
    '/data/project2/output2.csv',
    '/data/project10/output1.csv',
]

# PATH algorithm: sorts directory components and filenames separately
sorted_paths = natsorted(paths, alg=ns.PATH)
for p in sorted_paths:
    print(p)

Output:

/data/project1/output1.csv
/data/project2/output2.csv
/data/project2/output10.csv
/data/project10/output1.csv
/data/project10/output2.csv

Without ns.PATH, the slash characters in paths can interfere with the numeric detection. The PATH algorithm splits on path separators first, then applies natural sort within each component — giving you the result a file manager would show.

Using natsort_keygen() with sort()

If you need to sort an existing list in place (rather than getting a new sorted list), use natsort_keygen() to get a key function you can pass to list.sort():

# natsort_key.py
from natsort import natsort_keygen, ns

# Sort a list in place
logs = ['error_10.log', 'error_2.log', 'error_1.log', 'error_100.log']
key = natsort_keygen()
logs.sort(key=key)
print("Sorted logs:", logs)

# Use as a sort key in more complex operations
students = [
    {'name': 'Alice', 'grade': 'Grade 10'},
    {'name': 'Bob', 'grade': 'Grade 2'},
    {'name': 'Carol', 'grade': 'Grade 1'},
    {'name': 'Dave', 'grade': 'Grade 20'},
]
grade_key = natsort_keygen(key=lambda s: s['grade'])
students.sort(key=grade_key)
for s in students:
    print(f"  {s['grade']}: {s['name']}")

Output:

Sorted logs: ['error_1.log', 'error_2.log', 'error_10.log', 'error_100.log']
  Grade 1: Carol
  Grade 2: Bob
  Grade 10: Alice
  Grade 20: Dave

Real-Life Example: File Report Generator

This script scans a directory, groups files by type, sorts them naturally within each group, and produces a clean report with file sizes:

# file_report.py
from pathlib import Path
from natsort import natsorted, ns
from collections import defaultdict

def format_size(bytes_count: int) -> str:
    """Format bytes as human-readable size."""
    for unit in ['B', 'KB', 'MB', 'GB']:
        if bytes_count < 1024:
            return f"{bytes_count:.1f} {unit}"
        bytes_count /= 1024
    return f"{bytes_count:.1f} TB"

def generate_file_report(folder: str) -> None:
    base = Path(folder)
    if not base.exists():
        print(f"Folder not found: {folder}")
        return

    # Group files by extension
    groups = defaultdict(list)
    for path in base.rglob('*'):
        if path.is_file():
            ext = path.suffix.lower() or '.no_ext'
            groups[ext].append(path)

    # Sort extension groups naturally, then files within each group
    sorted_exts = natsorted(groups.keys())

    total_files = 0
    total_size = 0

    for ext in sorted_exts:
        files = natsorted(groups[ext], key=lambda p: p.name, alg=ns.PATH)
        ext_size = sum(f.stat().st_size for f in files)
        print(f"\n{ext} ({len(files)} files, {format_size(ext_size)})")
        print("-" * 40)
        for f in files[:5]:  # Show first 5 to keep output manageable
            size = f.stat().st_size
            rel = f.relative_to(base)
            print(f"  {str(rel):<35} {format_size(size):>8}")
        if len(files) > 5:
            print(f"  ... and {len(files) - 5} more")
        total_files += len(files)
        total_size += ext_size

    print(f"\nTotal: {total_files} files, {format_size(total_size)}")

generate_file_report('./my_project')

Output:

.csv (3 files, 48.2 KB)
----------------------------------------
  data/output1.csv                  12.1 KB
  data/output2.csv                  18.4 KB
  data/output10.csv                 17.7 KB

.py (5 files, 22.8 KB)
----------------------------------------
  src/module1.py                     4.2 KB
  src/module2.py                     5.1 KB
  src/module10.py                    6.8 KB
  src/module11.py                    3.9 KB
  src/utils.py                       2.8 KB

Total: 8 files, 71.0 KB

Frequently Asked Questions

Does natsort handle floating-point numbers in strings?

Yes — use the ns.FLOAT algorithm flag: natsorted(items, alg=ns.FLOAT). This treats sequences like 3.14 and 2.71 as floating-point numbers rather than two separate integer sequences. You can combine flags: ns.FLOAT | ns.IGNORECASE. For version strings specifically, ns.VERSION (or ns.V) is a dedicated shortcut that handles dotted version numbers correctly.

How do I sort with locale-aware character ordering?

Use natsorted(items, alg=ns.LOCALE) after calling locale.setlocale(locale.LC_ALL, '') to set the system locale. This sorts accented characters (like e, a, o in European languages) according to the locale’s collation rules rather than Unicode code point order. This is important for sorting names in languages where a and a are considered equal or adjacent in the alphabet.

Can I use natsort with pandas DataFrames?

Yes. Use natsort_keygen() as the key parameter in pandas sort_values(): df.sort_values('column', key=lambda s: s.map(natsort_keygen())). For sorting DataFrame row labels (index), use df.reindex(natsorted(df.index)). The natsort documentation has a dedicated pandas section with more examples for multi-column natural sorting.

Does natsort handle negative numbers?

By default, natsort does not treat a leading minus sign as part of a number — “file-10.txt” sorts as the string “file-” followed by natural sort on “10.txt”. To enable signed number handling, use ns.SIGNED: natsorted(items, alg=ns.SIGNED). This is useful for data with negative values in filenames or log entries, but can produce unexpected results if hyphens are used as separators in non-numeric contexts.

Is natsort fast enough for large lists?

natsort is implemented in pure Python with optional C acceleration via the fastnumbers package. Install it with pip install fastnumbers to get 3-5x speedup on numeric-heavy data. For most real-world use cases (thousands of filenames, hundreds of version strings), natsort’s performance is more than adequate. If you are sorting millions of strings, profile first — the Python overhead may matter more than the sorting algorithm itself.

Conclusion

You now know how to use natsort to make string sorting behave the way humans expect. We covered natsorted() for common cases, the ns.PATH algorithm for file system paths, natsort_keygen() for in-place sorting and complex key functions, case-insensitive and locale-aware sorting with ns flags, and a practical file report generator that groups and naturally sorts files by extension.

The key takeaway is simple: any time you are sorting strings that contain numbers — filenames, version strings, log entries, chapter titles, IP addresses — reach for natsort instead of writing a custom key function. It handles the edge cases (floats, negative numbers, leading zeros, locale) that custom one-liners miss.

See the natsort documentation for the full list of algorithm flags, pandas integration examples, and performance tips.

Related Articles

• How To Use Python memray for Memory Profiling
• How To Use Python beartype for Runtime Type Checking
• How To Use Python nox for Automated Test Sessions
• How To Use Python Typer and Click for Building Complex CLIs
• How To Use Python Enums and When You Should
• How To Use Python Click for CLI Applications
• How To Parse Dates and Times with Python dateutil

Intermediate

Dealing with dates and times in Python’s standard datetime module is notoriously error-prone. Timezone handling is verbose and fragile, parsing requires format strings you always have to look up, and arithmetic near daylight saving time transitions can silently give you wrong answers. If you have ever had a cron job fire at the wrong time after DST, or struggled to express “3 months from now” in stdlib, you will appreciate pendulum.

pendulum is a drop-in replacement for Python’s datetime that makes timezone-aware operations the default, not an afterthought. Every pendulum.DateTime object is timezone-aware, DST transitions are handled correctly, parsing is smart (ISO 8601 by default), and date math reads like plain English. It is widely used in data pipelines, scheduling systems, and API integrations where timezone correctness matters.

In this article, you will learn how to create and parse pendulum datetimes, work with timezones safely, perform date arithmetic, format dates for display and APIs, use periods and durations, and build a practical meeting scheduler that handles cross-timezone coordination. Install pendulum with pip install pendulum.

Written by Pubs

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

View all tutorials by Pubs →

pendulum Quick Example

# pendulum_quick.py
import pendulum

now = pendulum.now('Europe/London')
print("London time:", now.to_datetime_string())
print("Tokyo time:", now.in_timezone('Asia/Tokyo').to_datetime_string())
print("In 3 weeks:", now.add(weeks=3).to_date_string())
print("Diff from now:", pendulum.now().diff_for_humans(pendulum.now().subtract(hours=5)))

Output:

London time: 2026-05-02 10:00:00
Tokyo time: 2026-05-02 18:00:00
In 3 weeks: 2026-05-23
Diff from now: 5 hours ago

Every datetime created by pendulum is timezone-aware. Converting between timezones is a single method call. The diff_for_humans() method produces readable strings like “5 hours ago” or “in 3 weeks” — perfect for user-facing interfaces. The rest of this article dives deep into each of these capabilities.

What Is pendulum and How Does It Compare to datetime?

pendulum extends Python’s datetime.datetime class — every pendulum instance is a proper datetime, so it works anywhere datetime is expected. The key differences are in defaults and ergonomics: stdlib datetime objects are naive (no timezone) by default and require explicit pytz or zoneinfo handling, while every pendulum datetime has a timezone from creation.

pendulum uses the IANA timezone database (via tzdata) for all timezone operations. This means it knows about every DST rule for every timezone, including historical changes — an important detail for handling records from the past.

Creating and Parsing Datetimes

pendulum gives you several ways to create datetime objects. The most important: pendulum.now() for the current time, pendulum.datetime() for a specific moment, and pendulum.parse() for strings:

# pendulum_create.py
import pendulum

# Current time in a specific timezone
now_nyc = pendulum.now('America/New_York')
print("NYC now:", now_nyc)

# Create a specific datetime
launch = pendulum.datetime(2026, 6, 1, 9, 0, 0, tz='UTC')
print("Launch:", launch.to_iso8601_string())

# Parse from ISO 8601 string (no format string needed)
deadline = pendulum.parse('2026-12-31T23:59:59+05:30')
print("Deadline:", deadline.timezone_name, deadline.to_datetime_string())

# Parse a date-only string (becomes midnight UTC)
day = pendulum.parse('2026-05-15')
print("Day:", day.to_date_string())

# From a Unix timestamp
event = pendulum.from_timestamp(1746129600, tz='Asia/Tokyo')
print("Event in Tokyo:", event.to_datetime_string())

Output:

NYC now: 2026-05-02T06:00:00-04:00
Launch: 2026-06-01T09:00:00+00:00
Deadline: Asia/Kolkata 2026-12-31 23:59:59
Day: 2026-05-15
Event in Tokyo: 2026-05-02 09:00:00

Date Arithmetic with add() and subtract()

pendulum’s add() and subtract() methods accept any combination of years, months, weeks, days, hours, minutes, and seconds. Unlike timedelta, these methods understand calendar arithmetic — adding 1 month to January 31st gives you February 28th, not an error or overflow:

# pendulum_arithmetic.py
import pendulum

now = pendulum.now('UTC')

# Add calendar units
print("In 3 months:", now.add(months=3).to_date_string())
print("In 2 weeks:", now.add(weeks=2).to_date_string())
print("Yesterday:", now.subtract(days=1).to_date_string())
print("Next quarter:", now.add(months=3).start_of('month').to_date_string())

# End/start of period helpers
print("Start of week:", now.start_of('week').to_date_string())
print("End of month:", now.end_of('month').to_date_string())
print("Start of year:", now.start_of('year').to_date_string())

# DST-aware arithmetic -- clocks spring forward at 2am
eastern = pendulum.datetime(2026, 3, 8, 1, 30, 0, tz='America/New_York')
one_hour_later = eastern.add(hours=1)
print("\nBefore DST:", eastern.to_datetime_string(), eastern.timezone_abbreviation)
print("After +1hr:", one_hour_later.to_datetime_string(), one_hour_later.timezone_abbreviation)

Output:

In 3 months: 2026-08-02
In 2 weeks: 2026-05-16
Yesterday: 2026-05-01
Next quarter: 2026-08-01
Start of week: 2026-04-27
End of month: 2026-05-31
Start of year: 2026-01-01

Before DST: 2026-03-08 01:30:00 EST
After +1hr: 2026-03-08 03:30:00 EDT

Notice the DST transition: 1:30 AM + 1 hour = 3:30 AM (clocks skip 2:00-3:00 AM). stdlib timedelta arithmetic does not handle this correctly — you would need to use zoneinfo explicitly. pendulum does it automatically.

Periods and Durations

A Period in pendulum is the range between two datetimes. Unlike a bare timedelta, a Period knows about calendar months and years, so you can get “2 months and 5 days” rather than just a raw count of seconds. A Duration is an absolute measure of time (like timedelta) but with the same readable interface:

# pendulum_periods.py
import pendulum

start = pendulum.datetime(2026, 1, 15, tz='UTC')
end = pendulum.datetime(2026, 5, 2, tz='UTC')

period = end - start  # Returns a Period
print("Total days:", period.days)
print("In weeks:", period.weeks)
print("In months:(", period.in_months())
print("Remaining days:", period.remaining_days)
print("Human:", period.in_words())

# Iterate over dates in a range
print("\nFirst 3 days of May:")
may = pendulum.period(
    pendulum.datetime(2026, 5, 1, tz='UTC'),
    pendulum.datetime(2026, 5, 3, tz='UTC')
)
for dt in may.range('days'):
    print(" ", dt.to_date_string(), dt.day_of_week_name)

Output:

Total days: 107
In weeks: 15
In months: 3
Qmaining days: 22
Human: 3 months 2 weeks 2 days

First 3 days of May:
  2026-05-01 Friday
  2026-05-02 Saturday
  2026-05-03 Sunday

Real-Life Example: Cross-Timezone Meeting Scheduler

This scheduler takes a proposed meeting time in UTC and shows it in each participant’s timezone, flags unsociable hours, and generates an iCal-compatible timestamp:

# meeting_scheduler.py
import pendulum

TEAM = {
    "Alice (NYC)":    "America/New_York",
    "Bob (London)":   "Europe/London",
    "Carol (Sydney)": "Australia/Sydney",
    "Dave (Tokyo)":   "Asia/Tokyo",
}

WORK_HOURS = range(8, 18)  # 8am - 6pm considered acceptable

def schedule_meeting(utc_time_str: str, duration_minutes: int = 60) -> None:
    meeting_utc = pendulum.parse(utc_time_str, tz='UTC')
    meeting_end = meeting_utc.add(minutes=duration_minutes)

    print(f"Meeting: {meeting_utc.to_iso8601_string()}")
    print(f"Duration: {duration_minutes} minutes")
    print("-" * 55)

    all_ok = True
    for person, tz in TEAM.items():
        local = meeting_utc.in_timezone(tz)
        local_end = meeting_end.in_timezone(tz)
        hour = local.hour
        is_ok = hour in WORK_HOURS
        if not is_ok:
            all_ok = False
        status = "OK " if is_ok else "BAD"
        print(f"[{status}] {person:<20} {local.format('ddd DD MMM HH:mm')} - {local_end.format('HH:mm')} {local.timezone_abbreviation}")

    print("-" * 55)
    if all_ok:
        print("All team members available during work hours.")
    else:
        print("WARNING: Some participants outside work hours.")

    # iCal-compatible timestamp
    print(f"\nDTSTART:{meeting_utc.format('YYYYMMDDTHHmmss')}Z")
    print(f"DTEND:{meeting_end.format('YYYYMMDDTHHmmss')}Z")

schedule_meeting("2026-05-04T14:00:00Z", duration_minutes=45)

Output:

Meeting: 2026-05-04T14:00:00+00:00
Duration: 45 minutes
-------------------------------------------------------
[OK ] Alice (NYC)          Mon 04 May 10:00 - 10:45 ETT
[OK ] Bob (London)         Mon 04 May 15:00 - 15:45 BST
[BAD] Carol (Sydney)       Tue 05 May 00:00 - 00:45 AEST
[OK ] Dave (Tokyo)         Mon 04 May 23:00 - 23:45 JST
-------------------------------------------------------
WARNING: Some participants outside work hours.

DTSTART:20260504T140000Z
DTEND:20260504T144500Z

Frequently Asked Questions

Is pendulum compatible with stdlib datetime?

Yes -- pendulum's DateTime is a subclass of datetime.datetime. You can pass pendulum objects anywhere a datetime is expected: to database drivers, to json.dumps with a custom encoder, to Django model fields, to pandas Timestamp constructors. The only caveat is that libraries that create datetime objects (like SQLAlchemy or requests) return plain datetime, not pendulum -- use pendulum.instance(dt) to convert them.

What happens if I pass a naive datetime to pendulum?

Use pendulum.instance(naive_dt, tz='UTC') to wrap a naive datetime with a timezone. Never assume the timezone of a naive datetime -- if you are not sure whether it's local time or UTC, you need to find out from the data source. Silently assuming UTC for naive datetimes from user input or database rows is a common source of off-by-one-hour bugs after DST transitions.

What format strings does pendulum use?

pendulum uses its own token-based format strings in the .format() method (e.g., 'YYYY-MM-DD HH:mm:ss'), which are different from stdlib's strftime codes. There are also convenience methods: .to_iso8601_string(), .to_date_string(), .to_datetime_string(), .to_time_string(), and .to_rfc2822_string(). Use the convenience methods for common formats and .format() for custom patterns.

How do I store pendulum datetimes in a database?

Always store datetimes as UTC in the database. Convert to the user's timezone only for display. With SQLAlchemy, use a DateTime(timezone=True) column -- this stores and retrieves UTC-aware datetimes. In the application layer, convert: pendulum.instance(db_datetime).in_timezone(user_tz). This pattern prevents timezone ambiguity and makes your data portable across servers in different regions.

What changed in pendulum 3.x?

pendulum 3.0 (released 2024) rewrote the internals to use Python's built-in zoneinfo module (available since Python 3.9) instead of a bundled timezone database. This makes it lighter and keeps timezone data in sync with your OS. The API is largely backward-compatible, but check your pendulum.timezone() calls -- some legacy timezone names were updated. If you are on Python 3.8 or below, install the backports.zoneinfo package alongside pendulum 3.

Conclusion

You now have a complete toolkit for timezone-aware date and time handling with pendulum. We covered creating and parsing datetimes with automatic timezone support, performing calendar-aware arithmetic with add() and subtract() that handles DST transitions correctly, working with Periods for range iteration and human-readable durations, and a practical cross-timezone meeting scheduler that flags unsociable hours and outputs iCal timestamps.

The fundamental rule to take away: always work in UTC internally and convert to local time only for display. pendulum makes this the path of least resistance -- every pendulum.now('UTC') call gives you a timezone-aware datetime, and .in_timezone(user_tz) handles the conversion. Extend the meeting scheduler to find the optimal meeting time that minimizes "bad hours" across all participants -- that is a fun exercise in Period arithmetic.

See the pendulum documentation for the full list of format tokens, locale support, and testing helpers.

Related Articles

• How To Use Python freezegun for Mocking Time in Tests
• How To Use Python Type Hints For Better Code Quality
• Better organization of your projects with python imports
• How To Use Python whenever for Correct Date and Time Handling
• How To Use Python memray for Memory Profiling
• How To Use Python beartype for Runtime Type Checking
• How To Use Python icecream for Better Debugging

Intermediate

Standard Python logging gives you timestamped text messages, but when you are debugging a production incident at 2am, those messages rarely give you the context you need. Which user triggered the error? Which request ID? What was the state of the application at that moment? Structured logging solves this by making every log event a machine-readable data record — a dict of key-value pairs instead of a raw string. structlog is the leading Python library for structured logging, and it makes this easy while staying fully compatible with the standard logging module.

structlog works by building up a context dictionary as your code executes, then rendering it as JSON (or pretty-printed for development) when you emit a log event. You bind values once — like a request ID or user ID — and they appear automatically in every subsequent log message from that context. No more copy-pasting request IDs into every log call.

In this article, you will learn how to configure structlog, bind context with bind and new, set up processors for development and production output, integrate with the standard logging module, add request context in a Flask app, and build a practical async worker with rich structured logging. Install structlog with pip install structlog before starting.

Written by Pubs

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

View all tutorials by Pubs →

Structured Logging: Quick Example

# structlog_quick.py
import structlog

log = structlog.get_logger()

log.info("user_login", user_id=42, email="alice@example.com", source="web")
log.warning("login_failed", user_id=99, reason="invalid_password", attempt=3)

Output (development mode — pretty printed):

2026-05-02 10:00:00 [info     ] user_login         email=alice@example.com source=web user_id=42
2026-05-02 10:00:00 [warning  ] login_failed       attempt=3 reason=invalid_password user_id=99

Each log call takes a positional event string followed by keyword arguments that become fields in the structured record. In development, structlog renders these in a readable format; in production, you configure it to output JSON. The same code, different renderers — you never change your log calls based on environment.

What Is structlog and Why Use It?

structlog replaces the pattern of building log messages from string concatenation (f"User {user_id} failed to login with error {error}") with a pattern of logging discrete fields. This makes logs searchable and filterable in tools like Elasticsearch, Datadog, or CloudWatch Logs Insights — you can query user_id=42 and get all events for that user, rather than running regex searches over raw strings.

structlog does not replace stdlib logging — it wraps around it. In most production setups, structlog acts as the frontend you write to, while stdlib logging handles the backend (file handlers, syslog, etc.). This means you can adopt structlog incrementally in an existing codebase.

Configuring structlog for Development and Production

structlog is configured once at application startup. The key is the processor chain — a list of functions that transform the log event dict before rendering. Here is a complete configuration for both environments:

# structlog_config.py
import structlog
import logging
import sys
import os

def configure_logging() -> None:
    """Configure structlog for the current environment."""
    is_production = os.getenv('ENV', 'development') == 'production'

    shared_processors = [
        structlog.contextvars.merge_contextvars,       # Thread/async context
        structlog.stdlib.add_log_level,                # Add 'level' field
        structlog.stdlib.add_logger_name,              # Add 'logger' field
        structlog.processors.TimeStamper(fmt='iso'),   # ISO 8601 timestamp
        structlog.processors.StackInfoRenderer(),      # Stack traces
        structlog.processors.format_exc_info,          # Exception formatting
    ]

    if is_production:
        # JSON output for log aggregation tools
        renderer = structlog.processors.JSONRenderer()
    else:
        # Human-readable colored output for terminals
        renderer = structlog.dev.ConsoleRenderer(colors=True)

    structlog.configure(
        processors=shared_processors + [renderer],
        wrapper_class=structlog.make_filtering_bound_logger(logging.DEBUG),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

configure_logging()
log = structlog.get_logger("myapp")

log.info("app_started", version="1.2.3", environment=os.getenv('ENV', 'development'))
log.debug("config_loaded", database="postgres", cache="redis")

Output (development):

2026-05-02T10:00:00Z [debug    ] config_loaded      cache=redis database=postgres logger=myapp
2026-05-02T10:00:00Z [info     ] app_started        environment=development logger=myapp version=1.2.3

Output (production, ENV=production):

{"cache": "redis", "database": "postgres", "event": "config_loaded", "level": "debug", "logger": "myapp", "timestamp": "2026-05-02T10:00:00Z"}
{"environment": "production", "event": "app_started", "level": "info", "logger": "myapp", "timestamp": "2026-05-02T10:00:00Z", "version": "1.2.3"}

Binding Context with bind() and new()

The most powerful feature of structlog is context binding. Use bind() to add fields that persist for the lifetime of a logger, and new() to create a fresh logger with a clean context:

# structlog_bind.py
import structlog

log = structlog.get_logger()

def process_order(order_id: str, user_id: int) -> None:
    # Bind context once -- all subsequent logs from this logger include it
    order_log = log.bind(order_id=order_id, user_id=user_id)

    order_log.info("order_processing_started")

    # Further bind additional context
    item_log = order_log.bind(item_count=3, total_amount=99.99)
    item_log.info("items_validated")

    try:
        # Simulate an operation
        if order_id == "ORD-999":
            raise ValueError("Invalid payment method")
        item_log.info("payment_processed", gateway="stripe")
    except ValueError as e:
        item_log.error("payment_failed", error=str(e), retry=False)
        raise

    order_log.info("order_processing_complete")

process_order("ORD-101", user_id=42)

Output:

2026-05-02 [info ] order_processing_started  order_id=ORD-101 user_id=42
2026-05-02 [info ] items_validated           item_count=3 order_id=ORD-101 total_amount=99.99 user_id=42
2026-05-02 [info ] payment_processed         gateway=stripe item_count=3 order_id=ORD-101 total_amount=99.99 user_id=42
2026-05-02 [info ] order_processing_complete order_id=ORD-101 user_id=42

Notice how order_id and user_id appear in every log message without being passed repeatedly. When the error occurs, the full context is already present — you see exactly which order failed, for which user, with which item count. This is what makes structured logging so powerful for debugging production issues.

Integrating with stdlib logging

Most production codebases already use stdlib logging — your application, your third-party libraries, the frameworks they depend on. structlog plays nicely with all of them: it can wrap stdlib loggers (so existing code keeps working) AND redirect stdlib output through structlog’s pipeline (so everything ends up as JSON):

# File: hybrid_logging.py
import logging
import structlog

# 1) Configure stdlib logging to forward to structlog's processor chain
logging.basicConfig(
    format="%(message)s",
    level=logging.INFO,
)

# 2) Configure structlog to render as JSON
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# 3) Use either API — same JSON output
struct_logger = structlog.get_logger("api")
struct_logger.info("via structlog", user_id=42)

stdlib_logger = logging.getLogger("legacy")
stdlib_logger.info("via stdlib")

The processor chain is the heart of structlog. Each processor receives the event dict and returns the modified dict. JSONRenderer at the end serializes it. By inserting format_exc_info and StackInfoRenderer in the chain, every logger.exception() call gets a clean JSON traceback without you having to do anything.

Custom Processors for Filtering and Enrichment

A processor is just a function with signature (logger, method_name, event_dict) -> event_dict. You write them like middleware — each one inspects the dict, can mutate it, and returns either the dict or raises DropEvent to drop the line entirely. Two patterns you’ll use constantly:

# File: custom_processors.py
import os
import structlog

def add_environment(logger, method_name, event_dict):
    """Inject deploy environment into every log line."""
    event_dict["env"] = os.environ.get("APP_ENV", "dev")
    event_dict["service"] = "payments-api"
    return event_dict

def redact_secrets(logger, method_name, event_dict):
    """Strip values for any key that looks sensitive."""
    for key in list(event_dict):
        if any(s in key.lower() for s in ("password", "token", "secret", "card")):
            event_dict[key] = "***REDACTED***"
    return event_dict

structlog.configure(
    processors=[
        add_environment,
        redact_secrets,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)

log = structlog.get_logger()
log.info("charge attempted", user_id=42, card_number="4111111111111111", amount=99)

Output:

{"env": "dev", "service": "payments-api", "user_id": 42, "card_number": "***REDACTED***", "amount": 99, "timestamp": "2026-05-16T12:34:56", "event": "charge attempted"}

The redact_secrets processor is the most useful one most projects don’t have. Run every log through it and you can’t accidentally ship raw tokens or PII to your log aggregator. The add_environment processor solves the “which deploy did this come from” question that always shows up during incident response.

Async, FastAPI, and Per-Request Context

In an async web app, the killer feature of structlog is contextvars-based logging — bind context once per request and every log line inside that request inherits it automatically, across await boundaries:

# File: fastapi_structlog.py
from uuid import uuid4
import structlog
from fastapi import FastAPI, Request

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)

log = structlog.get_logger()
app = FastAPI()

@app.middleware("http")
async def log_context(request: Request, call_next):
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=str(uuid4()),
        path=request.url.path,
        method=request.method,
    )
    log.info("request_started")
    response = await call_next(request)
    log.info("request_completed", status=response.status_code)
    return response

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    log.info("looking up item", item_id=item_id)  # request_id auto-included
    return {"item_id": item_id}

Every log line emitted from anywhere inside the request — your handler, a service called by the handler, a coroutine awaited deep inside — automatically carries request_id, path, and method. When something breaks at 3 AM, you grep one request_id across the log aggregator and see the full timeline of a single request.

Production Patterns and Performance

• Configure once at startup. structlog.configure() should be called from your main entry point before anything else logs. Re-configuring at runtime is supported but confusing.
• Cache the logger. Set cache_logger_on_first_use=True in configure(). Without it, structlog rebuilds the bound logger on every call, which is slow under load.
• Use lazy formatting. structlog already does this for free — you pass kwargs, not f-strings. log.info("query ran", duration=t) is faster and safer than log.info(f"query ran {t}s").
• Avoid logging in tight loops. JSON serialization isn’t free. If you’re logging inside a 1M-iteration loop, log a summary at the end instead.
• Pin your processors in tests. Use structlog.testing.LogCapture as a test fixture to assert that specific events were logged with specific fields. Don’t grep stdout in tests.

FAQ

Q: structlog or python-json-logger — which one?
A: python-json-logger is a tiny adapter that just makes stdlib logging output JSON. structlog is a full structured-logging system with bound context, processor chains, and contextvars integration. For “I just want JSON”, use python-json-logger. For anything beyond that, structlog.

Q: Does structlog replace the stdlib logging module?
A: It can, but it doesn’t have to. The recommended setup is hybrid: structlog handles your code’s logs, stdlib logging keeps working for third-party libraries, and both end up in the same JSON output via the processor chain.

Q: How do I get structlog working with Django?
A: Configure structlog in settings.py alongside Django’s LOGGING dict. Use structlog.stdlib.LoggerFactory() as the factory so Django’s existing log levels and handlers still apply. Add structlog.contextvars.merge_contextvars as the first processor and bind request_id in middleware.

Q: Can I use structlog in scripts/CLIs, not just web apps?
A: Yes — structlog has no web-framework dependencies. The contextvars features are also useful in any async or threaded code, not just HTTP request handlers.

Q: What’s the performance cost vs stdlib logging?
A: With cache_logger_on_first_use=True and a JSON renderer, structlog is roughly 2-3x slower than bare logger.info("msg"). In practice, log emission is rarely a hotpath — you’ll feel it only if you’re logging inside a per-millisecond loop, which you shouldn’t be doing anyway.

Wrapping Up

structlog turns Python logging from “strings that you grep later” into “events that you query.” The two features that matter most are bind() (carry context forward) and the processor pipeline (filter, enrich, and redact in one place). Combined with contextvars integration in web apps, you get request-scoped log context for free — no manual passing of correlation IDs through every function.

The official structlog documentation has the full reference, including integrations with sentry, OpenTelemetry, and various web frameworks beyond FastAPI.

Related Python Tutorials

• How To Set Up Logging in Python 3 (Output to File and Console)
• How To Use the Python Logging Module for Application Logging
• How To Use Python Decorators for Logging and Timing

Related Articles

• How To Use Python Loguru for Simplified Logging
• How To Use Python Loguru for Modern Application Logging
• How To Use the Python logging Module for Application Logging
• How To Set Up Logging in Python 3 (Output to File and Console)
• How To Use Python Decorators To Add Logging and Timing to Functions
• How To Set Up Python Logging With Rotating File Handlers
• How To Use Python trio for Structured Concurrency

Intermediate

If you have ever opened a CSV file and seen garbled characters like a EUR" or \xc3\xa9 where accented letters should be, you have hit an encoding problem. Character encoding is one of the most frustrating silent failures in data pipelines — files load without errors but the content is wrong. Python’s default assumption is UTF-8, but a huge proportion of legacy files, especially from Windows systems, are encoded in latin-1, Windows-1252, or Shift-JIS. The chardet library detects the encoding automatically so you can fix these problems reliably.

chardet is a port of the encoding detection algorithm from Mozilla Firefox. It analyzes the byte patterns in your data and returns the most likely encoding along with a confidence score. You install it with a single pip command and use it with just a few lines of code. No manual byte inspection needed.

In this article, you will learn how to detect encodings with chardet, use it to open files correctly, detect encodings in bulk across directories, build a file encoding converter, and handle the edge cases that trip up beginners. By the end, you will have a reliable toolkit for dealing with any encoding problem you encounter in real data work.

Written by Pubs

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

View all tutorials by Pubs →

Detecting Encoding: Quick Example

# detect_encoding.py
import chardet

# Detect encoding of a byte string
data = b'\xc3\xa9l\xc3\xa8ve'  # "eleve" in UTF-8 with accents
result = chardet.detect(data)
print(result)

# Decode it correctly using the detected encoding
text = data.decode(result['encoding'])
print("Decoded text:", text)

Output:

{'encoding': 'utf-8', 'confidence': 0.99, 'language': ''}
Decoded text: eleve

The detect function returns a dict with three keys: encoding (the detected charset), confidence (0.0 to 1.0 — how certain chardet is), and language (a hint for multi-byte encodings like Chinese or Japanese). The confidence score is your reliability signal — anything above 0.8 is generally trustworthy for western encodings.

What Is Character Encoding and Why Does It Break?

A character encoding is a mapping between characters (letters, symbols, emoji) and bytes. UTF-8 is the modern universal standard — it encodes all Unicode characters and is the default in Python 3. But hundreds of older encodings exist, each covering a subset of characters in their own byte layout. When software writes a file in Windows-1252 but Python reads it as UTF-8, certain byte sequences are invalid UTF-8 and Python raises a UnicodeDecodeError — or worse, silently misinterprets them.

chardet solves this by treating encoding detection as a statistical problem — it looks at byte frequency distributions and byte-order marks (BOMs) to make an educated guess. It works best with longer strings (100+ bytes) and can struggle with very short snippets that do not contain enough pattern data.

Opening Files with Auto-Detected Encoding

The most common real-world use is opening a file whose encoding you do not know. Read the raw bytes, detect the encoding, then decode:

# open_detected.py
import chardet

def read_file_auto(filepath: str) -> str:
    """Read a text file with automatic encoding detection."""
    with open(filepath, 'rb') as f:  # Read as raw bytes
        raw = f.read()

    result = chardet.detect(raw)
    encoding = result['encoding']
    confidence = result['confidence']

    if confidence < 0.7:
        print(f"Warning: Low confidence ({confidence:.0%}) for {encoding}")

    print(f"Detected: {encoding} ({confidence:.0%} confidence)")
    return raw.decode(encoding, errors='replace')

# Test with a windows-1252 encoded file
text = read_file_auto('legacy_report.csv')
print(text[:100])

Output:

Detected: Windows-1252 (98% confidence)
Name,Department,Salary
Francois Dupont,Finance,52000
Rene Müller,Engineering,67000

The errors='replace' argument in decode substitutes the Unicode replacement character (U+FFFD) for any bytes that still cannot be decoded, instead of raising an error. This is a safe fallback for production pipelines where you cannot afford crashes, though you should log any replacement occurrences for later review.

Incremental Detection for Large Files

For large files, loading everything into memory just to detect encoding is wasteful. chardet's UniversalDetector lets you feed data in chunks and stop as soon as confidence is high enough:

# incremental_detect.py
from chardet.universaldetector import UniversalDetector

def detect_encoding_streaming(filepath: str, chunk_size: int = 4096) -> dict:
    """Detect encoding of a large file without loading it all into memory."""
    detector = UniversalDetector()
    with open(filepath, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            detector.feed(chunk)
            if detector.done:  # Confident enough -- stop early
                break
    detector.close()
    return detector.result

result = detect_encoding_streaming('large_log_file.txt')
print(f"Encoding: {result['encoding']}, Confidence: {result['confidence']:.1%}")

Output:

Encoding: UTF-8-SIG, Confidence: 100.0%

UTF-8-SIG means the file has a UTF-8 byte-order mark (BOM) -- a three-byte sequence at the start (\xef\xbb\xbf) that some Windows tools add. Python's utf-8-sig codec strips the BOM automatically when decoding, which is almost always what you want. Without chardet, you would have to check for the BOM manually.

Bulk Encoding Converter

When cleaning up a directory of legacy files, you often need to convert them all to UTF-8. This script scans a folder, detects each file's encoding, and converts non-UTF-8 files in place:

# bulk_convert.py
import chardet
from pathlib import Path

def convert_to_utf8(folder: str, extensions: list) -> None:
    """Convert all matching files in a folder to UTF-8."""
    base = Path(folder)
    converted = 0
    skipped = 0
    errors = 0

    for ext in extensions:
        for filepath in base.rglob(f'*.{ext}'):
            try:
                raw = filepath.read_bytes()
                result = chardet.detect(raw)
                encoding = result['encoding']

                if encoding is None:
                    print(f"Could not detect: {filepath.name}")
                    errors += 1
                    continue

                if encoding.lower().replace('-', '') in ('utf8', 'utf8sig', 'ascii'):
                    skipped += 1
                    continue  # Already UTF-8 compatible

                # Decode with detected encoding, re-encode as UTF-8
                text = raw.decode(encoding, errors='replace')
                filepath.write_text(text, encoding='utf-8')
                print(f"Converted {filepath.name}: {encoding} -> UTF-8")
                converted += 1

            except Exception as e:
                print(f"Error processing {filepath.name}: {e}")
                errors += 1

    print(f"\nDone: {converted} converted, {skipped} skipped, {errors} errors")

convert_to_utf8('./data', extensions=['csv', 'txt', 'log'])

Output:

Converted customers.csv: Windows-1252 -> UTF-8
Converted old_report.txt: ISO-8859-1 -> UTF-8
Converted access.log: ascii -> UTF-8

Done: 3 converted, 12 skipped, 0 errors

Real-Life Example: CSV Pipeline with Encoding Validation

This pipeline safely reads CSV files from multiple sources, handles encoding detection, logs any suspicious files, and outputs a clean UTF-8 combined CSV:

# csv_pipeline.py
import chardet
import csv
import io
from pathlib import Path
from dataclasses import dataclass

@dataclass
class FileReport:
    path: str
    encoding: str
    confidence: float
    row_count: int
    warning: str = ''

def process_csv_folder(input_folder: str, output_file: str) -> list:
    reports = []
    all_rows = []
    headers = None

    for csv_path in Path(input_folder).glob('*.csv'):
        raw = csv_path.read_bytes()
        result = chardet.detect(raw)
        encoding = result['encoding'] or 'utf-8'
        confidence = result['confidence']

        warning = ''
        if confidence < 0.8:
            warning = f'Low confidence ({confidence:.0%})'

        try:
            text = raw.decode(encoding, errors='replace')
            reader = csv.DictReader(io.StringIO(text))
            rows = list(reader)

            if headers is None and rows:
                headers = reader.fieldnames

            all_rows.extend(rows)
            reports.append(FileReport(
                path=csv_path.name,
                encoding=encoding,
                confidence=confidence,
                row_count=len(rows),
                warning=warning
            ))
        except Exception as e:
            reports.append(FileReport(
                path=csv_path.name,
                encoding=encoding,
                confidence=confidence,
                row_count=0,
                warning=f'Parse error: {e}'
            ))

    # Write combined output as UTF-8
    if all_rows and headers:
        with open(output_file, 'w', encoding='utf-8', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=headers)
            writer.writeheader()
            writer.writerows(all_rows)

    return reports

reports = process_csv_folder('./input_csvs', 'combined_output.csv')
for r in reports:
    status = f"WARNING: {r.warning}" if r.warning else "OK"
    print(f"{r.path}: {r.encoding} ({r.confidence:.0%}) | {r.row_count} rows | {status}")

Output:

sales_q1.csv: Windows-1252 (96%) | 142 rows | OK
customers.csv: UTF-8 (99%) | 89 rows | OK
legacy_data.csv: ISO-8859-1 (73%) | 31 rows | WARNING: Low confidence (73%)

Frequently Asked Questions

What if chardet detects the wrong encoding?

chardet is statistical, not deterministic -- it can be wrong, especially for short strings or files with predominantly ASCII content (since ASCII is a subset of many encodings, chardet cannot distinguish them). If you know the likely source of your files (e.g., all from a Windows 10 system in Germany), you can use the detected encoding as a hint and fall back to cp1252 or iso-8859-1 if the result looks wrong. Always validate the decoded output visually for critical data.

What confidence threshold should I use?

A confidence above 0.9 is generally reliable for common western encodings. For UTF-8 and ASCII, you typically see 0.99+. For ambiguous encodings like Windows-1252 vs latin-1 (which share most of their byte space), confidence may be 0.7-0.85. For production pipelines, log any detection below 0.85 and manually review those files. For interactive tools, warn the user and offer to retry with a different encoding.

Why does chardet fail on very short strings?

Encoding detection is statistical -- it needs enough bytes to observe characteristic patterns. A 10-byte string like a name or short code may not contain enough distinguishing bytes. As a rule of thumb, chardet needs at least 100-200 bytes for reliable detection, and 1KB+ for high confidence with multi-byte encodings. For short strings, you are better off knowing the source system's encoding and hardcoding it.

What is a BOM and should I remove it?

A Byte Order Mark (BOM) is a special marker at the start of a file that identifies the encoding. UTF-8-SIG files have a 3-byte BOM that Excel adds when saving CSV files. Python's utf-8-sig codec handles this automatically -- it strips the BOM on read and adds it on write. If you use the plain utf-8 codec, the BOM shows up as the character  at the start of your string, which can cause problems when parsing CSV files (the first column header gets a weird prefix).

Is there a faster alternative to chardet?

Yes -- charset-normalizer is a modern, actively maintained alternative that is often more accurate and faster than chardet for modern encodings. It is included as a dependency of the requests library. The API is compatible: from charset_normalizer import detect. For new projects, consider charset-normalizer over chardet, as chardet has been less actively maintained in recent years.

Conclusion

You now know how to use chardet to handle the encoding chaos that comes with real-world data. We covered basic detection with chardet.detect(), opening files safely with auto-detected encodings, streaming detection with UniversalDetector for large files, bulk conversion of legacy file directories to UTF-8, and a complete CSV pipeline that reports and handles encoding issues gracefully.

The most important takeaway: always read files as raw bytes (open(path, 'rb')) before deciding the encoding, never assume UTF-8 for data you did not create yourself, and treat low confidence scores as a signal to inspect the output manually. Encoding bugs are silent -- a file that "opens fine" but has garbled special characters is worse than one that crashes with a clear error.

See the chardet documentation for the full list of supported encodings and language detection capabilities.

Related Articles

• How To Use Python memray for Memory Profiling
• How To Use Python beartype for Runtime Type Checking
• How To Use Python nox for Automated Test Sessions
• How To Use Python Typer and Click for Building Complex CLIs
• How To Use Python Enums and When You Should
• How To Use Python Click for CLI Applications
• How To Use Python string Module for Templates and Constants

Intermediate

JSON serialization is in the hot path of almost every Python web service: every API response encodes data to JSON, every incoming request decodes it. If your service handles thousands of requests per second, JSON encoding time adds up fast. Python’s built-in json module is correct and convenient, but it was not built for speed — and libraries like orjson are fast but handle only basic Python types. What if you want both speed and type safety?

msgspec is a high-performance serialization library that encodes and decodes JSON (and MessagePack) 5-10x faster than the standard library while also providing automatic type validation. You define your data shape once using msgspec.Struct — like a Pydantic model but with a smaller memory footprint and faster instantiation — and msgspec handles encoding, decoding, and validation in a single C-extension call.

This article covers installing msgspec, defining Struct classes, encoding and decoding JSON, using type annotations for automatic validation, handling optional and nested fields, working with MessagePack, and benchmarking against the standard library. By the end you will have a complete toolkit for high-performance, type-safe JSON serialization in Python.

Written by Pubs

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

View all tutorials by Pubs →

msgspec Quick Example

# quick_msgspec.py
import msgspec
import msgspec.json

class User(msgspec.Struct):
    name: str
    email: str
    age: int
    is_active: bool = True

# Encode to JSON bytes
user = User(name="Alice", email="alice@example.com", age=30)
encoded = msgspec.json.encode(user)
print(encoded)
print(type(encoded))

# Decode from JSON bytes -- returns a User instance with validation
raw = b'{"name":"Bob","email":"bob@example.com","age":25}'
decoded = msgspec.json.decode(raw, type=User)
print(decoded)
print(type(decoded))

b'{"name":"Alice","email":"alice@example.com","age":30,"is_active":true}'
<class 'bytes'>
User(name='Bob', email='bob@example.com', age=25, is_active=True)
<class '__main__.User'>

The encode() function serializes a Struct (or any supported Python type) to JSON bytes. The decode() function deserializes JSON bytes and validates them against a type — if the JSON does not match the expected shape, a ValidationError is raised. Unlike the stdlib json module, the output is bytes, not str, which is what HTTP servers and most network libraries expect anyway.

What Is msgspec and When Should You Use It?

msgspec is a C-extension library by Jim Crist-Harif that provides both serialization performance and type safety. It is designed as a faster, lower-footprint alternative to Pydantic for use cases where you need validated deserialization at high throughput.

Use msgspec when you need high-throughput JSON encoding/decoding with type safety and low memory usage — particularly in web APIs (FastAPI, Starlette), event streaming, and data pipeline code where JSON parsing is in the hot path. Use Pydantic when you need rich validators, field aliases, custom serializers, or deep ecosystem integration. Use orjson when you need maximum speed with no schema requirements.

Installation

pip install msgspec

Successfully installed msgspec-0.18.6

msgspec ships as a pre-compiled C extension for Linux, macOS, and Windows on both CPython and PyPy. Import with import msgspec for Struct definitions and import msgspec.json for JSON operations.

Defining Structs

A msgspec.Struct is a fast, memory-efficient data class. It uses Python type annotations to define its fields and generates optimized __init__, __repr__, __eq__, and encoder/decoder hooks automatically.

# struct_basics.py
import msgspec
from typing import Optional, List
from datetime import datetime

class Address(msgspec.Struct):
    street: str
    city: str
    country: str
    postal_code: str = ""          # Default value

class Order(msgspec.Struct):
    order_id: str
    customer_name: str
    items: List[str]
    total: float
    shipping_address: Address       # Nested Struct
    created_at: datetime            # datetime is supported natively
    notes: Optional[str] = None    # Optional field

# Create instances
addr = Address(street="123 Main St", city="Austin", country="US", postal_code="78701")
order = Order(
    order_id="ORD-001",
    customer_name="Alice Smith",
    items=["Widget A", "Gadget B"],
    total=89.99,
    shipping_address=addr,
    created_at=datetime(2026, 4, 30, 10, 0, 0),
)

print(order)
print()
print(f"Order ID: {order.order_id}")
print(f"City: {order.shipping_address.city}")
print(f"Notes: {order.notes}")   # None -- the default

Order(order_id='ORD-001', customer_name='Alice Smith', items=['Widget A', 'Gadget B'], total=89.99, shipping_address=Address(street='123 Main St', city='Austin', country='US', postal_code='78701'), created_at=datetime.datetime(2026, 4, 30, 10, 0), notes=None)

Order ID: ORD-001
City: Austin
Notes: None

Structs are immutable by default (frozen). They have no __dict__, which makes them significantly more memory-efficient than regular Python objects or dataclasses. Access fields as attributes. Nested Structs, lists, dicts, and Python’s standard types (datetime, UUID, Decimal) are all supported as field types.

Encoding: Struct to JSON

Use msgspec.json.encode() to serialize any Struct or supported Python value to JSON bytes:

# encoding.py
import msgspec
import msgspec.json
from datetime import datetime

class Product(msgspec.Struct):
    id: int
    name: str
    price: float
    in_stock: bool
    tags: list

product = Product(id=1, name="Widget Pro", price=29.99,
                  in_stock=True, tags=["electronics", "gadgets"])

# encode returns bytes
json_bytes = msgspec.json.encode(product)
print(json_bytes)
print(type(json_bytes))

# Decode bytes to string if needed
json_str = json_bytes.decode("utf-8")
print(json_str)

# Encode a plain Python dict (msgspec handles these too)
data = {"key": "value", "num": 42, "flag": True}
print(msgspec.json.encode(data))

# Encode a list of Structs
products = [
    Product(id=i, name=f"Product {i}", price=9.99 * i,
            in_stock=i % 2 == 0, tags=[])
    for i in range(1, 4)
]
print(msgspec.json.encode(products))

b'{"id":1,"name":"Widget Pro","price":29.99,"in_stock":true,"tags":["electronics","gadgets"]}'
<class 'bytes'>
{"id":1,"name":"Widget Pro","price":29.99,"in_stock":true,"tags":["electronics","gadgets"]}
b'{"key":"value","num":42,"flag":true}'
b'[{"id":1,...},{"id":2,...},{"id":3,...}]'

Field names in the JSON output match the Struct attribute names exactly. If you need different JSON field names (e.g., camelCase in JSON, snake_case in Python), use msgspec.Struct with the rename option or rename individual fields with msgspec.field(name="camelCaseName").

Decoding with Validation

The decode() function deserializes JSON and validates it against your type annotation simultaneously. Invalid data raises a descriptive msgspec.ValidationError:

# decoding_validation.py
import msgspec
import msgspec.json
from typing import Optional, List

class UserProfile(msgspec.Struct):
    user_id: int
    username: str
    email: str
    score: float
    tags: List[str]
    bio: Optional[str] = None

# Decode valid JSON
valid_json = b'''
{
  "user_id": 42,
  "username": "alice",
  "email": "alice@example.com",
  "score": 9.5,
  "tags": ["python", "developer"],
  "bio": "Python enthusiast"
}
'''

profile = msgspec.json.decode(valid_json, type=UserProfile)
print(f"User: {profile.username}, Score: {profile.score}")
print(f"Tags: {profile.tags}")
print()

# Decode with missing optional field -- OK
minimal = b'{"user_id":1,"username":"bob","email":"b@b.com","score":7.0,"tags":[]}'
p2 = msgspec.json.decode(minimal, type=UserProfile)
print(f"Bio (optional): {p2.bio}")   # None
print()

# Decode with wrong type -- raises ValidationError
try:
    bad = b'{"user_id":"not_an_int","username":"x","email":"x@x.com","score":1.0,"tags":[]}'
    msgspec.json.decode(bad, type=UserProfile)
except msgspec.ValidationError as e:
    print(f"Validation error: {e}")

# Decode with missing required field -- raises ValidationError
try:
    missing = b'{"user_id":1,"username":"alice"}'
    msgspec.json.decode(missing, type=UserProfile)
except msgspec.ValidationError as e:
    print(f"Missing field error: {e}")

User: alice, Score: 9.5
Tags: ['python', 'developer']

Bio (optional): None

Validation error: Expected `int`, got `str` - at `$.user_id`
Missing field error: Object missing required field `email` - at `$`

The validation error messages include a JSON path ($.user_id) that tells you exactly which field failed and why. This replaces the typical pattern of calling json.loads() and then manually checking types — msgspec does both in one step, at C speed. The path notation ($ for the root, $.field for a specific field, $.list[0] for a list element) makes debugging API input validation errors straightforward.

Reusable Encoder and Decoder Objects

For high-throughput code, create Encoder and Decoder objects once and reuse them. This avoids per-call setup overhead and enables encoder customization:

# encoder_decoder.py
import msgspec
import msgspec.json

class Event(msgspec.Struct):
    event_type: str
    payload: dict
    timestamp: float

# Create once at module level -- reuse for every request
encoder = msgspec.json.Encoder()
decoder = msgspec.json.Decoder(Event)

# Encode
event = Event(event_type="user_login", payload={"user_id": 42}, timestamp=1714483200.0)
data = encoder.encode(event)
print(data)

# Decode
raw = b'{"event_type":"purchase","payload":{"order_id":"123"},"timestamp":1714483260.0}'
decoded = decoder.decode(raw)
print(decoded)
print(f"Type: {decoded.event_type}, Order: {decoded.payload.get('order_id')}")

b'{"event_type":"user_login","payload":{"user_id":42},"timestamp":1714483200.0}'
Event(event_type='purchase', payload={'order_id': '123'}, timestamp=1714483260.0)
Type: purchase, Order: 123

Reusable encoder/decoder objects are the pattern to use in web servers (FastAPI, Flask) where the same type is encoded or decoded on every request. Create them at module level (outside of request handlers) so the per-type setup cost is paid once at startup, not on every request.

Real-Life Example: FastAPI Response Serialization

Here is how to use msgspec for high-performance JSON responses in a FastAPI application:

# fastapi_msgspec.py
# Install: pip install fastapi uvicorn msgspec
from fastapi import FastAPI
from fastapi.responses import Response
import msgspec
import msgspec.json
from typing import List
from datetime import datetime

app = FastAPI()

class Product(msgspec.Struct):
    id: int
    name: str
    price: float
    in_stock: bool
    updated_at: datetime

# Prebuilt encoder -- created once at module load time
encoder = msgspec.json.Encoder()

# Sample data (in a real app, this comes from a database)
PRODUCTS = [
    Product(id=i, name=f"Product {i}", price=round(9.99 * i, 2),
            in_stock=i % 3 != 0, updated_at=datetime(2026, 4, 30))
    for i in range(1, 101)
]

@app.get("/products")
def list_products() -> Response:
    """Return all products as JSON -- using msgspec for fast encoding."""
    return Response(
        content=encoder.encode(PRODUCTS),
        media_type="application/json"
    )

@app.get("/products/{product_id}")
def get_product(product_id: int) -> Response:
    """Return a single product by ID."""
    product = next((p for p in PRODUCTS if p.id == product_id), None)
    if product is None:
        return Response(content=b'{"error":"not found"}',
                        media_type="application/json", status_code=404)
    return Response(
        content=encoder.encode(product),
        media_type="application/json"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# Run with: python fastapi_msgspec.py
# GET /products returns 100 products as JSON
# GET /products/1 returns a single product

By bypassing FastAPI’s built-in JSON serialization (which uses the stdlib json module via Pydantic) and using a msgspec Encoder directly, you get 5-10x faster serialization for response bodies. The key pattern: return fastapi.responses.Response with pre-encoded bytes instead of returning a Python dict and letting FastAPI encode it. This is the approach used by high-traffic Python APIs to shave milliseconds off every response.

Frequently Asked Questions

When should I use msgspec instead of Pydantic?

Use msgspec when performance is the primary concern and you do not need Pydantic’s ecosystem features: validators on individual fields, aliases, computed fields, custom serializers, or deep framework integration (Django, SQLAlchemy). msgspec Structs are faster to create, use less memory, and encode/decode faster than Pydantic models. Use Pydantic when you need its rich validation API or when your framework requires it (FastAPI’s dependency injection uses Pydantic models natively).

What is MessagePack and when should I use it instead of JSON?

MessagePack is a binary serialization format that is more compact and faster to parse than JSON. Use msgspec.msgpack.encode() and msgspec.msgpack.decode() for internal service-to-service communication where both sides speak Python. MessagePack is about 30% smaller than equivalent JSON for typical data and faster to encode/decode. Stick with JSON when the data needs to be human-readable, logged, or consumed by a non-msgspec client.

How do I make a Struct mutable?

Pass frozen=False to the Struct class: class MyStruct(msgspec.Struct, frozen=False). By default, Structs are frozen (immutable). Mutable Structs allow field assignment (obj.field = value) and can be used where you need to update individual fields after construction. Frozen Structs are hashable (can be used as dict keys or in sets); mutable Structs are not.

Does msgspec support Struct inheritance?

Yes, with limitations. A Struct subclass inherits all parent fields. However, a subclass cannot override parent fields or change their types. Struct inheritance is useful for adding fields to a base type without redefining everything, but it is not as flexible as Python class inheritance. For polymorphic types, use typing.Union with a tag field to distinguish subtypes during decoding.

msgspec encodes to bytes, but I need a str. How do I convert?

Call .decode() on the bytes: json_str = msgspec.json.encode(obj).decode("utf-8"). This adds one small allocation, but the total is still faster than json.dumps(). Most HTTP servers (WSGI/ASGI) accept bytes directly in the response body, so the conversion is often unnecessary in practice.

Conclusion

msgspec gives you the best of both worlds: the speed of a C-extension JSON library and the type safety of a schema validation library. You have seen how to define Struct classes with type annotations, encode and decode JSON with automatic validation, use reusable Encoder/Decoder objects for high-throughput code, handle optional and nested fields, and wire msgspec into FastAPI for fast API responses. The official documentation at jcristharif.com/msgspec covers advanced topics including custom hooks, YAML/TOML support, and the full type system.

Related Articles

• How To Use Python Ruff for Fast Linting and Formatting
• How To Use uv: The Fast Python Package Manager
• How To Use Python beartype for Runtime Type Checking
• How To Use Python nox for Automated Test Sessions
• How To Use Python Typer and Click for Building Complex CLIs
• How To Use Python Enums and When You Should
• How To Use Python cachetools for Flexible Caching Strategies