Home
python
typing

How to Achieve Type‑Safe Todo Processing with Protocols and TypedDict

Mausam

When you pull task data from a public JSON API, you often end up with loose dictionaries that mypy can’t verify, so a subtle schema change can explode your pipeline at runtime. By pairing a `TypedDict` that mirrors the payload with a `Protocol` that defines the processing contract, you get compile‑time guarantees for your processors and a lightweight runtime guard that rejects malformed records before they cause trouble.

Key Takeaways

  • `TypedDict` provides a precise static description of JSON objects without sacrificing the native dict interface.
  • `Protocol` enables duck‑typed, interchangeable processing classes that mypy can validate.
  • Layering a small runtime validator on top of `TypedDict` catches malformed data early, keeping production pipelines robust.
  • The pattern scales to larger APIs and fits naturally into plugin architectures like the RSS handler system you read about earlier.

Data and Sources

Data accessed on 2026-11-11

Using Protocols to Abstract Todo Processing

First, define the contract that any Todo processor must satisfy. The @runtime_checkable decorator lets us verify at runtime that an object adheres to the protocol, which is handy for plugin discovery.

from typing import Protocol
from typing_extensions import runtime_checkable

@runtime_checkable
class TodoProcessor(Protocol):
    def process(self, todo: "TodoDict") -> None: ...

Two concrete implementations illustrate the flexibility:

class PrintProcessor:
    def process(self, todo: "TodoDict") -> None:
        print(f"[{todo['id']}] {todo['title']} (completed={todo['completed']})")


class StatsProcessor:
    def __init__(self):
        self.total = 0
        self.completed = 0
        self.per_user = {}

    def process(self, todo: "TodoDict") -> None:
        self.total += 1
        if todo["completed"]:
            self.completed += 1
        user = todo["userId"]
        self.per_user[user] = self.per_user.get(user, 0) + 1

    def report(self) -> None:
        print(f"Total todos: {self.total}")
        if self.total:
            print(f"Completed ratio: {self.completed / self.total:.2%}")
        print("Tasks per user:")
        for user, count in sorted(self.per_user.items()):
            print(f"  User {user}: {count}")

Running mypy on this file will confirm that both classes satisfy TodoProcessor without any inheritance relationship, proving the static safety of the duck‑typed approach.

TypedDict for Structured Todo Payloads

Next, describe the exact shape of the JSON objects you expect from the API.

from typing_extensions import TypedDict

class TodoDict(TypedDict):
    userId: int
    id: int
    title: str
    completed: bool

Static type checkers now know the exact keys and value types. To enforce the same contract at runtime, add a tiny validator that checks presence and basic types before handing the dict to a processor.

def validate_todo(item: dict) -> TodoDict:
    required = {"userId", "id", "title", "completed"}
    missing = required - item.keys()
    if missing:
        raise ValueError(f"Missing keys: {missing}")

    if not isinstance(item["userId"], int):
        raise TypeError("userId must be int")
    if not isinstance(item["id"], int):
        raise TypeError("id must be int")
    if not isinstance(item["title"], str):
        raise TypeError("title must be str")
    if not isinstance(item["completed"], bool):
        raise TypeError("completed must be bool")

    return item  # type: ignore[arg-type]

The validator is deliberately lightweight: it avoids full schema libraries while still protecting against the most common data‑corruption scenarios (missing keys, wrong primitive types).

Limitations and Tradeoffs

  • Runtime enforcement cost: TypedDict itself does not validate data; you must add a validator (as shown) or a third‑party library, which introduces a small import‑time and per‑item overhead.
  • Value constraints are absent: Protocols and TypedDict can express types but not domain rules (e.g., userId > 0). For richer validation you would layer Pydantic or attrs on top.
  • Schema evolution: Adding optional fields to the API is safe—our validator only checks required keys. Removing or renaming required fields will raise errors, which is desirable because it forces a code change.
  • Dynamic schemas: If the payload structure is highly fluid (e.g., arbitrary key/value pairs), the static approach becomes cumbersome and a generic dict with runtime checks may be preferable.

Frequently Asked Questions

Do I need typing‑extensions on Python 3.11?

No. Python 3.11 includes Protocol in the standard typing module. You only need typing‑extensions for older versions (3.8‑3.10) or for the @runtime_checkable decorator.

How does this differ from using dataclasses?

dataclasses create concrete classes that you must instantiate, which adds an extra conversion step from the raw JSON dict. TypedDict preserves the native dictionary interface, letting you pass the payload straight to downstream code while still giving mypy a clear contract.

Can Protocols be used with async processors?

Yes. The protocol only cares about the method signature. An async implementation would declare async def process(self, todo: TodoDict) -> None and the protocol could be updated to return Awaitable[None]. The rest of the pipeline stays unchanged.

What if the API adds extra fields?

Extra fields are ignored by the validator because it only checks the required keys. The resulting TodoDict will still be a regular dict containing the extra keys, so existing processors continue to work unchanged.

Is mypy enough to catch all errors?

mypy guarantees that the static types line up, but it cannot detect malformed JSON at runtime. That’s why the explicit validate_todo step is essential for production‑grade pipelines.

Recommendation and Next Step

Adopt the Protocol + TypedDict pattern for every external JSON integration you write. It gives you compile‑time confidence, interchangeable processor implementations, and a single place to inject runtime validation. As a next step, replace the manual validator with a Pydantic model to gain automatic value constraints and richer error messages without changing the surrounding protocol‑based architecture.

Complete Script

#!/usr/bin/env python3
"""
Type‑safe Todo processing demo.

- Fetches the public JSONPlaceholder /todos endpoint.
- Validates each payload against a TypedDict.
- Routes validated items through a Protocol‑based processor.
- Prints a concise statistics report.

Dependencies:
    requests>=2.31
    typing-extensions>=4.0
"""

from __future__ import annotations

import sys
from typing import Dict, List

import requests
from typing_extensions import TypedDict, Protocol, runtime_checkable


class TodoDict(TypedDict):
    userId: int
    id: int
    title: str
    completed: bool


def validate_todo(item: dict) -> TodoDict:
    """Raise if the dict does not conform to TodoDict."""
    required = {"userId", "id", "title", "completed"}
    missing = required - item.keys()
    if missing:
        raise ValueError(f"Missing keys: {missing}")

    if not isinstance(item["userId"], int):
        raise TypeError("userId must be int")
    if not isinstance(item["id"], int):
        raise TypeError("id must be int")
    if not isinstance(item["title"], str):
        raise TypeError("title must be str")
    if not isinstance(item["completed"], bool):
        raise TypeError("completed must be bool")

    return item  # type: ignore[arg-type]


@runtime_checkable
class TodoProcessor(Protocol):
    def process(self, todo: TodoDict) -> None: ...


class StatsProcessor:
    """Collects simple statistics about the todo stream."""

    def __init__(self) -> None:
        self.total: int = 0
        self.completed: int = 0
        self.per_user: Dict[int, int] = {}

    def process(self, todo: TodoDict) -> None:
        self.total += 1
        if todo["completed"]:
            self.completed += 1
        user = todo["userId"]
        self.per_user[user] = self.per_user.get(user, 0) + 1

    def report(self) -> None:
        print(f"Total todos: {self.total}")
        if self.total:
            print(f"Completed ratio: {self.completed / self.total:.2%}")
        print("Tasks per user:")
        for user, count in sorted(self.per_user.items()):
            print(f"  User {user}: {count}")


def fetch_todos(url: str) -> List[dict]:
    """Retrieve the raw JSON list, handling network errors."""
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
    except requests.RequestException as exc:
        raise RuntimeError(f"Failed to fetch todos: {exc}") from exc

    data = response.json()
    if not isinstance(data, list):
        raise ValueError("Expected a list of todo objects")
    return data


def main() -> None:
    url = "https://jsonplaceholder.typicode.com/todos"
    try:
        raw_items = fetch_todos(url)
    except RuntimeError as e:
        print(e, file=sys.stderr)
        sys.exit(1)

    processor: TodoProcessor = StatsProcessor()

    for item in raw_items:
        try:
            todo = validate_todo(item)
        except (ValueError, TypeError) as err:
            print(f"Skipping invalid item: {err}", file=sys.stderr)
            continue
        processor.process(todo)

    # The concrete StatsProcessor knows how to report.
    if isinstance(processor, StatsProcessor):
        processor.report()


if __name__ == "__main__":
    main()

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.