Have you ever found yourself staring at a cryptic `KeyError` or `TypeError` in your data processing pipeline, wondering why the API data that worked yesterday suddenly broke today? The issue often isn't the API itself, but the implicit data contract we assume when fetching and parsing JSON payloads. For Python developers and data scientists managing high-stakes data flows, ensuring data consistency and improving code clarity is crucial. This post will show you how to move beyond basic type hints and leverage Python's `TypedDict` and `Protocol` to enforce both data structure and behavior at scale, using the JSONPlaceholder Todos API as a real-world example.
Key Takeaways
- `TypedDict` allows you to define explicit, statically checked schemas for dictionary-like data, preventing common runtime `KeyError` and `TypeError` bugs.
- Combine `TypedDict` with runtime validation to ensure data consistency and catch unexpected format variations.
- Use `Protocol` to create flexible, type-checked behavioral contracts for data processing, making your code more modular and maintainable.
The Problem
In high-stakes production environments, relying on implicit data contracts from external APIs often leads to runtime errors, difficult-to-debug issues, and brittle code that breaks with minor API changes. This post is for Python developers and data scientists who manage data pipelines consuming external JSON APIs and are struggling with ensuring data consistency, improving code clarity, and making their systems resilient to unexpected data format variations without sacrificing flexibility.
Data and Sources
We'll be using the JSONPlaceholder Todos API, accessible at https://jsonplaceholder.typicode.com/todos, as our real-world example. Data accessed on 2026-08-14. For more information on the API and its usage, please refer to the official documentation.
Loading the Data
To start, we need to fetch the data from the JSONPlaceholder Todos API. We'll use the `requests` library to send a GET request and parse the response as JSON.
import requests
response = requests.get("https://jsonplaceholder.typicode.com/todos")
data = response.json()
Defining the Data Schema with `TypedDict`
Next, we define a `TypedDict` to specify the expected structure of the API response. This includes the `userId`, `id`, `title`, and `completed` fields.
from typing import TypedDict
class TodoItem(TypedDict):
userId: int
id: int
title: str
completed: bool
Validating the Data
To ensure data consistency, we'll implement runtime validation using the `isinstance` function to check if each item in the response conforms to our `TodoItem` schema.
def validate_todo_item(item: TodoItem) -> bool:
return isinstance(item["userId"], int) and isinstance(item["id"], int) and isinstance(item["title"], str) and isinstance(item["completed"], bool)
Crafting Behavioral Contracts with `Protocol`
We'll define a `Protocol` to specify the expected behavior of our data processing functions. This includes a `process` method that takes a `TodoItem` as input and returns a processed result.
from typing import Protocol
class TodoProcessor(Protocol):
def process(self, item: TodoItem) -> str:
...
Implementing Flexible Processors
Now, we'll create concrete implementations of our `TodoProcessor` protocol. For example, a `SimpleStatusProcessor` that returns a simple status message, and a `PriorityFormatterProcessor` that formats the output based on priority.
class SimpleStatusProcessor:
def process(self, item: TodoItem) -> str:
return f"Todo {item['id']} is {'completed' if item['completed'] else 'not completed'}"
class PriorityFormatterProcessor:
def process(self, item: TodoItem) -> str:
priority = "High" if item["userId"] == 1 else "Low"
return f"Todo {item['id']} ({priority}) is {'completed' if item['completed'] else 'not completed'}"
Orchestrating Type-Safe Data Flows
Finally, we'll put everything together, fetching the data, validating it, and processing it using our type-safe processors.
if __name__ == "__main__":
data = requests.get("https://jsonplaceholder.typicode.com/todos").json()
for item in data:
if validate_todo_item(item):
simple_processor = SimpleStatusProcessor()
priority_processor = PriorityFormatterProcessor()
print(simple_processor.process(item))
print(priority_processor.process(item))
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import requests
from typing import TypedDict, Protocol
class TodoItem(TypedDict):
userId: int
id: int
title: str
completed: bool
def validate_todo_item(item: TodoItem) -> bool:
return isinstance(item["userId"], int) and isinstance(item["id"], int) and isinstance(item["title"], str) and isinstance(item["completed"], bool)
class TodoProcessor(Protocol):
def process(self, item: TodoItem) -> str:
...
class SimpleStatusProcessor:
def process(self, item: TodoItem) -> str:
return f"Todo {item['id']} is {'completed' if item['completed'] else 'not completed'}"
class PriorityFormatterProcessor:
def process(self, item: TodoItem) -> str:
priority = "High" if item["userId"] == 1 else "Low"
return f"Todo {item['id']} ({priority}) is {'completed' if item['completed'] else 'not completed'}"
if __name__ == "__main__":
data = requests.get("https://jsonplaceholder.typicode.com/todos").json()
for item in data:
if validate_todo_item(item):
simple_processor = SimpleStatusProcessor()
priority_processor = PriorityFormatterProcessor()
print(simple_processor.process(item))
print(priority_processor.process(item))
Expected Output
When you run the script, you should see the processed output for each todo item using both the `SimpleStatusProcessor` and `PriorityFormatterProcessor`.
Limitations and Tradeoffs
This approach assumes that the API response conforms to the expected schema. In cases where the schema is highly dynamic or unpredictable, additional validation and error handling may be necessary. Furthermore, the use of `Protocol` and `TypedDict` requires Python 3.8 or later, which may not be compatible with all production environments.
Frequently Asked Questions
What is the difference between `TypedDict` and `dataclass`?
`TypedDict` is used for defining the structure of dictionary-like data, while `dataclass` is used for defining classes that contain data. In this example, we use `TypedDict` to specify the expected structure of the API response.
How do I handle cases where the API response is missing expected fields?
You can use the `validate_todo_item` function to check if the response conforms to the expected schema. If a field is missing, the function will return `False`, and you can handle the error accordingly.
Can I use this approach with other types of data, such as CSV or XML?
Yes, you can use this approach with other types of data. However, you may need to modify the `validate_todo_item` function to accommodate the specific data format.
What I'd Change
In conclusion, leveraging `TypedDict` and `Protocol` is a powerful way to build resilient pipelines that can handle external API data with confidence. However, I would change the approach to include additional error handling and validation for cases where the API response is highly dynamic or unpredictable. By doing so, you can ensure that your system is robust and maintainable, even in the face of unexpected data format variations.
