Your service reads events from a queue as JSON, drops duplicates, and writes each event to a database as a row. The same few fields show up in three places, and each place wants a different shape. The standard library has a record type for each.
from dataclasses import dataclass
from typing import NamedTuple, TypedDict
@dataclass(frozen=True)
class Event:
tenant_id: str
event_id: str
class EventKey(NamedTuple):
tenant_id: str
event_id: str
class EventRow(TypedDict):
tenant_id: str
event_id: str
event = Event("acme", "evt-1")
key = EventKey("acme", "evt-1")
row: EventRow = {"tenant_id": "acme", "event_id": "evt-1"}
print(event.tenant_id, key.tenant_id, row["tenant_id"]) # acme acme acme
print(key == ("acme", "evt-1")) # True
print(type(row)) # <class 'dict'>
What each type is
- A dataclass is an ordinary class. You read fields as attributes, add
methods, and choose options such as
frozen=True. Its==compares only objects of the same class. - A NamedTuple is a tuple whose positions have names.
key.tenant_idandkey[0]are the same value, andtenant, event_id = keyunpacks it. Like any tuple, it cannot change, and it equals any tuple with the same items. - A TypedDict describes a dictionary with known keys to the type
checker. When the program runs,
rowis a plaindict.
Pick the type by what the code that uses the record needs:
| The code needs | Use |
|---|---|
Attributes, methods, defaults or frozen=True |
a dataclass |
| Tuple unpacking, indexing, or a small read-only key | a NamedTuple |
| A dictionary with known keys, such as decoded JSON or a database row | a TypedDict |
Use each one where it fits
In the pipeline, the core code works with Event objects. Removing
duplicates needs a key, and the database layer wants a dictionary:
from dataclasses import dataclass, replace
from typing import NamedTuple, TypedDict
class EventKey(NamedTuple):
tenant_id: str
event_id: str
class EventRow(TypedDict):
tenant_id: str
event_id: str
enriched: bool
@dataclass(frozen=True, kw_only=True)
class Event:
tenant_id: str
event_id: str
enriched: bool = False
def dedup_key(self) -> EventKey:
return EventKey(self.tenant_id, self.event_id)
def to_row(self) -> EventRow:
return {
"tenant_id": self.tenant_id,
"event_id": self.event_id,
"enriched": self.enriched,
}
original = Event(tenant_id="acme", event_id="evt-1")
enriched = replace(original, enriched=True)
print(original == enriched) # False
print(original.dedup_key() == enriched.dedup_key()) # True
print(enriched.to_row())
# {'tenant_id': 'acme', 'event_id': 'evt-1', 'enriched': True}
The two events are not equal, because the generated __eq__ compares
every field, enriched included. So the code removes duplicates by
dedup_key(), which holds only the fields that identify an event.
kw_only=True makes callers name every argument, as in the exercise Fix a
Dataclass Hierarchy with Keyword-Only Fields.
Common mistakes
Expecting the type to check values. None of the three checks types
when the program runs. EventRow(tenant_id=1, event_id=2, enriched="no")
builds a dictionary with no error, and Event(tenant_id=1, event_id=2)
builds an event. mypy reports both calls in your own code. Data from
outside the program arrives unchecked, so check it once, where it comes
in. The exercise Parse Untrusted JSON into a Dataclass shows how.
Putting two NamedTuple key types in one dictionary. A NamedTuple compares as a plain tuple, so keys of different types can collide:
from typing import NamedTuple
class EventKey(NamedTuple):
tenant_id: str
event_id: str
class UserKey(NamedTuple):
tenant_id: str
user_id: str
seen = {EventKey("acme", "42"): "event"}
print(UserKey("acme", "42") in seen) # True
If two kinds of key share a collection, make them frozen dataclasses. With
the generated __eq__, objects of two different dataclasses are never
equal.
Check your understanding
Your code reads row: Row = json.loads(text), where Row is a TypedDict with count: int. The JSON is {"count": "3"}. What is row["count"]?
Show answer
Hide answer
The string "3". mypy accepts the line because json.loads returns Any,
which matches every type. When the program runs, nothing checks the value,
because a TypedDict is a plain dict. Check outside data before the rest
of the code relies on its types.
Going further
- Third-party libraries add what these three leave out.
attrswrites dataclass-style classes with converters and validators. Pydantic andmsgspecparse and check input at runtime, and their errors name the field that failed. typing.NotRequiredmarks aTypedDictkey that may be missing. Read such a key withrow.get("key"), becauserow["key"]raisesKeyErrorwhen the key is absent.- No record type is fastest at everything. If speed or memory matters, measure with your own data.
Further reading: Python typing, attrs initialization, Pydantic TypeAdapter and msgspec typed decoding.