Skip to content

Typed Attributes & Records: Between Dicts and Dataclasses

A stream processor lives on the JSON boundary: every input is a dict decoded from the wire, and every output and every state write goes back through json.dumps. Handled as bare dicts, that boundary leaks into everything — each read re-checks presence and re-parses timestamps, a datetime assigned three hops earlier blows up only when the record is finally serialized, and a field that silently became null surfaces as a KeyError in some consumer far from the code that dropped it.

The conventional fix is custom dataclasses, and most frameworks lean into it: register a class (plus serialization glue) per message shape and let the framework marshal it to and from the wire. In a pipeline, that convention turns into cruft fast — every transformation step produces a new intermediate shape (the input plus one enrichment field, a state record of two counters, a trimmed projection for the output topic), and each wants its own class definition even when it exists only between two hops.

The flechtwerk.attribute library is the middle ground between those two poles. Each field — not each shape — is declared exactly once, as a typed handle pairing a wire name with an explicit Codec[V], and validation moves to the write site. Records combine any subset of these handles as freely as dicts combine keys, so intermediate shapes need no class definitions at all — yet every field access is type-safe:

from datetime import datetime, timezone

from flechtwerk import Event
from flechtwerk.attribute import Attribute, DATETIME, LIST, STR

DEVICE = Attribute("device", STR)
LAST_SEEN = Attribute("last_seen", DATETIME)
TAGS = Attribute("tags", LIST(STR), optional=True)

event = Event({
    DEVICE: "sensor-1",
    LAST_SEEN: datetime(2026, 7, 12, 9, 30, tzinfo=timezone.utc),
    TAGS: ["a", "b"],
})

event[LAST_SEEN]   # datetime(2026, 7, 12, 9, 30, tzinfo=timezone.utc) — a real datetime
event.raw          # {'device': 'sensor-1', 'last_seen': '2026-07-12T09:30:00Z', 'tags': ['a', 'b']}

# Both of these raise at the write site, not at serialization time:
#   event[DEVICE] = 42     # expected str, got int
#   event[DEVICE] = None   # cannot assign None to required Attribute('device')

Event, State, and Config are Record subclasses — dict-like containers indexed by these handles rather than string keys. (Message is a frozen dataclass envelope carrying a key, a topic, a value, and an optional timestamp — key and value each accept any Payload: bytes, str, or Event.) The codec runs on every write, so the underlying .raw payload stays JSON-native by construction: wire encoding is a straight json.dumps(event.raw), decoding is a straight Event.wrap(raw), and nothing in between ever needs to re-validate.

JSON Is the Only Wire Format — For Now

JSON is currently the only supported wire format for Kafka messages, and it covers every Event, State, and Config. Support for other serialization protocols (Avro, Protobuf, and the like) is a possible future extension, but is not currently planned.

Required vs. Optional

A required attribute (the default) rejects None so a dropped value can't silently land as JSON null; declare fields where absence is legal with optional=True.

The read distinction is carried by the method, not the declaration:

  • event[LAST_SEEN] reads-or-raises.
  • event.get(TAGS) tolerates absence and returns V | None.

Codecs Compose

Codecs are built from atoms and constructors:

  • Atoms: STR, INT, BOOL, BYTES, DATE, FLOAT, DATETIME, TIME, ZONE_INFO, RECORD, ANY.
  • Constructors: LIST(V), SET(V), TUPLE(V), DICT(V).

Nest them freely — DICT(LIST(INT)), LIST(RECORD), and so on — and the whole tree is validated on every write.

BYTES is the one atom whose Python type is not JSON-native: it carries binary as RFC 4648 base64 (standard alphabet, padded), strictly — whitespace, stray characters and non-canonical trailing bits are all rejected rather than repaired, so the wire form and the value stay in step. Binary is never inferred: ANY still refuses bytes, so a blob reaches the wire through an explicit Attribute(name, BYTES) or not at all. Base64 costs 4/3 of the payload against Kafka's 1 MiB record ceiling — for a message that is a blob, send it as a bytes Payload instead and skip the framing.

ZONE_INFO carries a zoneinfo.ZoneInfo as its IANA key ("Europe/Berlin", "UTC"). Decoding is strict: an unknown key raises ValueError, and so does a key in the wrong case. ZoneInfo("europe/berlin") would load on a case-insensitive filesystem such as macOS and fail on Linux, so the codec rejects it everywhere. A fixed offset (datetime.timezone) is not a zone and does not encode. Like binary, a zone is never inferred: ANY refuses it, because its wire form would read back as a plain string. ZoneInfo reads the host's time zone database, falling back to the tzdata package — an image with neither rejects every key.

Resolving Name Conflicts

Codec names are short and upper-case on purpose — they match the ALL_CAPS style of attribute declarations. That is also the style of your own constants, so sooner or later an attribute wants a codec's name for itself. Reach the codec through its module instead:

from flechtwerk.attribute import Attribute, codec

DATE = Attribute("date", codec.DATE)
TIME = Attribute("time", codec.TIME)

or rename it on import:

from flechtwerk.attribute import Attribute
from flechtwerk.attribute.codec import DATE as DATE_CODEC

DATE = Attribute("date", DATE_CODEC)

flechtwerk.attribute.codec holds the complete catalogue — every atom and constructor above, the Codec type itself, and record_codec for Record subclasses. The flechtwerk.attribute package re-exports all of it, so both spellings name the very same objects and mix freely within one module. The one codec that lives elsewhere is ENCRYPTED, in flechtwerk.secrets — see Encrypted Secrets.

Spreading: Enrich Without Mutating

Record subclasses spread like plain dicts, so enrichment is always a copy-with-overrides — never an in-place edit:

later = datetime.now(timezone.utc)

# a new Event with one field overridden — the original `event` is untouched
enriched = Event({**event, LAST_SEEN: later})

# spread across records too: carry an incoming payload forward and add a field
out = Event({**msg.value, SEEN: seen})

This is why enrichment never has to mutate its input. {**record, NEW: value} builds a fresh record from the old one's fields plus your overrides, leaving the original untouched — so inside a transform (or a poll, or a relay) you derive the next output Event or the next State by spreading, rather than editing in place.

And it is a backstop, not merely a convention:

Parameters are read-only — mutations are ignored

The runner hands every stage hook a private copy of its mutable parameters — the running state, and, for extractors, the config — and it never reuses the single-use msg after the call returns. Mutating a passed parameter in place therefore has no effect; the change is silently discarded. The only way to emit output or change state is to yield, and the way to enrich is to spread into a fresh record.

And spreading is not an escape hatch around the type system: the spread runs through the same typed-write path, so every field — carried over or overridden — is codec-checked, and the result's .raw stays JSON-native by construction.

Why It Matters

The point isn't ceremony; it's that the boundary between "Python object graph" and "JSON on the wire" is enforced at assignment time, once per Attribute declaration, rather than re-derived on every serialize/deserialize — and without a dataclass per message shape. Fields are the unit of declaration; shapes stay free.