Skip to content

API Reference

The vocabulary of the loom — set down here exactly as the source declares it.

API Reference

Flechtwerk's public surface is small and settled in shape. Every entry below is generated directly from the source docstrings, so it never drifts from the code. The core vocabulary is Extractor, Transformer, Message, State, Event, Config, ConfigStore, the runtime handle Flechtwerk, and the typed-record handles of flechtwerk.attribute — with a few supporting types (Payload, IncomingMessage, InvalidMessageError, MqttBrokerConfig, CompressionType) documented alongside them below.

Stages

flechtwerk.Transformer

Bases: Stage, ABC

Event transformer (stateless or stateful).

Three ways to construct one:

  • Declaratively with the @transformer(...) decorator, which binds a transform function to its input topics — the decorated name becomes the stage::

    @transformer(input_topics=["my-topic"]) async def stage(msg, state): ...

  • Functionally with Transformer.of(...), the factory the decorator wraps, when the transform function must stay callable under its own name::

    stage = Transformer.of( input_topics=["my-topic"], transform=my_transform_fn, )

  • As a subclass for lifecycle management (HTTP clients, dedup instances)::

    class MyTransformer(Transformer): input_topics = ["my-topic"]

      async def __aenter__(self):
      self.http = httpx.AsyncClient()
      return self

  async def __aexit__(self, *exc_info):
      await self.http.aclose()

  async def transform(self, msg, state):
      ...

The Kafka consumer group ID (driving consumer group membership, transactional offset commits, and changelog topic naming) is set on Flechtwerk by the caller; stages don't carry it.

A transformer may additionally declare config_topics (see Stage) and look their entries up via configs (inherited from Stage, where its contract is documented) — a config table joined against the partitioned input stream. A record produced to a config topic is visible there no later than the next batch.

Source code in src/flechtwerk/transformer.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class Transformer(Stage, ABC):
    """Event transformer (stateless or stateful).

    Three ways to construct one:

    * Declaratively with the ``@transformer(...)`` decorator, which binds a
      transform function to its input topics — the decorated name becomes the
      stage::

          @transformer(input_topics=["my-topic"])
          async def stage(msg, state):
              ...

    * Functionally with ``Transformer.of(...)``, the factory the decorator
      wraps, when the transform function must stay callable under its own name::

          stage = Transformer.of(
              input_topics=["my-topic"],
              transform=my_transform_fn,
          )

    * As a subclass for lifecycle management (HTTP clients, dedup instances)::

          class MyTransformer(Transformer):
              input_topics = ["my-topic"]

              async def __aenter__(self):
                  self.http = httpx.AsyncClient()
                  return self

              async def __aexit__(self, *exc_info):
                  await self.http.aclose()

              async def transform(self, msg, state):
                  ...

    The Kafka consumer group ID (driving consumer group membership,
    transactional offset commits, and changelog topic naming) is set on
    `Flechtwerk` by the caller; stages don't carry it.

    A transformer may additionally declare ``config_topics`` (see `Stage`)
    and look their entries up via `configs` (inherited from `Stage`, where
    its contract is documented) — a config table joined against the
    partitioned input stream. A record produced to a config topic is
    visible there no later than the next batch.
    """

    input_topics: list[str]

    @classmethod
    def of(
            cls,
            *,
            input_topics: list[str],
            transform: TransformFn,
            enrich_config: EnrichConfigFn | None = None,
            extract_state_key: ExtractStateKeyFn | None = None,
            on_invalid_message: OnInvalidMessageFn | None = None,
    ) -> "Transformer":
        """Build a Transformer from a transform function and input topics.

        Use this for stateless or simply-stateful stages that don't need
        lifecycle management; subclass directly for stages that own resources
        (HTTP clients, dedup instances, etc.).

        Patches the supplied callables in as instance attributes that
        shadow the class-level abstract method ``transform`` (and, when
        provided, the default ``enrich_config`` / ``extract_state_key`` /
        ``on_invalid_message``). The ABC
        discipline still applies to every other construction path —
        ``Transformer()`` and any abstract subclass remain uninstantiable.
        """
        instance = _FunctionalTransformer()
        instance.input_topics = input_topics
        instance.transform = transform  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if enrich_config is not None:
            instance.enrich_config = enrich_config  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if extract_state_key is not None:
            instance.extract_state_key = extract_state_key  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if on_invalid_message is not None:
            instance.on_invalid_message = on_invalid_message  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        return instance

    @abstractmethod
    def transform(self, msg: IncomingMessage, state: State) -> AsyncIterator[Message | State]:
        """Transform an incoming message into zero or more output Messages.

        Declared without ``async`` so that implementations — ``async def``
        functions containing ``yield``, i.e. async generator functions whose
        call returns an ``AsyncIterator`` directly — are compatible overrides
        under strict type checking. A coroutine-typed abstract (``async def``
        with no ``yield``) would make every real override incompatible.

        Yield a State to signal the desired state. The runner persists it only
        if it differs from the current state. If no State is yielded, nothing
        is persisted (stateless behavior). Yielding an empty/falsy State deletes
        the entry from the state store (and writes a Kafka tombstone to the
        changelog) atomically with the output messages.

        ``msg.value`` is ``None`` for a tombstone (a null wire value) — a
        stage reading a compacted table sees deletions this way, and one that
        writes such a table deletes a row by yielding ``Message(value=None)``.

        Both parameters are read-only. The runner hands ``state`` a defensive
        copy and never reuses ``msg`` after the call, so mutating either in
        place has no effect — it is silently discarded. Produce output only by
        yielding, and enrich by spreading (``Event({**msg.value, ...})``) rather
        than mutating in place.
        """

of classmethod 

of(
    *,
    input_topics: list[str],
    transform: TransformFn,
    enrich_config: EnrichConfigFn | None = None,
    extract_state_key: ExtractStateKeyFn | None = None,
    on_invalid_message: OnInvalidMessageFn | None = None
) -> Transformer

Build a Transformer from a transform function and input topics.

Use this for stateless or simply-stateful stages that don't need lifecycle management; subclass directly for stages that own resources (HTTP clients, dedup instances, etc.).

Patches the supplied callables in as instance attributes that shadow the class-level abstract method transform (and, when provided, the default enrich_config / extract_state_key / on_invalid_message). The ABC discipline still applies to every other construction path — Transformer() and any abstract subclass remain uninstantiable.

Source code in src/flechtwerk/transformer.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@classmethod
def of(
        cls,
        *,
        input_topics: list[str],
        transform: TransformFn,
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | None = None,
        on_invalid_message: OnInvalidMessageFn | None = None,
) -> "Transformer":
    """Build a Transformer from a transform function and input topics.

    Use this for stateless or simply-stateful stages that don't need
    lifecycle management; subclass directly for stages that own resources
    (HTTP clients, dedup instances, etc.).

    Patches the supplied callables in as instance attributes that
    shadow the class-level abstract method ``transform`` (and, when
    provided, the default ``enrich_config`` / ``extract_state_key`` /
    ``on_invalid_message``). The ABC
    discipline still applies to every other construction path —
    ``Transformer()`` and any abstract subclass remain uninstantiable.
    """
    instance = _FunctionalTransformer()
    instance.input_topics = input_topics
    instance.transform = transform  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if enrich_config is not None:
        instance.enrich_config = enrich_config  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if extract_state_key is not None:
        instance.extract_state_key = extract_state_key  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if on_invalid_message is not None:
        instance.on_invalid_message = on_invalid_message  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    return instance

transform abstractmethod 

transform(
    msg: IncomingMessage, state: State
) -> AsyncIterator[Message | State]

Transform an incoming message into zero or more output Messages.

Declared without async so that implementations — async def functions containing yield, i.e. async generator functions whose call returns an AsyncIterator directly — are compatible overrides under strict type checking. A coroutine-typed abstract (async def with no yield) would make every real override incompatible.

Yield a State to signal the desired state. The runner persists it only if it differs from the current state. If no State is yielded, nothing is persisted (stateless behavior). Yielding an empty/falsy State deletes the entry from the state store (and writes a Kafka tombstone to the changelog) atomically with the output messages.

msg.value is None for a tombstone (a null wire value) — a stage reading a compacted table sees deletions this way, and one that writes such a table deletes a row by yielding Message(value=None).

Both parameters are read-only. The runner hands state a defensive copy and never reuses msg after the call, so mutating either in place has no effect — it is silently discarded. Produce output only by yielding, and enrich by spreading (Event({**msg.value, ...})) rather than mutating in place.

Source code in src/flechtwerk/transformer.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@abstractmethod
def transform(self, msg: IncomingMessage, state: State) -> AsyncIterator[Message | State]:
    """Transform an incoming message into zero or more output Messages.

    Declared without ``async`` so that implementations — ``async def``
    functions containing ``yield``, i.e. async generator functions whose
    call returns an ``AsyncIterator`` directly — are compatible overrides
    under strict type checking. A coroutine-typed abstract (``async def``
    with no ``yield``) would make every real override incompatible.

    Yield a State to signal the desired state. The runner persists it only
    if it differs from the current state. If no State is yielded, nothing
    is persisted (stateless behavior). Yielding an empty/falsy State deletes
    the entry from the state store (and writes a Kafka tombstone to the
    changelog) atomically with the output messages.

    ``msg.value`` is ``None`` for a tombstone (a null wire value) — a
    stage reading a compacted table sees deletions this way, and one that
    writes such a table deletes a row by yielding ``Message(value=None)``.

    Both parameters are read-only. The runner hands ``state`` a defensive
    copy and never reuses ``msg`` after the call, so mutating either in
    place has no effect — it is silently discarded. Produce output only by
    yielding, and enrich by spreading (``Event({**msg.value, ...})``) rather
    than mutating in place.
    """

flechtwerk.transformer.transformer 

transformer(
    *,
    input_topics: list[str],
    enrich_config: EnrichConfigFn | None = None,
    extract_state_key: ExtractStateKeyFn | None = None,
    on_invalid_message: OnInvalidMessageFn | None = None
) -> Callable[[TransformFn], Transformer]

Decorator form of Transformer.of — bind a transform function to its input topics.

The decorated async generator becomes the built Transformer, so the name you define is the stage, ready to hand to Flechtwerk.of::

@transformer(input_topics=["my-input"])
async def stage(msg: IncomingMessage, state: State) -> AsyncIterator[Message | State]:
    ...

enrich_config, extract_state_key, and on_invalid_message are the same optional overrides as on Transformer.of — this is exactly that call with transform supplied by the decoration. Call Transformer.of directly when the transform function must stay callable under its own name, and subclass Transformer when you need lifecycle management (__aenter__ / __aexit__).

Source code in src/flechtwerk/transformer.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def transformer(
        *,
        input_topics: list[str],
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | None = None,
        on_invalid_message: OnInvalidMessageFn | None = None,
) -> Callable[[TransformFn], Transformer]:
    """Decorator form of `Transformer.of` — bind a transform function to its input topics.

    The decorated async generator becomes the built `Transformer`, so the name
    you define *is* the stage, ready to hand to `Flechtwerk.of`::

        @transformer(input_topics=["my-input"])
        async def stage(msg: IncomingMessage, state: State) -> AsyncIterator[Message | State]:
            ...

    ``enrich_config``, ``extract_state_key``, and ``on_invalid_message`` are the
    same optional overrides as on
    `Transformer.of` — this is exactly that call with ``transform`` supplied by
    the decoration. Call `Transformer.of` directly when the transform function
    must stay callable under its own name, and subclass `Transformer` when you
    need lifecycle management (``__aenter__`` / ``__aexit__``).
    """
    def decorator(transform: TransformFn) -> Transformer:
        return Transformer.of(
            enrich_config=enrich_config,
            extract_state_key=extract_state_key,
            input_topics=input_topics,
            on_invalid_message=on_invalid_message,
            transform=transform,
        )
    return decorator

flechtwerk.Extractor

Bases: Stage, ABC

Poll-driven data extractor (stateful or stateless).

Three ways to construct one:

  • Declaratively with the @extractor(...) decorator, which binds a poll function to its config topics — the decorated name becomes the stage::

    @extractor(config_topics=["my-config"]) async def stage(config, state): ...

  • Functionally with Extractor.of(...), the factory the decorator wraps, when the poll function must stay callable under its own name::

    stage = Extractor.of( config_topics=["my-config"], poll=my_poll_fn, )

  • As a subclass for lifecycle management (HTTP clients, MQTT sessions)::

    class MyExtractor(Extractor): config_topics = ["my-config"]

      async def __aenter__(self):
      self.http = httpx.AsyncClient()
      return self

  async def __aexit__(self, *exc_info):
      await self.http.aclose()

  async def poll(self, config, state):
      ...

Config topics are re-read from the earliest on every startup — never through a committed-offset consumer group. The runner's membership consumer does join the application_id group, but purely for ownership leases: it never commits offsets and every record it fetches is discarded (see ExtractorRunner). Replicas up to the config topics' partition count split the configs between them; further replicas are hot standbys. self.configs (inherited from Stage) always holds the GLOBAL config store — scale-out only narrows which configs poll is invoked for. The caller sets the application_id used for changelog topic naming (and the membership group) on Flechtwerk; stages don't carry it.

A stage that overrides _run_in to take the broker-dispatched loop opts out of all of that: an external MQTT broker already sharded its work, so it takes no leases and polls every active config on every replica — at the price of being stateless by contract. Nothing else declares which model a stage uses: the Kafka resources each needs (a membership consumer, a changelog topic, a pinned token space) are acquired by the loop that needs them, so a stage that names no loop gets none of them.

Source code in src/flechtwerk/extractor.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class Extractor(Stage, ABC):
    """Poll-driven data extractor (stateful or stateless).

    Three ways to construct one:

    * Declaratively with the ``@extractor(...)`` decorator, which binds a poll
      function to its config topics — the decorated name becomes the stage::

          @extractor(config_topics=["my-config"])
          async def stage(config, state):
              ...

    * Functionally with ``Extractor.of(...)``, the factory the decorator wraps,
      when the poll function must stay callable under its own name::

          stage = Extractor.of(
              config_topics=["my-config"],
              poll=my_poll_fn,
          )

    * As a subclass for lifecycle management (HTTP clients, MQTT sessions)::

          class MyExtractor(Extractor):
              config_topics = ["my-config"]

              async def __aenter__(self):
                  self.http = httpx.AsyncClient()
                  return self

              async def __aexit__(self, *exc_info):
                  await self.http.aclose()

              async def poll(self, config, state):
                  ...

    Config topics are re-read from the earliest on every startup — never
    through a committed-offset consumer group. The runner's membership
    consumer does join the ``application_id`` group, but purely for
    ownership leases: it never commits offsets and every record it fetches
    is discarded (see `ExtractorRunner`). Replicas up to the config topics'
    partition count split the configs between them; further replicas are
    hot standbys. ``self.configs`` (inherited from `Stage`) always holds
    the GLOBAL config store — scale-out only narrows which configs ``poll``
    is invoked for. The caller sets the ``application_id`` used for
    changelog topic naming (and the membership group) on `Flechtwerk`;
    stages don't carry it.

    A stage that overrides ``_run_in`` to take the broker-dispatched loop
    opts out of all of that: an external MQTT broker already sharded its
    work, so it takes no leases and polls every active config on every
    replica — at the price of being stateless by contract. Nothing else
    declares which model a stage uses: the Kafka resources each needs (a
    membership consumer, a changelog topic, a pinned token space) are
    acquired by the loop that needs them, so a stage that names no loop gets
    none of them.
    """

    wakeup: asyncio.Event | None = None
    """Optional wakeup for push-driven sources.

    ``None`` (the default) keeps the runner on a plain
    ``poll_interval`` sleep between cycles. A stage whose input
    arrives asynchronously (e.g. an MQTT subscription) sets this in
    ``__aenter__`` and fires it on arrival; the runner then treats the
    interval as an upper bound, polling as soon as the event is set —
    the interval degrades to the idle/config-drain cadence.
    """

    @classmethod
    def of(
            cls,
            *,
            config_topics: list[str],
            poll: PollFn,
            enrich_config: EnrichConfigFn | None = None,
            extract_state_key: ExtractStateKeyFn | None = None,
            on_invalid_message: OnInvalidMessageFn | None = None,
    ) -> "Extractor":
        """Build an Extractor from a poll function and config topics.

        ``enrich_config``, ``extract_state_key``, and ``on_invalid_message`` are
        optional overrides; omit them
        to use the defaults (no enrichment, ``extract_state_key`` returns the
        Kafka message key, an undecodable config record raises).

        Patches the supplied callables in as instance attributes that
        shadow the class-level abstract method ``poll`` (and, when
        provided, the default ``enrich_config`` / ``extract_state_key`` /
        ``on_invalid_message`` methods). The
        ABC discipline still applies to every other construction path —
        ``Extractor()`` and any abstract subclass remain uninstantiable.
        """
        instance = _FunctionalExtractor()
        instance.config_topics = config_topics
        instance.poll = poll  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if enrich_config is not None:
            instance.enrich_config = enrich_config  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if extract_state_key is not None:
            instance.extract_state_key = extract_state_key  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        if on_invalid_message is not None:
            instance.on_invalid_message = on_invalid_message  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
        return instance

    async def on_active_configs(self, configs: dict[str, Config]) -> None:
        """Reconciliation hook: the configs this instance actively polls.

        The runner is the SOLE caller — implementations override this,
        never invoke it. The contract, guaranteed at every call:

        - ``configs`` is the complete active set, keyed by wire key: every
          config this instance currently owns that is not suspended — for a
          broker-dispatched stage, which owns nothing, simply every
          non-suspended config. Tombstoned, suspended, rewritten, and (on
          the token path) disowned configs simply disappear from the
          mapping — one idempotent reconciliation covers every removal
          shape.
        - No poll is in flight, and every page a completed poll produced
          has committed (the re-entry contract) — the hook may dispose
          per-config resources without racing ``poll()``.
        - It fires eventually after any change to the active set, and may
          fire when nothing changed — implementations must be idempotent.
        - It is NOT called on shutdown: end-of-life cleanup belongs in
          ``__aexit__``. (The MQTT template deliberately keeps its
          persistent MQTT broker session across restarts.)

        The mapping and its configs are the runner's cache — treat them as
        read-only. The default does nothing; the MQTT template overrides
        this to unsubscribe every topic no active config declares.
        """

    @abstractmethod
    def poll(self, config: Config, state: State) -> AsyncIterator[Message | State]:
        """Poll an external API and yield Messages.

        Declared without ``async`` so that implementations — ``async def``
        functions containing ``yield``, i.e. async generator functions whose
        call returns an ``AsyncIterator`` directly — are compatible overrides
        under strict type checking. A coroutine-typed abstract (``async def``
        with no ``yield``) would make every real override incompatible.

        Yield a State to end a page: every ``State`` yield is a COMMIT
        BOUNDARY — the messages yielded since the previous boundary and the
        state change (persisted only if it differs from the last committed
        value) commit in one Kafka transaction. Order matters: yield a
        page's messages FIRST and the State that accounts for them LAST —
        the State yield is what closes the page. Inverting that splits them
        across two transactions, cursor before messages, and a crash
        between the two commits loses the page for good: the re-poll
        resumes from a cursor that already skipped past it. Yield your
        resume cursor once per page of source data, and at least once every
        10 minutes during long extractions (the transaction timeout).
        Messages yielded after the last boundary — or by a poll that never
        yields State at all — are NOT lost: they commit as one trailing
        page when the generator completes, only without a cursor, so the
        next poll re-enters with the prior state. Yielding an empty/falsy
        State deletes the entry from the state store (and writes a Kafka
        tombstone to the changelog). On crash, the last-COMMITTED page is
        retained — its messages are already durable, and the uncommitted
        page's messages were aborted, so re-polling from the committed
        cursor duplicates nothing.

        Both parameters are read-only. The runner hands each poll a private copy
        of ``config`` and ``state``, so mutating either in place has no effect —
        it is silently discarded. Emit records by yielding a ``Message`` and
        persist your resume cursor by yielding a ``State``.
        """

    async def _run_in(self, runner: "ExtractorRunner") -> Never:
        """Run this stage under the ownership model it needs.

        `ExtractorRunner` delegates here instead of inspecting the stage: a
        stage KNOWS how its work is divided across replicas, so it names its
        loop rather than describing itself and leaving the runner to branch.
        The default is Kafka-leased token sharding. A stage whose work an
        external broker already sharded overrides this with
        ``runner.run_dispatched()`` — which is also what keeps
        `ExtractorRunner` free of any import of `flechtwerk.mqtt`, and what
        makes a third ownership model a new override rather than a third
        branch.

        Framework-internal, hence the underscore: this is a callback in the
        runner↔stage protocol, not an application hook. The runner is the
        SOLE caller and calls it exactly once, from inside the stage's own
        ``__aenter__`` / ``__aexit__``; overriding it is reserved for
        framework transport adapters (`MqttExtractor` is the only one).
        Neither loop returns under normal operation.
        """
        return await runner.run_sharded()

wakeup class-attribute instance-attribute 

wakeup: Event | None = None

Optional wakeup for push-driven sources.

None (the default) keeps the runner on a plain poll_interval sleep between cycles. A stage whose input arrives asynchronously (e.g. an MQTT subscription) sets this in __aenter__ and fires it on arrival; the runner then treats the interval as an upper bound, polling as soon as the event is set — the interval degrades to the idle/config-drain cadence.

of classmethod 

of(
    *,
    config_topics: list[str],
    poll: PollFn,
    enrich_config: EnrichConfigFn | None = None,
    extract_state_key: ExtractStateKeyFn | None = None,
    on_invalid_message: OnInvalidMessageFn | None = None
) -> Extractor

Build an Extractor from a poll function and config topics.

enrich_config, extract_state_key, and on_invalid_message are optional overrides; omit them to use the defaults (no enrichment, extract_state_key returns the Kafka message key, an undecodable config record raises).

Patches the supplied callables in as instance attributes that shadow the class-level abstract method poll (and, when provided, the default enrich_config / extract_state_key / on_invalid_message methods). The ABC discipline still applies to every other construction path — Extractor() and any abstract subclass remain uninstantiable.

Source code in src/flechtwerk/extractor.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@classmethod
def of(
        cls,
        *,
        config_topics: list[str],
        poll: PollFn,
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | None = None,
        on_invalid_message: OnInvalidMessageFn | None = None,
) -> "Extractor":
    """Build an Extractor from a poll function and config topics.

    ``enrich_config``, ``extract_state_key``, and ``on_invalid_message`` are
    optional overrides; omit them
    to use the defaults (no enrichment, ``extract_state_key`` returns the
    Kafka message key, an undecodable config record raises).

    Patches the supplied callables in as instance attributes that
    shadow the class-level abstract method ``poll`` (and, when
    provided, the default ``enrich_config`` / ``extract_state_key`` /
    ``on_invalid_message`` methods). The
    ABC discipline still applies to every other construction path —
    ``Extractor()`` and any abstract subclass remain uninstantiable.
    """
    instance = _FunctionalExtractor()
    instance.config_topics = config_topics
    instance.poll = poll  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if enrich_config is not None:
        instance.enrich_config = enrich_config  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if extract_state_key is not None:
        instance.extract_state_key = extract_state_key  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    if on_invalid_message is not None:
        instance.on_invalid_message = on_invalid_message  # pyright: ignore[reportAttributeAccessIssue] — instance shadow of the method
    return instance

on_active_configs async 

on_active_configs(configs: dict[str, Config]) -> None

Reconciliation hook: the configs this instance actively polls.

The runner is the SOLE caller — implementations override this, never invoke it. The contract, guaranteed at every call:

  • configs is the complete active set, keyed by wire key: every config this instance currently owns that is not suspended — for a broker-dispatched stage, which owns nothing, simply every non-suspended config. Tombstoned, suspended, rewritten, and (on the token path) disowned configs simply disappear from the mapping — one idempotent reconciliation covers every removal shape.
  • No poll is in flight, and every page a completed poll produced has committed (the re-entry contract) — the hook may dispose per-config resources without racing poll().
  • It fires eventually after any change to the active set, and may fire when nothing changed — implementations must be idempotent.
  • It is NOT called on shutdown: end-of-life cleanup belongs in __aexit__. (The MQTT template deliberately keeps its persistent MQTT broker session across restarts.)

The mapping and its configs are the runner's cache — treat them as read-only. The default does nothing; the MQTT template overrides this to unsubscribe every topic no active config declares.

Source code in src/flechtwerk/extractor.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
async def on_active_configs(self, configs: dict[str, Config]) -> None:
    """Reconciliation hook: the configs this instance actively polls.

    The runner is the SOLE caller — implementations override this,
    never invoke it. The contract, guaranteed at every call:

    - ``configs`` is the complete active set, keyed by wire key: every
      config this instance currently owns that is not suspended — for a
      broker-dispatched stage, which owns nothing, simply every
      non-suspended config. Tombstoned, suspended, rewritten, and (on
      the token path) disowned configs simply disappear from the
      mapping — one idempotent reconciliation covers every removal
      shape.
    - No poll is in flight, and every page a completed poll produced
      has committed (the re-entry contract) — the hook may dispose
      per-config resources without racing ``poll()``.
    - It fires eventually after any change to the active set, and may
      fire when nothing changed — implementations must be idempotent.
    - It is NOT called on shutdown: end-of-life cleanup belongs in
      ``__aexit__``. (The MQTT template deliberately keeps its
      persistent MQTT broker session across restarts.)

    The mapping and its configs are the runner's cache — treat them as
    read-only. The default does nothing; the MQTT template overrides
    this to unsubscribe every topic no active config declares.
    """

poll abstractmethod 

poll(
    config: Config, state: State
) -> AsyncIterator[Message | State]

Poll an external API and yield Messages.

Declared without async so that implementations — async def functions containing yield, i.e. async generator functions whose call returns an AsyncIterator directly — are compatible overrides under strict type checking. A coroutine-typed abstract (async def with no yield) would make every real override incompatible.

Yield a State to end a page: every State yield is a COMMIT BOUNDARY — the messages yielded since the previous boundary and the state change (persisted only if it differs from the last committed value) commit in one Kafka transaction. Order matters: yield a page's messages FIRST and the State that accounts for them LAST — the State yield is what closes the page. Inverting that splits them across two transactions, cursor before messages, and a crash between the two commits loses the page for good: the re-poll resumes from a cursor that already skipped past it. Yield your resume cursor once per page of source data, and at least once every 10 minutes during long extractions (the transaction timeout). Messages yielded after the last boundary — or by a poll that never yields State at all — are NOT lost: they commit as one trailing page when the generator completes, only without a cursor, so the next poll re-enters with the prior state. Yielding an empty/falsy State deletes the entry from the state store (and writes a Kafka tombstone to the changelog). On crash, the last-COMMITTED page is retained — its messages are already durable, and the uncommitted page's messages were aborted, so re-polling from the committed cursor duplicates nothing.

Both parameters are read-only. The runner hands each poll a private copy of config and state, so mutating either in place has no effect — it is silently discarded. Emit records by yielding a Message and persist your resume cursor by yielding a State.

Source code in src/flechtwerk/extractor.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@abstractmethod
def poll(self, config: Config, state: State) -> AsyncIterator[Message | State]:
    """Poll an external API and yield Messages.

    Declared without ``async`` so that implementations — ``async def``
    functions containing ``yield``, i.e. async generator functions whose
    call returns an ``AsyncIterator`` directly — are compatible overrides
    under strict type checking. A coroutine-typed abstract (``async def``
    with no ``yield``) would make every real override incompatible.

    Yield a State to end a page: every ``State`` yield is a COMMIT
    BOUNDARY — the messages yielded since the previous boundary and the
    state change (persisted only if it differs from the last committed
    value) commit in one Kafka transaction. Order matters: yield a
    page's messages FIRST and the State that accounts for them LAST —
    the State yield is what closes the page. Inverting that splits them
    across two transactions, cursor before messages, and a crash
    between the two commits loses the page for good: the re-poll
    resumes from a cursor that already skipped past it. Yield your
    resume cursor once per page of source data, and at least once every
    10 minutes during long extractions (the transaction timeout).
    Messages yielded after the last boundary — or by a poll that never
    yields State at all — are NOT lost: they commit as one trailing
    page when the generator completes, only without a cursor, so the
    next poll re-enters with the prior state. Yielding an empty/falsy
    State deletes the entry from the state store (and writes a Kafka
    tombstone to the changelog). On crash, the last-COMMITTED page is
    retained — its messages are already durable, and the uncommitted
    page's messages were aborted, so re-polling from the committed
    cursor duplicates nothing.

    Both parameters are read-only. The runner hands each poll a private copy
    of ``config`` and ``state``, so mutating either in place has no effect —
    it is silently discarded. Emit records by yielding a ``Message`` and
    persist your resume cursor by yielding a ``State``.
    """

flechtwerk.extractor.extractor 

extractor(
    *,
    config_topics: list[str],
    enrich_config: EnrichConfigFn | None = None,
    extract_state_key: ExtractStateKeyFn | None = None,
    on_invalid_message: OnInvalidMessageFn | None = None
) -> Callable[[PollFn], Extractor]

Decorator form of Extractor.of — bind a poll function to its config topics.

The decorated async generator becomes the built Extractor, so the name you define is the stage, ready to hand to Flechtwerk.of::

@extractor(config_topics=["my-config"])
async def stage(config: Config, state: State) -> AsyncIterator[Message | State]:
    ...

enrich_config, extract_state_key, and on_invalid_message are the same optional overrides as on Extractor.of — this is exactly that call with poll supplied by the decoration. Call Extractor.of directly when the poll function must stay callable under its own name, and subclass Extractor when you need lifecycle management (__aenter__ / __aexit__).

Source code in src/flechtwerk/extractor.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def extractor(
        *,
        config_topics: list[str],
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | None = None,
        on_invalid_message: OnInvalidMessageFn | None = None,
) -> Callable[[PollFn], Extractor]:
    """Decorator form of `Extractor.of` — bind a poll function to its config topics.

    The decorated async generator becomes the built `Extractor`, so the name you
    define *is* the stage, ready to hand to `Flechtwerk.of`::

        @extractor(config_topics=["my-config"])
        async def stage(config: Config, state: State) -> AsyncIterator[Message | State]:
            ...

    ``enrich_config``, ``extract_state_key``, and ``on_invalid_message`` are the
    same optional overrides as on
    `Extractor.of` — this is exactly that call with ``poll`` supplied by the
    decoration. Call `Extractor.of` directly when the poll function must stay
    callable under its own name, and subclass `Extractor` when you need
    lifecycle management (``__aenter__`` / ``__aexit__``).
    """
    def decorator(poll: PollFn) -> Extractor:
        return Extractor.of(
            config_topics=config_topics,
            enrich_config=enrich_config,
            extract_state_key=extract_state_key,
            on_invalid_message=on_invalid_message,
            poll=poll,
        )
    return decorator

Records & Messages

flechtwerk.Event

Bases: Record

Event object read from or written to a Kafka data topic.

Source code in src/flechtwerk/types.py 
15
16
class Event(Record):
    """Event object read from or written to a Kafka data topic."""

flechtwerk.State

Bases: Record

Mutable per-key state managed by the framework.

Source code in src/flechtwerk/types.py 
19
20
class State(Record):
    """Mutable per-key state managed by the framework."""

flechtwerk.Config

Bases: Record

Configuration object read from a Kafka config topic.

Source code in src/flechtwerk/types.py 
11
12
class Config(Record):
    """Configuration object read from a Kafka config topic."""

flechtwerk.Message dataclass

A message to be written to Kafka.

key and value each accept any Payload — see its docs for the encoding rules and for how to express other shapes — and value additionally accepts None, which produces a Kafka tombstone. Construction validates both fields so a mistake fails at the yield site, not inside the runner's transactional send path.

Source code in src/flechtwerk/types.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
@dataclass(frozen=True, slots=True)
class Message:
    """A message to be written to Kafka.

    ``key`` and ``value`` each accept any `Payload` — see its docs for the
    encoding rules and for how to express other shapes — and ``value``
    additionally accepts ``None``, which produces a Kafka tombstone. Construction
    validates both fields so a mistake fails at the yield site, not inside
    the runner's transactional send path.
    """

    key: Payload
    topic: str
    value: Payload | None
    timestamp: datetime | None = None

    def __post_init__(self) -> None:
        for name in ("key", "value"):
            v = getattr(self, name)
            if isinstance(v, (bytes, str, Event)):
                continue
            if v is None:
                if name == "value":
                    continue  # a tombstone
                raise TypeError(
                    "Message.key must not be None: Flechtwerk always writes keyed"
                    " records — a key is partitioning and, downstream, state identity."
                    " Only the value may be None, which writes a Kafka tombstone."
                )
            if isinstance(v, Config):
                raise TypeError(
                    f"Message.{name} must not be a Config: on the wire it travels as"
                    " data, and the reader assigns the semantics. Wrap it in"
                    " Event(config) — identical bytes — to mark the handoff."
                )
            if isinstance(v, State):
                raise TypeError(
                    f"Message.{name} must not be a State: a State inside a Message is"
                    " emitted, not persisted. Yield the State bare to persist it, or"
                    " wrap it in Event(state) to emit its contents."
                )
            if isinstance(v, dict):
                raise TypeError(
                    f"Message.{name} must not be a raw dict: wrap it in Event.wrap(...)"
                    " for identical wire bytes plus codec validation."
                )
            raise TypeError(
                f"Message.{name} must be bytes | str | Event, got"
                f" {type(v).__name__}. Encode other shapes to bytes yourself."
            )

flechtwerk.Payload module-attribute 

Payload = bytes | str | Event

What a Message may carry as key or value — one wire encoding per member.

  • bytes: sent as-is; the application has already encoded it. The escape hatch for foreign wire formats (Avro, msgpack, JSON scalars/arrays, ...).
  • str: UTF-8 text, deliberately NOT JSON-quoted — a wire-format commitment: string keys are decode_key's exact mirror, and plain-text values feed foreign readers (e.g. Druid lookup tables). Quoting them would remap every partition and state identity.
  • Event: canonical JSON — compact, sorted keys — so equal records produce identical bytes (stable partitioning for structured keys).

Event is the one Flechtwerk-schema payload: the wire carries no type, so the reader assigns the semantics — a config topic's consumers wrap the same bytes as Config regardless of what the producer held. Raw dicts are rejected: wrap them in Event.wrap(d) for identical wire bytes plus codec validation. State and Config are excluded on purpose — a State inside a Message would be emitted, not persisted (yield it bare to persist it), and a Config travels as data (wrap it in Event(config)); the explicit conversion marks the semantic handoff.

A Message value may additionally be None — a Kafka tombstone, the one thing the wire does type by itself. Log compaction removes a key whose latest value is null; b"" is NOT a tombstone but an ordinary, zero-length value that compaction retains forever. The two are never conflated in either direction: None round-trips as IncomingMessage.value None, b"" reads back as Event({}). Keys stay non-null — a null key is a partitioning decision, not a value.

flechtwerk.IncomingMessage dataclass

A message read from Kafka.

value is None exactly when the record's wire value is null — a tombstone (see Payload). An empty wire value (b"") is not a tombstone and decodes to Event({}), like b"{}". Only the wire views carry None: State and Config are store views, where a deletion is absence, never a null value.

Source code in src/flechtwerk/types.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass(frozen=True, slots=True)
class IncomingMessage:
    """A message read from Kafka.

    ``value`` is ``None`` exactly when the record's wire value is null — a
    tombstone (see `Payload`). An empty wire value (``b""``) is not a
    tombstone and decodes to ``Event({})``, like ``b"{}"``. Only the wire
    views carry ``None``: `State` and `Config` are store views, where a
    deletion is absence, never a null value.
    """

    key: str
    offset: int
    partition: int
    timestamp: datetime | None
    topic: str
    value: Event | None

flechtwerk.InvalidMessageError

Bases: Exception

A Kafka record whose key or value could not be decoded.

Raised by default when a record's key is not valid UTF-8 or its value is not a JSON dict; Stage.on_invalid_message overrides that policy (skip the record, or substitute a decoded value). The original UnicodeDecodeError / json.JSONDecodeError / non-dict ValueError is always chained in as __cause__.

str(...) is forensically complete on its own — part, topic, partition, offset — because the default policy is to raise and the traceback is the only announcement the framework makes (it logs nothing, on any outcome). The raw wire key and value stay available as attributes but out of the message: a value may be a megabyte.

Source code in src/flechtwerk/types.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class InvalidMessageError(Exception):
    """A Kafka record whose key or value could not be decoded.

    Raised by default when a record's key is not valid UTF-8 or its value is
    not a JSON dict; `Stage.on_invalid_message` overrides that policy (skip
    the record, or substitute a decoded value). The original
    ``UnicodeDecodeError`` / ``json.JSONDecodeError`` / non-dict
    ``ValueError`` is always chained in as ``__cause__``.

    ``str(...)`` is forensically complete on its own — part, topic,
    partition, offset — because the default policy is to raise and the
    traceback is the only announcement the framework makes (it logs
    nothing, on any outcome). The raw wire ``key`` and ``value`` stay
    available as attributes but out of the message: a value may be a
    megabyte.
    """

    def __init__(
            self,
            *,
            part: Literal["key", "value"],
            topic: str,
            partition: int,
            offset: int,
            key: bytes | str | None,
            value: bytes | str | None,
    ) -> None:
        super().__init__(
            f"Undecodable {part} in the record at {topic}/{partition}/{offset}"
            " — see __cause__ for the decode error, and the key/value attributes"
            " for the raw wire bytes. Override Stage.on_invalid_message to skip"
            " the record or substitute a value."
        )
        self.part = part
        self.topic = topic
        self.partition = partition
        self.offset = offset
        self.key = key
        self.value = value

Typed Attributes

flechtwerk.attribute.Attribute

A typed handle on one key in a RawDict, paired with an encode/decode codec.

optional=False (the default) declares a required field whose write_to rejects None; optional=True allows None (stored as JSON null). The read methods are kind-agnostic — which presence semantic applies is chosen by the caller's method ([] vs .get / .pop), not by this flag.

Public attributes:

  • name: the wire-level dict key this attribute reads/writes.
  • codec: the Codec[V] driving encode/decode. Exposed so callers can compose codecs (e.g. LIST(some_attr.codec) lifts an attribute's codec into a list-of-V codec). Direct encode/decode access goes through attr.codec.encode / attr.codec.decode; the Attribute itself only surfaces dict-operating methods (read_from, write_to, present_in, delete_from, get_from, pop_from).
  • optional: whether None / absence is permitted on write.
Source code in src/flechtwerk/attribute/attribute.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class Attribute[V]:
    """A typed handle on one key in a `RawDict`, paired with an encode/decode codec.

    `optional=False` (the default) declares a required field whose
    `write_to` rejects `None`; `optional=True` allows `None` (stored as
    JSON `null`). The read methods are kind-agnostic — which presence
    semantic applies is chosen by the caller's method (`[]` vs
    `.get` / `.pop`), not by this flag.

    Public attributes:

    - `name`: the wire-level dict key this attribute reads/writes.
    - `codec`: the `Codec[V]` driving encode/decode. Exposed so callers can
      compose codecs (e.g. `LIST(some_attr.codec)` lifts an attribute's
      codec into a list-of-V codec). Direct encode/decode access goes
      through `attr.codec.encode` / `attr.codec.decode`; the Attribute
      itself only surfaces dict-operating methods (`read_from`,
      `write_to`, `present_in`, `delete_from`, `get_from`, `pop_from`).
    - `optional`: whether `None` / absence is permitted on write.
    """

    def __init__(self, name: str, codec: Codec[V], *, optional: bool = False) -> None:
        self.name = name
        self.codec = codec
        self.optional = optional

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Reject subclasses outside this module — the Attribute hierarchy is sealed."""
        super().__init_subclass__(**kwargs)
        if cls.__module__ != Attribute.__module__:
            raise TypeError(f"{cls.__qualname__} cannot extend the sealed Attribute hierarchy")

    @property
    def required(self) -> bool:
        """Whether this attribute must be present and non-`None` — the inverse of `optional`."""
        return not self.optional

    def read_from(self, raw: RawDict) -> V:
        """Look up this attribute in `raw` and return the decoded value.

        Raises `MissingAttributeError` if the key is absent or the stored
        value is `None`. This is kind-agnostic — a `[]` read raises on
        absence regardless of `optional`; absence-tolerance lives in
        `Record.get` / `Record.pop`.
        """
        # Null ≡ missing is duplicated in get_from and pop_from — change all in lockstep.
        v = raw.get(self.name)
        if v is None:
            raise MissingAttributeError(f"attribute {self!r} is missing")
        return self.codec.decode(v)

    def write_to(self, raw: RawDict, value: V | None) -> None:
        """Encode `value` and store it under this attribute's name in `raw`.

        A required attribute (`optional=False`) rejects `None` at the
        write-site so it can't land silently as JSON `null`; an optional
        attribute stores `None` as `null`. The `not self.optional` check
        rides the `None` branch only, so the common non-`None` write pays
        nothing for it.
        """
        if value is None:
            if not self.optional:
                raise ValueError(f"cannot assign None to required {self!r}")
            raw[self.name] = None
        else:
            raw[self.name] = self.codec.encode(value)

    def present_in(self, raw: RawDict) -> bool:
        """Whether this attribute is present in `raw` (key exists, value may be `None`)."""
        return self.name in raw

    def delete_from(self, raw: RawDict) -> None:
        """Remove this attribute's entry from `raw`. Raises `KeyError` if absent."""
        del raw[self.name]

    def get_from(self, raw: RawDict, default: V | None = None) -> V | None:
        """Return the decoded value, or `default` if missing or `None`."""
        v = raw.get(self.name)
        return self.codec.decode(v) if v is not None else default

    def pop_from(self, raw: RawDict, *default: V) -> V | None:
        """Remove and return the decoded value; raise `KeyError` if missing and no default.

        A stored `None` is returned as `None` (no decode), and the key is
        removed — mirroring `dict.pop` semantics for `None` writes.
        """
        if not self.present_in(raw):
            if default:
                return default[0]
            raise KeyError(self)
        v = raw.get(self.name)
        if v is not None:
            v = self.codec.decode(v)
        self.delete_from(raw)
        return v

    # Identity is (type, name) — never the codec or `optional`. The name is the
    # wire key this handle addresses, so two handles for the same slot are equal.
    # The codec is excluded on purpose: codecs compare by object identity, so a
    # composite codec like LIST(STR) is rebuilt per call and would make
    # "same field" attributes unequal. `type` is in the key so a ViewAttribute
    # stays distinct from a plain Attribute of the same name — the dict-spread
    # override relies on both write_to calls running, the typed one landing last.
    def __eq__(self, other: object) -> bool:
        return type(other) is type(self) and other.name == self.name  # type: ignore[attr-defined]

    def __hash__(self) -> int:
        return hash((type(self), self.name))

    def __repr__(self) -> str:
        kind = ", optional=True" if self.optional else ""
        return f"{type(self).__name__}({self.name!r}{kind})"

required property 

required: bool

Whether this attribute must be present and non-None — the inverse of optional.

__init_subclass__ 

__init_subclass__(**kwargs: Any) -> None

Reject subclasses outside this module — the Attribute hierarchy is sealed.

Source code in src/flechtwerk/attribute/attribute.py 
60
61
62
63
64
def __init_subclass__(cls, **kwargs: Any) -> None:
    """Reject subclasses outside this module — the Attribute hierarchy is sealed."""
    super().__init_subclass__(**kwargs)
    if cls.__module__ != Attribute.__module__:
        raise TypeError(f"{cls.__qualname__} cannot extend the sealed Attribute hierarchy")

read_from 

read_from(raw: RawDict) -> V

Look up this attribute in raw and return the decoded value.

Raises MissingAttributeError if the key is absent or the stored value is None. This is kind-agnostic — a [] read raises on absence regardless of optional; absence-tolerance lives in Record.get / Record.pop.

Source code in src/flechtwerk/attribute/attribute.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def read_from(self, raw: RawDict) -> V:
    """Look up this attribute in `raw` and return the decoded value.

    Raises `MissingAttributeError` if the key is absent or the stored
    value is `None`. This is kind-agnostic — a `[]` read raises on
    absence regardless of `optional`; absence-tolerance lives in
    `Record.get` / `Record.pop`.
    """
    # Null ≡ missing is duplicated in get_from and pop_from — change all in lockstep.
    v = raw.get(self.name)
    if v is None:
        raise MissingAttributeError(f"attribute {self!r} is missing")
    return self.codec.decode(v)

write_to 

write_to(raw: RawDict, value: V | None) -> None

Encode value and store it under this attribute's name in raw.

A required attribute (optional=False) rejects None at the write-site so it can't land silently as JSON null; an optional attribute stores None as null. The not self.optional check rides the None branch only, so the common non-None write pays nothing for it.

Source code in src/flechtwerk/attribute/attribute.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def write_to(self, raw: RawDict, value: V | None) -> None:
    """Encode `value` and store it under this attribute's name in `raw`.

    A required attribute (`optional=False`) rejects `None` at the
    write-site so it can't land silently as JSON `null`; an optional
    attribute stores `None` as `null`. The `not self.optional` check
    rides the `None` branch only, so the common non-`None` write pays
    nothing for it.
    """
    if value is None:
        if not self.optional:
            raise ValueError(f"cannot assign None to required {self!r}")
        raw[self.name] = None
    else:
        raw[self.name] = self.codec.encode(value)

present_in 

present_in(raw: RawDict) -> bool

Whether this attribute is present in raw (key exists, value may be None).

Source code in src/flechtwerk/attribute/attribute.py 
101
102
103
def present_in(self, raw: RawDict) -> bool:
    """Whether this attribute is present in `raw` (key exists, value may be `None`)."""
    return self.name in raw

delete_from 

delete_from(raw: RawDict) -> None

Remove this attribute's entry from raw. Raises KeyError if absent.

Source code in src/flechtwerk/attribute/attribute.py 
105
106
107
def delete_from(self, raw: RawDict) -> None:
    """Remove this attribute's entry from `raw`. Raises `KeyError` if absent."""
    del raw[self.name]

get_from 

get_from(
    raw: RawDict, default: V | None = None
) -> V | None

Return the decoded value, or default if missing or None.

Source code in src/flechtwerk/attribute/attribute.py 
109
110
111
112
def get_from(self, raw: RawDict, default: V | None = None) -> V | None:
    """Return the decoded value, or `default` if missing or `None`."""
    v = raw.get(self.name)
    return self.codec.decode(v) if v is not None else default

pop_from 

pop_from(raw: RawDict, *default: V) -> V | None

Remove and return the decoded value; raise KeyError if missing and no default.

A stored None is returned as None (no decode), and the key is removed — mirroring dict.pop semantics for None writes.

Source code in src/flechtwerk/attribute/attribute.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def pop_from(self, raw: RawDict, *default: V) -> V | None:
    """Remove and return the decoded value; raise `KeyError` if missing and no default.

    A stored `None` is returned as `None` (no decode), and the key is
    removed — mirroring `dict.pop` semantics for `None` writes.
    """
    if not self.present_in(raw):
        if default:
            return default[0]
        raise KeyError(self)
    v = raw.get(self.name)
    if v is not None:
        v = self.codec.decode(v)
    self.delete_from(raw)
    return v

flechtwerk.attribute.codec

Every built-in codec, in one namespace.

Attribute[V] is constructed with a Codec[V]. Both directions are required: the codec is the single source of truth for how a V round-trips through JSON-native form. The [V] type parameter on Attribute is inferred from the codec by the type checker — there's no runtime type introspection.

This module is the complete codec catalogue: the Codec type and its Decoder / Encoder aliases, the atoms (STR, INT, BOOL, BYTES, DATE, FLOAT, DATETIME, TIME, ZONE_INFO, RECORD, ANY), the constructors (LIST, SET, TUPLE, DICT), and the record_codec factory for Record subclasses. The flechtwerk.attribute package re-exports all of them, so the short import works too; this module exists for the disambiguating one, when an application name collides with a codec:

from flechtwerk.attribute import codec

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

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

The one codec deliberately absent is ENCRYPTED: it lives in flechtwerk.secrets, the only framework module allowed to import joserfc (the flechtwerk[secrets] extra).

BOOL module-attribute 

BOOL: Final = Codec[bool](_validate(bool), _validate(bool))

BYTES module-attribute 

BYTES: Final = Codec[bytes](_decode_bytes, _encode_bytes)

Codec for bytes — RFC 4648 §4 base64 (standard alphabet, padded) as a JSON string.

The one atom whose Python type is not JSON-native, so the wire form is a lossy-looking str that only this codec knows how to read back. Two consequences worth declaring:

bytes is deliberately absent from the ANY walker — an ANY field holding binary would encode to a base64 string and decode back as that string, indistinguishable from text, and Record.wrap would start committing raw dicts to a base64 wire form nobody declared. Binary in a record is an explicit Attribute(name, BYTES) or a TypeError.

Base64 costs 4/3 of the payload before JSON escaping, against Kafka's 1 MiB record ceiling (see state_record_bytes). For a whole message that is a blob, Payload's bytes member ships it pre-encoded with no framing at all — BYTES is for a binary field beside other typed ones.

DATE module-attribute 

DATE: Final = Codec[date](
    date.fromisoformat, date.isoformat
)

DATETIME module-attribute 

DATETIME: Final = Codec[datetime](
    datetime.fromisoformat, _encode_datetime
)

FLOAT module-attribute 

FLOAT: Final = Codec[float](
    _validate(float), _validate(float)
)

INT module-attribute 

INT: Final = Codec[int](_validate(int), _validate(int))

STR module-attribute 

STR: Final = Codec[str](_validate(str), _validate(str))

TIME module-attribute 

TIME: Final = Codec[time](
    time.fromisoformat, time.isoformat
)

ZONE_INFO module-attribute 

ZONE_INFO: Final = Codec[ZoneInfo](
    _decode_zone_info, _encode_zone_info
)

Codec for an IANA time zone (Europe/Berlin, UTC, CET) as its key string.

Fixed-offset datetime.timezone values are out of scope: they have no IANA key, and an offset is not a zone (it knows nothing about DST). ZoneInfo reads the host's tz database, falling back to the tzdata package — an image without either rejects every key.

Like BYTES, the zone is deliberately absent from the ANY walker: its wire form is a plain string that would read back as text.

ANY module-attribute 

ANY: Final = Codec[Any](
    decode=lambda x: x, encode=_encode_any
)

The universal escape-hatch codec for Any-typed values.

Decode is identity (the wire value passes through unchanged — JSON load already produced JSON-native shape). Encode runs the recursive _encode_any walker, which dispatches on runtime type and handles Record, dict, list, set, tuple, datetime, date, time, and the JSON-native primitives.

Compose with the container constructors for typed-but-heterogeneous fields: LIST(DICT(ANY)) for list[dict[str, Any]], DICT(ANY) for dict[str, Any], etc.

RECORD module-attribute 

RECORD: Final = record_codec(Record)

Codec for an untyped Record value.

Use as Attribute("data", RECORD) for fields whose value is a plain Record. For Record subclasses, build a per-subclass codec via record_codec(cls) (or use the constants exported from flechtwerk.types for Config, Event, State).

Decoder 

Decoder = Callable[[Any], V]

A function that decodes a JSON-native wire value to a Python value of type V.

Encoder 

Encoder = Callable[[V], Any]

A function that encodes a Python value of type V to a JSON-native wire value.

Codec dataclass

A pair of (decode, encode) callables for a value of type V.

Both directions are required — there is no fallback registry. The value type V is the single source of truth: pass a Codec[V] to an Attribute and the type checker infers the Attribute[V] parameter.

Equality is identity (eq=False): the fields are functions, so a generated field-wise __eq__ would compare them by object identity anyway while looking like value equality — composite codecs like LIST(STR) rebuild fresh closures per call and would never compare equal. Identity is the only honest contract, and Attribute already excludes the codec from its own equality for this reason.

Source code in src/flechtwerk/attribute/_codec.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass(frozen=True, slots=True, eq=False)
class Codec[V]:
    """A pair of `(decode, encode)` callables for a value of type `V`.

    Both directions are required — there is no fallback registry. The value
    type `V` is the single source of truth: pass a `Codec[V]` to an
    `Attribute` and the type checker infers the `Attribute[V]` parameter.

    Equality is identity (`eq=False`): the fields are functions, so a
    generated field-wise `__eq__` would compare them by object identity
    anyway while *looking* like value equality — composite codecs like
    `LIST(STR)` rebuild fresh closures per call and would never compare
    equal. Identity is the only honest contract, and `Attribute` already
    excludes the codec from its own equality for this reason.
    """
    decode: Decoder[V]
    encode: Encoder[V]
decode instance-attribute 
decode: Decoder[V]
encode instance-attribute 
encode: Encoder[V]

DICT 

DICT(inner: Codec[V]) -> Codec[dict[str, V]]

Codec for dict[str, V] — runs inner over each value.

Keys must be exactly str — enforced statically by the dict[str, V] parameter and asserted at runtime via exact-type check (same discipline as _validate). Str subclasses (e.g. StrEnum members) are rejected — accepting them silently would lose type information at the boundary.

Source code in src/flechtwerk/attribute/_codec.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def DICT[V](inner: Codec[V]) -> Codec[dict[str, V]]:
    """Codec for `dict[str, V]` — runs `inner` over each value.

    Keys must be exactly `str` — enforced statically by the
    `dict[str, V]` parameter and asserted at runtime via exact-type
    check (same discipline as `_validate`). Str subclasses (e.g.
    `StrEnum` members) are rejected — accepting them silently would
    lose type information at the boundary.
    """

    def encode(d: dict[str, V]) -> dict[str, Any]:
        for k in d:
            assert type(k) is str, f"DICT key must be str, got {type(k).__name__}: {k!r}"
        return {k: inner.encode(v) for k, v in d.items()}

    return Codec(
        decode=lambda d: {k: inner.decode(v) for k, v in d.items()},
        encode=encode,
    )

LIST 

LIST(inner: Codec[V]) -> Codec[list[V]]

Codec for list[V] — runs inner over each element.

Source code in src/flechtwerk/attribute/_codec.py 
219
220
221
222
223
224
def LIST[V](inner: Codec[V]) -> Codec[list[V]]:
    """Codec for `list[V]` — runs `inner` over each element."""
    return Codec(
        decode=lambda lst: [inner.decode(x) for x in lst],
        encode=lambda lst: [inner.encode(x) for x in lst],
    )

SET 

SET(inner: Codec[V]) -> Codec[set[V]]

Codec for set[V] — runs inner over each element.

Encode emits a sorted list (deterministic wire form for diff stability). Sort is on raw values, then each is encoded — preserves the natural ordering of the source type. V must be hashable (a Python set requirement) and the raw values must be comparable to each other.

Source code in src/flechtwerk/attribute/_codec.py 
227
228
229
230
231
232
233
234
235
236
237
238
def SET[V](inner: Codec[V]) -> Codec[set[V]]:
    """Codec for `set[V]` — runs `inner` over each element.

    Encode emits a sorted list (deterministic wire form for diff stability).
    Sort is on raw values, then each is encoded — preserves the natural
    ordering of the source type. V must be hashable (a Python set
    requirement) and the raw values must be comparable to each other.
    """
    return Codec(
        decode=lambda items: {inner.decode(x) for x in items},
        encode=lambda s: [inner.encode(x) for x in sorted(s)],  # pyright: ignore[reportArgumentType] — comparable by contract
    )

TUPLE 

TUPLE(inner: Codec[V]) -> Codec[tuple[V, ...]]

Codec for tuple[V, ...] — runs inner over each element, preserves order.

Source code in src/flechtwerk/attribute/_codec.py 
241
242
243
244
245
246
def TUPLE[V](inner: Codec[V]) -> Codec[tuple[V, ...]]:
    """Codec for `tuple[V, ...]` — runs `inner` over each element, preserves order."""
    return Codec(
        decode=lambda items: tuple(inner.decode(x) for x in items),
        encode=lambda t: [inner.encode(x) for x in t],
    )

record_codec 

record_codec(cls: type[T]) -> Codec[T]

Build a Codec[T] for a Record subclass T.

Decode wraps the raw dict in cls; encode returns a shallow copy of .raw. Top-level isolation is preserved (the new owner gets its own dict to mutate via __setitem__); nested dicts/lists are shared, matching the framework's "treat .raw as immutable from outside" contract (which raw enforces at the top level).

Source code in src/flechtwerk/attribute/record.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
def record_codec[T: Record](cls: type[T]) -> Codec[T]:
    """Build a `Codec[T]` for a `Record` subclass `T`.

    Decode wraps the raw dict in `cls`; encode returns a shallow copy of
    `.raw`. Top-level isolation is preserved (the new owner gets its own
    dict to mutate via `__setitem__`); nested dicts/lists are shared,
    matching the framework's "treat `.raw` as immutable from outside"
    contract (which `raw` enforces at the top level).
    """
    return Codec(
        decode=cls.wrap,
        encode=lambda r: r._raw.copy(),
    )

Runtime & Configuration

flechtwerk.Flechtwerk

Bases: ABC

The application-facing handle for a Flechtwerk stage.

An application constructs one with the of factory and calls run()::

await Flechtwerk.of(
    application_id="my-extractor",
    bootstrap_servers="localhost:9092",
    client_id="my-extractor-0",
    poll_interval=timedelta(minutes=1),
    stage=my_extractor,
).run()

This is deliberately a narrow surface — of / run / the async context manager, nothing else. The Kafka resources (consumers, transactional producers, state stores, runners) live on the private _FlechtwerkModule reactor-di container that of returns; an application must not reach past this handle into the wiring, whose invariants (EOS-v1 fencing, changelog restore ordering, config-store bootstrap sequencing) are the framework's to keep. Same idiom as Extractor.of / Transformer.of: a factory on the public abstraction that returns a private concrete subclass typed as the abstraction.

To embed Flechtwerk as a component of a larger reactor-di module, wire the concrete container directly — declare make[Flechtwerk, _FlechtwerkModule] and let the parent module fill every lookup field by attribute name. That includes kafka_tls, which every Kafka client build reads: wire it to None for plaintext.

Several stages may share one process: build one handle per stage and run them as sibling tasks (asyncio.TaskGroup, so one stage's crash takes the process down), each with its own application_id and client_id. A metrics_port they share is ONE scrape endpoint — the stages are told apart by their metrics_labels values — and a keyring they pass must be the same one. See the getting-started guide, "Several Stages in One Process".

Source code in src/flechtwerk/module.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
class Flechtwerk(ABC):
    """The application-facing handle for a Flechtwerk stage.

    An application constructs one with the ``of`` factory and calls
    ``run()``::

        await Flechtwerk.of(
            application_id="my-extractor",
            bootstrap_servers="localhost:9092",
            client_id="my-extractor-0",
            poll_interval=timedelta(minutes=1),
            stage=my_extractor,
        ).run()

    This is deliberately a narrow surface — ``of`` / ``run`` / the async
    context manager, nothing else. The Kafka resources (consumers,
    transactional producers, state stores, runners) live on the private
    ``_FlechtwerkModule`` reactor-di container that ``of`` returns; an
    application must not reach past this handle into the wiring, whose
    invariants (EOS-v1 fencing, changelog restore ordering, config-store
    bootstrap sequencing) are the framework's to keep. Same idiom as
    ``Extractor.of`` / ``Transformer.of``: a factory on the public
    abstraction that returns a private concrete subclass typed as the
    abstraction.

    To embed Flechtwerk as a component of a larger reactor-di module, wire
    the concrete container directly — declare ``make[Flechtwerk,
    _FlechtwerkModule]`` and let the parent module fill every ``lookup``
    field by attribute name. That includes ``kafka_tls``, which every Kafka
    client build reads: wire it to ``None`` for plaintext.

    Several stages may share one process: build one handle per stage and
    run them as sibling tasks (``asyncio.TaskGroup``, so one stage's crash
    takes the process down), each with its own ``application_id`` and
    ``client_id``. A ``metrics_port`` they share is ONE scrape endpoint —
    the stages are told apart by their ``metrics_labels`` values — and a
    ``keyring`` they pass must be the same one. See the getting-started
    guide, "Several Stages in One Process".
    """

    @classmethod
    def of(
        cls,
        *,
        application_id: str,
        bootstrap_servers: str,
        client_id: str,
        stage: Extractor | Transformer,
        compression_type: CompressionType | None = "zstd",
        kafka_tls: ssl.SSLContext | None = None,
        keyring: Keyring | None = None,
        max_poll_records: int = 500,
        metrics_labels: dict[str, str] | None = None,
        metrics_port: int = 0,
        mqtt: MqttBrokerConfig | None = None,
        poll_interval: timedelta | None = None,
    ) -> "Flechtwerk":
        """Build a fully-configured application handle.

        Use this when running Flechtwerk as the program's entry point.
        ``client_id`` is the process identity: every Kafka client this
        module opens derives its ID from it, and for an MQTT-sourced stage
        it also names the persistent MQTT session and completes that
        stage's transactional ID (``{application_id}-{client_id}``) — so it
        must be unique per instance and stable across restarts (production
        K8s passes the pod name; a StatefulSet's ordinal identity is what
        lets a restarted replica resume its own session and its own
        undelivered tail). ``compression_type`` defaults to ``"zstd"`` because
        Flechtwerk outputs JSON everywhere (encode_json) and JSON
        compresses ~13×; pass ``None`` to disable. ``kafka_tls`` is a stdlib
        ``ssl.SSLContext`` for reaching the Kafka broker over TLS — mutual TLS
        when it carries a client certificate — and applies to every Kafka
        client this module opens; ``None``, the default, is plaintext, and a
        context that would not authenticate the Kafka broker is refused at
        startup (``__aenter__``). ``keyring`` carries the
        process keyring for ``flechtwerk.secrets`` encrypted attributes — it is
        installed once per process at stage startup (``__aenter__``; a
        conflicting second install raises) and may be left ``None`` by stages
        without encrypted attributes. Constructing a handle has no
        side effects; standalone producers/tooling call
        ``flechtwerk.secrets.install_keyring(...)`` directly.
        ``max_poll_records`` caps how many records one ``getmany()`` fetch
        may return — Kafka's ``max.poll.records``, and the same default of
        500 (aiokafka alone leaves it unbounded). The cap is what makes a
        backlog drain as many ordinary batches instead of one giant one: an
        unbounded first fetch after downtime returns the entire backlog as
        a single batch, which pins it in memory in parsed form, floods any
        per-message I/O in ``transform`` with one concurrent call per state
        key, and outlives ``max.poll.interval.ms`` — a mid-batch group
        eviction. ``metrics_labels``
        defaults to an empty dict and ``metrics_port`` defaults to 0
        (Prometheus disabled). ``mqtt`` carries the platform's shared MQTT
        MQTT broker settings; it is used only by MQTT-sourced stages and ignored
        everywhere else, so the caller may pass it unconditionally.
        ``poll_interval`` is likewise consumed only by extractors — the poll
        cadence (the runner's idle / wakeup wait). It defaults to ``None``
        ("unset"); an extractor requires a positive ``timedelta`` and a
        transformer ignores it, so this too may be passed unconditionally.
        ``application_id``, ``bootstrap_servers``, ``client_id`` and ``stage``
        are required. An extractor scales out by itself: replicas of the
        same ``application_id`` split the configs along the config topics'
        partitions (see ``ExtractorRunner``), so the deployment's replica
        count is the only scaling knob.
        """
        instance = _FlechtwerkModule()
        instance.application_id = application_id
        instance.bootstrap_servers = bootstrap_servers
        instance.client_id = client_id
        instance.compression_type = compression_type
        instance.kafka_tls = kafka_tls
        instance.keyring = keyring
        instance.max_poll_records = max_poll_records
        instance.metrics_labels = dict(metrics_labels) if metrics_labels else {}
        instance.metrics_port = metrics_port
        instance.mqtt = mqtt
        instance.poll_interval = poll_interval
        instance.stage = stage
        return instance

    @abstractmethod
    async def run(self) -> Never:
        """Bootstrap resources and run the configured stage forever.

        The runner's main loop is ``while True``, so under normal operation
        this coroutine never returns — it terminates only by cancellation
        or an unrecovered exception.
        """
        ...

    @abstractmethod
    async def __aenter__(self) -> Self:
        ...

    @abstractmethod
    async def __aexit__(self, *exc_info: object) -> None:
        ...

of classmethod 

of(
    *,
    application_id: str,
    bootstrap_servers: str,
    client_id: str,
    stage: Extractor | Transformer,
    compression_type: CompressionType | None = "zstd",
    kafka_tls: SSLContext | None = None,
    keyring: Keyring | None = None,
    max_poll_records: int = 500,
    metrics_labels: dict[str, str] | None = None,
    metrics_port: int = 0,
    mqtt: MqttBrokerConfig | None = None,
    poll_interval: timedelta | None = None
) -> Flechtwerk

Build a fully-configured application handle.

Use this when running Flechtwerk as the program's entry point. client_id is the process identity: every Kafka client this module opens derives its ID from it, and for an MQTT-sourced stage it also names the persistent MQTT session and completes that stage's transactional ID ({application_id}-{client_id}) — so it must be unique per instance and stable across restarts (production K8s passes the pod name; a StatefulSet's ordinal identity is what lets a restarted replica resume its own session and its own undelivered tail). compression_type defaults to "zstd" because Flechtwerk outputs JSON everywhere (encode_json) and JSON compresses ~13×; pass None to disable. kafka_tls is a stdlib ssl.SSLContext for reaching the Kafka broker over TLS — mutual TLS when it carries a client certificate — and applies to every Kafka client this module opens; None, the default, is plaintext, and a context that would not authenticate the Kafka broker is refused at startup (__aenter__). keyring carries the process keyring for flechtwerk.secrets encrypted attributes — it is installed once per process at stage startup (__aenter__; a conflicting second install raises) and may be left None by stages without encrypted attributes. Constructing a handle has no side effects; standalone producers/tooling call flechtwerk.secrets.install_keyring(...) directly. max_poll_records caps how many records one getmany() fetch may return — Kafka's max.poll.records, and the same default of 500 (aiokafka alone leaves it unbounded). The cap is what makes a backlog drain as many ordinary batches instead of one giant one: an unbounded first fetch after downtime returns the entire backlog as a single batch, which pins it in memory in parsed form, floods any per-message I/O in transform with one concurrent call per state key, and outlives max.poll.interval.ms — a mid-batch group eviction. metrics_labels defaults to an empty dict and metrics_port defaults to 0 (Prometheus disabled). mqtt carries the platform's shared MQTT MQTT broker settings; it is used only by MQTT-sourced stages and ignored everywhere else, so the caller may pass it unconditionally. poll_interval is likewise consumed only by extractors — the poll cadence (the runner's idle / wakeup wait). It defaults to None ("unset"); an extractor requires a positive timedelta and a transformer ignores it, so this too may be passed unconditionally. application_id, bootstrap_servers, client_id and stage are required. An extractor scales out by itself: replicas of the same application_id split the configs along the config topics' partitions (see ExtractorRunner), so the deployment's replica count is the only scaling knob.

Source code in src/flechtwerk/module.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
@classmethod
def of(
    cls,
    *,
    application_id: str,
    bootstrap_servers: str,
    client_id: str,
    stage: Extractor | Transformer,
    compression_type: CompressionType | None = "zstd",
    kafka_tls: ssl.SSLContext | None = None,
    keyring: Keyring | None = None,
    max_poll_records: int = 500,
    metrics_labels: dict[str, str] | None = None,
    metrics_port: int = 0,
    mqtt: MqttBrokerConfig | None = None,
    poll_interval: timedelta | None = None,
) -> "Flechtwerk":
    """Build a fully-configured application handle.

    Use this when running Flechtwerk as the program's entry point.
    ``client_id`` is the process identity: every Kafka client this
    module opens derives its ID from it, and for an MQTT-sourced stage
    it also names the persistent MQTT session and completes that
    stage's transactional ID (``{application_id}-{client_id}``) — so it
    must be unique per instance and stable across restarts (production
    K8s passes the pod name; a StatefulSet's ordinal identity is what
    lets a restarted replica resume its own session and its own
    undelivered tail). ``compression_type`` defaults to ``"zstd"`` because
    Flechtwerk outputs JSON everywhere (encode_json) and JSON
    compresses ~13×; pass ``None`` to disable. ``kafka_tls`` is a stdlib
    ``ssl.SSLContext`` for reaching the Kafka broker over TLS — mutual TLS
    when it carries a client certificate — and applies to every Kafka
    client this module opens; ``None``, the default, is plaintext, and a
    context that would not authenticate the Kafka broker is refused at
    startup (``__aenter__``). ``keyring`` carries the
    process keyring for ``flechtwerk.secrets`` encrypted attributes — it is
    installed once per process at stage startup (``__aenter__``; a
    conflicting second install raises) and may be left ``None`` by stages
    without encrypted attributes. Constructing a handle has no
    side effects; standalone producers/tooling call
    ``flechtwerk.secrets.install_keyring(...)`` directly.
    ``max_poll_records`` caps how many records one ``getmany()`` fetch
    may return — Kafka's ``max.poll.records``, and the same default of
    500 (aiokafka alone leaves it unbounded). The cap is what makes a
    backlog drain as many ordinary batches instead of one giant one: an
    unbounded first fetch after downtime returns the entire backlog as
    a single batch, which pins it in memory in parsed form, floods any
    per-message I/O in ``transform`` with one concurrent call per state
    key, and outlives ``max.poll.interval.ms`` — a mid-batch group
    eviction. ``metrics_labels``
    defaults to an empty dict and ``metrics_port`` defaults to 0
    (Prometheus disabled). ``mqtt`` carries the platform's shared MQTT
    MQTT broker settings; it is used only by MQTT-sourced stages and ignored
    everywhere else, so the caller may pass it unconditionally.
    ``poll_interval`` is likewise consumed only by extractors — the poll
    cadence (the runner's idle / wakeup wait). It defaults to ``None``
    ("unset"); an extractor requires a positive ``timedelta`` and a
    transformer ignores it, so this too may be passed unconditionally.
    ``application_id``, ``bootstrap_servers``, ``client_id`` and ``stage``
    are required. An extractor scales out by itself: replicas of the
    same ``application_id`` split the configs along the config topics'
    partitions (see ``ExtractorRunner``), so the deployment's replica
    count is the only scaling knob.
    """
    instance = _FlechtwerkModule()
    instance.application_id = application_id
    instance.bootstrap_servers = bootstrap_servers
    instance.client_id = client_id
    instance.compression_type = compression_type
    instance.kafka_tls = kafka_tls
    instance.keyring = keyring
    instance.max_poll_records = max_poll_records
    instance.metrics_labels = dict(metrics_labels) if metrics_labels else {}
    instance.metrics_port = metrics_port
    instance.mqtt = mqtt
    instance.poll_interval = poll_interval
    instance.stage = stage
    return instance

run abstractmethod async 

run() -> Never

Bootstrap resources and run the configured stage forever.

The runner's main loop is while True, so under normal operation this coroutine never returns — it terminates only by cancellation or an unrecovered exception.

Source code in src/flechtwerk/module.py 
389
390
391
392
393
394
395
396
397
@abstractmethod
async def run(self) -> Never:
    """Bootstrap resources and run the configured stage forever.

    The runner's main loop is ``while True``, so under normal operation
    this coroutine never returns — it terminates only by cancellation
    or an unrecovered exception.
    """
    ...

flechtwerk.ConfigStore

Latest config per wire key, merged across a stage's config topics.

Values are kept as wire bytes and parsed on every get() — each call returns a fresh Config (a protective copy by construction).

From a stage's perspective the store is read-only: query it with get() (and in / len / nbytes). There is no public write surface — _put/_delete are internal to the config machinery in this module, and reaching for them, or otherwise mutating the store, from application code is an error. The store is a projection of the config topics, fed exclusively by bootstrap_config_store / drain_config_updates; a stage-side write never reaches Kafka (see the "config topics never participate in a Kafka transaction" invariant), corrupts only this instance, and is silently reverted on the next record for the key or on the next restart.

Source code in src/flechtwerk/configs.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class ConfigStore:
    """Latest config per wire key, merged across a stage's config topics.

    Values are kept as wire bytes and parsed on every `get()` — each call
    returns a fresh `Config` (a protective copy by construction).

    From a stage's perspective the store is **read-only**: query it with
    `get()` (and ``in`` / ``len`` / `nbytes`). There is no public write
    surface — `_put`/`_delete` are internal to the config machinery in this
    module, and reaching for them, or otherwise mutating the store, from
    application code is an error. The store is a projection of the config
    topics, fed exclusively by `bootstrap_config_store` /
    `drain_config_updates`; a stage-side write never reaches Kafka (see the
    "config topics never participate in a Kafka transaction" invariant),
    corrupts only this instance, and is silently reverted on the next record
    for the key or on the next restart.
    """

    def __init__(self) -> None:
        self._nbytes = 0
        self._raw: dict[str, bytes] = {}

    def __contains__(self, key: str) -> bool:
        return key in self._raw

    def __len__(self) -> int:
        return len(self._raw)

    @classmethod
    def of(cls, entries: dict[str, Record]) -> "ConfigStore":
        """Build a pre-seeded store — the test-side entry point."""
        store = cls()
        for key, value in entries.items():
            store._put(key, value)  # one accounting site for `nbytes`
        return store

    def get(self, key: str) -> Config | None:
        """Return the latest config for ``key``, or None if absent.

        The store only ever holds `encode_json` output — `_put` re-encodes
        every value — so a malformed value is impossible by construction. If
        one appears anyway it is a framework bug, and the ``ValueError`` from
        the strict decode crashes rather than laundering it into an empty
        `Config`; an application's policy hook (`Stage.on_invalid_message`)
        already ran on the way in.
        """
        raw = self._raw.get(key)
        return None if raw is None else decode_record(raw, Config)

    @property
    def nbytes(self) -> int:
        """Wire footprint held: the UTF-8 keys plus the encoded values.

        Tracked incrementally, so reading it is O(1). The runners emit it as
        `config_store_bytes` on every store change, and a table large enough
        to be worth watching is exactly the one a `sum()` per observation
        would punish. It weighs the bytes, not the Python objects around
        them: real RSS runs a small multiple higher, and a startup bootstrap
        peaks higher still.
        """
        return self._nbytes

    @staticmethod
    def _weigh(key: str, raw: bytes) -> int:
        return len(key.encode()) + len(raw)

    def _put(self, key: str, config: Record) -> None:
        raw = encode_json(config)
        previous = self._raw.get(key)
        if previous is not None:
            self._nbytes -= self._weigh(key, previous)
        self._nbytes += self._weigh(key, raw)
        self._raw[key] = raw

    def _delete(self, key: str) -> None:
        raw = self._raw.pop(key, None)
        if raw is not None:
            self._nbytes -= self._weigh(key, raw)

nbytes property 

nbytes: int

Wire footprint held: the UTF-8 keys plus the encoded values.

Tracked incrementally, so reading it is O(1). The runners emit it as config_store_bytes on every store change, and a table large enough to be worth watching is exactly the one a sum() per observation would punish. It weighs the bytes, not the Python objects around them: real RSS runs a small multiple higher, and a startup bootstrap peaks higher still.

of classmethod 

of(entries: dict[str, Record]) -> ConfigStore

Build a pre-seeded store — the test-side entry point.

Source code in src/flechtwerk/configs.py 
 95
 96
 97
 98
 99
100
101
@classmethod
def of(cls, entries: dict[str, Record]) -> "ConfigStore":
    """Build a pre-seeded store — the test-side entry point."""
    store = cls()
    for key, value in entries.items():
        store._put(key, value)  # one accounting site for `nbytes`
    return store

get 

get(key: str) -> Config | None

Return the latest config for key, or None if absent.

The store only ever holds encode_json output — _put re-encodes every value — so a malformed value is impossible by construction. If one appears anyway it is a framework bug, and the ValueError from the strict decode crashes rather than laundering it into an empty Config; an application's policy hook (Stage.on_invalid_message) already ran on the way in.

Source code in src/flechtwerk/configs.py 
103
104
105
106
107
108
109
110
111
112
113
114
def get(self, key: str) -> Config | None:
    """Return the latest config for ``key``, or None if absent.

    The store only ever holds `encode_json` output — `_put` re-encodes
    every value — so a malformed value is impossible by construction. If
    one appears anyway it is a framework bug, and the ``ValueError`` from
    the strict decode crashes rather than laundering it into an empty
    `Config`; an application's policy hook (`Stage.on_invalid_message`)
    already ran on the way in.
    """
    raw = self._raw.get(key)
    return None if raw is None else decode_record(raw, Config)

flechtwerk.MqttBrokerConfig dataclass

Shared MQTT connection settings — one MQTT broker serves the whole platform.

Keyword-only by construction. The fields are kept alphabetical (this repo's convention), so any future one lands in the MIDDLE of the list — which under positional construction would silently re-bind a caller's arguments to the wrong fields. kw_only makes field order a non-event: a stale call fails loudly at construction instead of producing a config whose session_expiry is somebody's username.

Defined here rather than in flechtwerk/mqtt.py so the container can annotate its mqtt slot without importing paho: reactor-di evaluates all class annotations at decoration time, and paho must stay an opt-in import confined to flechtwerk/mqtt.py (what makes the flechtwerk[mqtt] optional extra work).

Deliberately MQTT-broker-only: the identity of the instance's persistent MQTT session is the module-wide client_id (see Flechtwerk.of), and the shared-subscription group is the application_id — both injected onto the stage by configured_stage alongside these settings.

session_expiry is the MQTT 5 Session Expiry Interval sent in CONNECT: how long the MQTT broker keeps an offline session — and the QoS ≥ 1 messages queued into it — alive after a disconnect. At a single replica it IS the data-protection window: if it lapses while the stage is away, the session's subscription goes with it and everything published in the gap is lost for good.

It is always sent, and cannot be delegated to the broker. A v5 CONNECT that omits the property gets 0 — session discarded at disconnect, no backlog at all; a broker's own session_expiry_interval applies only to MQTT 3.1.1 clients, which have no way to ask (both verified on EMQX 5.8.9). So the framework has to pick a number, and 24 hours is that pick: long enough to ride out an outage nobody is awake for, and nearly free, because how much an offline session may hold is capped by the broker's per-session queue limit rather than by this value. The two error directions are not symmetric — too short loses data silently and permanently, too long only leaves a session lingering that an operator can kick.

Treat it as an upper bound rather than a promise: the real budget is min(session_expiry, queue limit / message rate) and the queue usually binds first. Shorten it deliberately when running more than one replica, where a lingering session withholds traffic from its survivors — see the MQTT guide's "Sizing the Outage Budget".

qos is the QoS the stage subscribes with — the ceiling at which the MQTT broker delivers to it. 1, the default, is the only level the manual-ACK design is built for: an un-ACKed PUBLISH comes back with its payload when the session resumes. 0 has no ACK and no backlog. 2 is accepted on the wire but NOT supported: paho answers PUBREC on receipt, before the message reaches the stage, and parks it in process memory until PUBREL — after a crash the MQTT broker resends only the PUBREL, which paho then neither delivers nor PUBCOMPs, so an in-flight message is lost instead of redelivered. See the MQTT guide's "Quality of Service".

tls is a stdlib ssl.SSLContext for connecting over TLS; None, the default, connects in plaintext. The caller builds the context, so the caller owns the trust and the policy: ssl.create_default_context() for an MQTT broker with a publicly trusted certificate, create_default_context(cafile=...) for one signed by a private CA, and load_cert_chain(...) on top for mutual TLS — a context that carries a client certificate presents it, and nothing else changes. A context that would not authenticate the MQTT broker (verify_mode other than CERT_REQUIRED, or check_hostname off) is refused here, at construction: encryption without authentication is not offered. The port stays the caller's to set — 8883 is the TLS convention, and no default changes with tls. A context reads its certificate files once, so a rotated certificate takes effect at the next process start. See the MQTT guide's "TLS".

Source code in src/flechtwerk/module.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass(frozen=True, kw_only=True)
class MqttBrokerConfig:
    """Shared MQTT connection settings — one MQTT broker serves the whole platform.

    Keyword-only by construction. The fields are kept alphabetical (this
    repo's convention), so any future one lands in the MIDDLE of the list —
    which under positional construction would silently re-bind a caller's
    arguments to the wrong fields. ``kw_only`` makes field order a
    non-event: a stale call fails loudly at construction instead of
    producing a config whose ``session_expiry`` is somebody's username.

    Defined here rather than in ``flechtwerk/mqtt.py`` so the container can
    annotate its ``mqtt`` slot without importing paho: reactor-di evaluates
    all class annotations at decoration time, and paho must stay an opt-in
    import confined to ``flechtwerk/mqtt.py`` (what makes the ``flechtwerk[mqtt]``
    optional extra work).

    Deliberately MQTT-broker-only: the identity of the instance's persistent MQTT
    session is the module-wide ``client_id`` (see ``Flechtwerk.of``), and
    the shared-subscription group is the ``application_id`` — both injected
    onto the stage by ``configured_stage`` alongside these settings.

    ``session_expiry`` is the MQTT 5 ``Session Expiry Interval`` sent in
    CONNECT: how long the MQTT broker keeps an offline session — and the
    QoS ≥ 1 messages queued into it — alive after a disconnect. At a single
    replica it IS the data-protection window: if it lapses while the stage is
    away, the session's subscription goes with it and everything published in
    the gap is lost for good.

    It is always sent, and cannot be delegated to the broker. A v5 CONNECT
    that omits the property gets **0** — session discarded at disconnect, no
    backlog at all; a broker's own ``session_expiry_interval`` applies only
    to MQTT 3.1.1 clients, which have no way to ask (both verified on EMQX
    5.8.9). So the framework has to pick a number, and 24 hours is that pick:
    long enough to ride out an outage nobody is awake for, and nearly free,
    because how much an offline session may hold is capped by the broker's
    per-session queue limit rather than by this value. The two error
    directions are not symmetric — too short loses data silently and
    permanently, too long only leaves a session lingering that an operator
    can kick.

    Treat it as an upper bound rather than a promise: the real budget is
    ``min(session_expiry, queue limit / message rate)`` and the queue usually
    binds first. Shorten it deliberately when running more than one replica,
    where a lingering session withholds traffic from its survivors — see the
    MQTT guide's "Sizing the Outage Budget".

    ``qos`` is the QoS the stage subscribes with — the ceiling at which the
    MQTT broker delivers to it. 1, the default, is the only level the
    manual-ACK design is built for: an un-ACKed PUBLISH comes back with its
    payload when the session resumes. 0 has no ACK and no backlog. 2 is
    accepted on the wire but NOT supported: paho answers PUBREC on receipt,
    before the message reaches the stage, and parks it in process memory
    until PUBREL — after a crash the MQTT broker resends only the PUBREL,
    which paho then neither delivers nor PUBCOMPs, so an in-flight message
    is lost instead of redelivered. See the MQTT guide's "Quality of
    Service".

    ``tls`` is a stdlib ``ssl.SSLContext`` for connecting over TLS; ``None``,
    the default, connects in plaintext. The caller builds the context, so the
    caller owns the trust and the policy: ``ssl.create_default_context()`` for
    an MQTT broker with a publicly trusted certificate,
    ``create_default_context(cafile=...)`` for one signed by a private CA, and
    ``load_cert_chain(...)`` on top for mutual TLS — a context that carries a
    client certificate presents it, and nothing else changes. A context that
    would not authenticate the MQTT broker (``verify_mode`` other than
    ``CERT_REQUIRED``, or ``check_hostname`` off) is refused here, at
    construction: encryption without authentication is not offered. The port
    stays the caller's to set — 8883 is the TLS convention, and no default
    changes with ``tls``. A context reads its certificate files once, so a
    rotated certificate takes effect at the next process start. See the MQTT
    guide's "TLS".
    """
    broker: str
    port: int
    password: str = ""
    qos: int = 1
    session_expiry: timedelta = timedelta(hours=24)
    tls: ssl.SSLContext | None = None
    username: str = ""

    def __post_init__(self) -> None:
        validate_tls(self.tls, "mqtt.tls")

flechtwerk.CompressionType module-attribute 

CompressionType = Literal['gzip', 'snappy', 'lz4', 'zstd']

Kafka producer compression codec — closed set matching aiokafka's accepted values.

Secrets

The flechtwerk[secrets] optional extra — field-level encryption and signing for attributes. See the Encrypted Secrets concept page for usage and the full spec (wire format, rotation, migration, post-quantum posture). The keyring classes and the token primitives live in the flechtctl package, which the extra installs, and are importable from flechtwerk.keyring and flechtwerk.secrets too — they are the same objects.

flechtwerk.secrets.ENCRYPTED 

ENCRYPTED(
    inner: Codec[V],
    *,
    scope: str = "",
    read_plaintext: bool = False
) -> Codec[V]

Wrap inner so the attribute's value is stored as an encrypted token.

A plain composable codec: ENCRYPTED(LIST(STR)), LIST(ENCRYPTED(STR)), and ENCRYPTED(RECORD) are all valid.

scope (default none) binds the token into a domain-separation compartment — a token stamped with one scope is rejected by a codec declaring a different one. It is a one-way ratchet: a scoped codec still accepts an unscoped token (so a scope can be added to an already-encrypted field without a flag-day, then swept in with reencrypt), but an unscoped codec rejects a scoped token (a scope cannot be silently dropped).

read_plaintext (default off) tolerates a legacy plaintext value on read for this attribute during a migration, logging a WARNING and emitting secret_plaintext_read; off, a plaintext value raises PlaintextSecretError.

Source code in src/flechtwerk/secrets.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def ENCRYPTED[V](inner: Codec[V], *, scope: str = "", read_plaintext: bool = False) -> Codec[V]:
    """Wrap ``inner`` so the attribute's value is stored as an encrypted token.

    A plain composable codec: ``ENCRYPTED(LIST(STR))``, ``LIST(ENCRYPTED(STR))``,
    and ``ENCRYPTED(RECORD)`` are all valid.

    ``scope`` (default none) binds the token into a domain-separation
    compartment — a token stamped with one scope is rejected by a codec
    declaring a different one. It is a **one-way ratchet**: a scoped codec still
    accepts an unscoped token (so a scope can be added to an already-encrypted
    field without a flag-day, then swept in with ``reencrypt``), but an unscoped
    codec rejects a scoped token (a scope cannot be silently dropped).

    ``read_plaintext`` (default off) tolerates a legacy plaintext value on read
    for this attribute during a migration, logging a WARNING and emitting
    ``secret_plaintext_read``; off, a plaintext value raises
    ``PlaintextSecretError``.
    """
    def encode(value: V) -> str:
        return _encrypt_payload(scope, canonical_json(inner.encode(value)))

    def decode(value: object) -> object:
        if is_encrypted(value):
            return inner.decode(json.loads(_decrypt_to_payload(scope, value)))  # type: ignore[arg-type]
        if read_plaintext:
            log.warning("Secret value read as legacy plaintext (scope=%r) — migrate this record to ciphertext", scope)
            active_observer().secret_plaintext_read(scope)
            return inner.decode(value)
        raise PlaintextSecretError(scope=scope)

    return _EncryptedCodec(decode=decode, encode=encode, scope=scope)

flechtwerk.secrets.SIGNED 

SIGNED(
    inner: Codec[V],
    *,
    scope: str,
    read_unsigned: bool = False
) -> Codec[V]

Wrap inner so the attribute's value is stored as an Ed25519-signed token.

The token is flsig:jws:<compact JWS>, signed by the keyring's signing key for scope (signing[scope]). scope is required: a signature always names the compartment it vouches for, and a reader accepts it only for exactly that scope, from any signing key in its keyring bound to it.

Composes like any codec; the recommended composition is sign-then-encrypt, ENCRYPTED(SIGNED(inner, scope=s), scope=s) — the signer's kid stays inside the ciphertext, and the signature's scope keeps a decrypting reader from re-sealing the value into another field.

read_unsigned (default off) tolerates an unsigned value on read during a migration, logging a WARNING and emitting secret_unsigned_read; off, an unsigned value raises UnsignedSecretError.

Source code in src/flechtwerk/secrets.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def SIGNED[V](inner: Codec[V], *, scope: str, read_unsigned: bool = False) -> Codec[V]:
    """Wrap ``inner`` so the attribute's value is stored as an Ed25519-signed token.

    The token is ``flsig:jws:<compact JWS>``, signed by the keyring's signing
    key for ``scope`` (``signing[scope]``). ``scope`` is required: a signature
    always names the compartment it vouches for, and a reader accepts it only
    for exactly that scope, from any signing key in its keyring bound to it.

    Composes like any codec; the recommended composition is sign-then-encrypt,
    ``ENCRYPTED(SIGNED(inner, scope=s), scope=s)`` — the signer's kid stays
    inside the ciphertext, and the signature's scope keeps a decrypting reader
    from re-sealing the value into another field.

    ``read_unsigned`` (default off) tolerates an unsigned value on read during
    a migration, logging a WARNING and emitting ``secret_unsigned_read``; off,
    an unsigned value raises ``UnsignedSecretError``.
    """
    if not scope:
        raise ValueError("SIGNED needs a non-empty scope — a signature always names the compartment it vouches for")

    def encode(value: V) -> str:
        return sign_bytes(active_keyring(), canonical_json(inner.encode(value)), scope=scope)

    def decode(value: object) -> object:
        if is_signed(value):
            return inner.decode(json.loads(_verify_to_payload(scope, value)))  # type: ignore[arg-type]
        if read_unsigned:
            log.warning("Signed value read unsigned (scope=%r) — migrate this record to a signed one", scope)
            active_observer().secret_unsigned_read(scope)
            return inner.decode(value)
        raise UnsignedSecretError(scope=scope)

    return _SignedCodec(decode=decode, encode=encode, scope=scope)

flechtctl.keyring.Keyring dataclass

The keys and which one is primary — pure key material.

Build via Keyring.of(...) (raw bytes) or Keyring.from_json(...) (a JWK Set document). Both validate, raising KeyringError naming the offending kid or scope:

  1. at least one key in keys or recipients;
  2. every kid is a non-empty string, held in only one of keys, recipients and signers;
  3. every oct key is 32 bytes, every recipient / signing key has a 32-byte public and a private that is None or 32 bytes;
  4. primary names a key in keys or recipients;
  5. every kid in scopes is held, bound to a non-empty set of non-empty scope names;
  6. every recipient and signing kid is bound — public-key material always is; an oct key may be or not;
  7. every primaries entry maps a non-empty scope to a kid in keys or recipients whose binding, if any, covers that scope;
  8. every signing entry maps a non-empty scope to a signing kid holding a private half, whose binding covers that scope.
Source code in flechtctl/src/flechtctl/keyring.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
@dataclass(frozen=True)
class Keyring:
    """The keys and which one is primary — pure key material.

    Build via `Keyring.of(...)` (raw bytes) or `Keyring.from_json(...)` (a
    JWK Set document). Both validate, raising `KeyringError` naming the
    offending kid or scope:

    1. at least one key in `keys` or `recipients`;
    2. every kid is a non-empty string, held in only one of `keys`,
       `recipients` and `signers`;
    3. every oct key is 32 bytes, every recipient / signing key has a 32-byte
       `public` and a `private` that is `None` or 32 bytes;
    4. `primary` names a key in `keys` or `recipients`;
    5. every kid in `scopes` is held, bound to a non-empty set of non-empty
       scope names;
    6. every recipient and signing kid is bound — public-key material always
       is; an oct key may be or not;
    7. every `primaries` entry maps a non-empty scope to a kid in `keys` or
       `recipients` whose binding, if any, covers that scope;
    8. every `signing` entry maps a non-empty scope to a signing kid holding a
       `private` half, whose binding covers that scope.
    """

    keys: dict[str, bytes]
    primary: str
    primaries: dict[str, str] = field(default_factory=dict)
    recipients: dict[str, X25519Key] = field(default_factory=dict)
    scopes: dict[str, frozenset[str]] = field(default_factory=dict)
    signers: dict[str, Ed25519Key] = field(default_factory=dict)
    signing: dict[str, str] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "scopes", {
            kid: bound if isinstance(bound, str) else frozenset(bound) for kid, bound in self.scopes.items()
        })
        if not self.keys and not self.recipients:
            raise KeyringError("a keyring needs at least one key")
        kinds = (self.keys, self.recipients, self.signers)
        held = [kid for kind in kinds for kid in kind]
        for kid in held:
            if not isinstance(kid, str) or not kid:
                raise KeyringError("a key id must be a non-empty string")
            if held.count(kid) > 1:
                raise KeyringError(f"kid {kid!r} is held as more than one kind of key")
        for kid, key in self.keys.items():
            if len(key) != KEY_BYTES:
                raise KeyringError(f"key {kid!r} is {len(key)} bytes, expected {KEY_BYTES} (AES-256)")
        for kind, pairs in (("X25519", self.recipients), ("Ed25519", self.signers)):
            for kid, pair in pairs.items():
                if len(pair.public) != KEY_BYTES:
                    raise KeyringError(f"{kind} key {kid!r} has a {len(pair.public)}-byte public half, expected {KEY_BYTES}")
                if pair.private is not None and len(pair.private) != KEY_BYTES:
                    raise KeyringError(f"{kind} key {kid!r} has a {len(pair.private)}-byte private half, expected {KEY_BYTES}")
        if self.primary not in self.keys and self.primary not in self.recipients:
            raise KeyringError(
                f"primary {self.primary!r} is not among the keys {sorted(self.keys) + sorted(self.recipients)}"
            )
        for kid, bound in self.scopes.items():
            if kid not in held:
                raise KeyringError(f"scopes name kid {kid!r}, which the keyring does not hold")
            if (not isinstance(bound, frozenset) or not bound
                    or not all(isinstance(scope, str) and scope for scope in bound)):
                raise KeyringError(f"key {kid!r} must be bound to a non-empty set of non-empty scope names")
        for kid in [*self.recipients, *self.signers]:
            if kid not in self.scopes:
                raise KeyringError(f"public-key kid {kid!r} must be bound to the scopes it serves")
        for scope, kid in self.primaries.items():
            if not isinstance(scope, str) or not scope:
                raise KeyringError(f"primaries map an empty scope to {kid!r}")
            if kid not in self.keys and kid not in self.recipients:
                raise KeyringError(f"primaries map scope {scope!r} to {kid!r}, which is not a sealing key")
            if kid in self.scopes and scope not in self.scopes[kid]:
                raise KeyringError(f"primaries map scope {scope!r} to {kid!r}, which is not bound to it")
        for scope, kid in self.signing.items():
            if not isinstance(scope, str) or not scope:
                raise KeyringError(f"signing maps an empty scope to {kid!r}")
            if kid not in self.signers or self.signers[kid].private is None:
                raise KeyringError(f"signing maps scope {scope!r} to {kid!r}, which is not a private signing key")
            if scope not in self.scopes[kid]:
                raise KeyringError(f"signing maps scope {scope!r} to {kid!r}, which is not bound to it")

    @classmethod
    def of(
        cls,
        keys: dict[str, bytes],
        *,
        primary: str,
        primaries: dict[str, str] | None = None,
        recipients: dict[str, X25519Key] | None = None,
        scopes: Mapping[str, Set[str]] | None = None,
        signers: dict[str, Ed25519Key] | None = None,
        signing: dict[str, str] | None = None,
    ) -> "Keyring":
        """Build a keyring from raw key bytes — the programmatic / test entry point."""
        return cls(
            keys=dict(keys),
            primary=primary,
            primaries=dict(primaries or {}),
            recipients=dict(recipients or {}),
            # `__post_init__` normalizes every bound to a frozenset.
            scopes=dict(scopes or {}),  # pyright: ignore[reportArgumentType]
            signers=dict(signers or {}),
            signing=dict(signing or {}),
        )

    @classmethod
    def from_json(cls, text: str) -> "Keyring":
        """Parse an RFC 7517 JWK Set with a `primary` extension member.

        Accepts oct keys with an unpadded-base64url `k`, and OKP keys with
        `crv` `X25519` (recipients) or `Ed25519` (signers), a public `x` and an
        optional private `d`; every key needs a `kid`. A key's optional
        `flenc_scopes` member (a non-empty array of scope names) binds it —
        required for OKP keys. Two optional top-level members map scopes to
        kids: `flenc_primaries` (the sealing key per scope) and `flenc_signing`
        (the signing key per scope). Unknown members (top-level and per-key)
        are ignored, per JWK. This is the deployment entry point — the
        application reads the document from wherever it lives (a mounted
        secret, a file, an env var it reads itself) and hands the text in.
        """
        if is_encrypted_keyring(text):
            raise KeyringError(
                "keyring document is password-protected — open it with flechtctl (decrypt_keyring) "
                "or write a plain copy with flechtctl keyring unprotect"
            )
        try:
            doc = json.loads(text)
        except json.JSONDecodeError as e:
            raise KeyringError(f"keyring document is not valid JSON: {e}") from e
        if not isinstance(doc, dict):
            raise KeyringError(f"keyring document must be a JSON object, got {type(doc).__name__}")
        raw_keys = doc.get("keys", [])
        if not isinstance(raw_keys, list):
            raise KeyringError("keyring 'keys' must be a JSON array")
        keys: dict[str, bytes] = {}
        recipients: dict[str, X25519Key] = {}
        scopes: dict[str, frozenset[str]] = {}
        signers: dict[str, Ed25519Key] = {}
        for jwk in raw_keys:
            kid, material, bound = _parse_jwk(jwk, held=frozenset([*keys, *recipients, *signers]))
            if bound is not None:
                scopes[kid] = bound
            if isinstance(material, X25519Key):
                recipients[kid] = material
            elif isinstance(material, Ed25519Key):
                signers[kid] = material
            else:
                keys[kid] = material
        if "primary" not in doc:
            raise KeyringError("keyring document has no top-level 'primary' member")
        return cls(
            keys=keys,
            primary=doc["primary"],
            primaries=_parse_scope_map(doc, "flenc_primaries"),
            recipients=recipients,
            scopes=scopes,
            signers=signers,
            signing=_parse_scope_map(doc, "flenc_signing"),
        )

    def to_json(self) -> str:
        """The inverse of `from_json`: this keyring as an RFC 7517 JWK Set document.

        Keys sorted by `kid`, members sorted, two-space indent, newline-terminated;
        `Keyring.from_json(k.to_json()) == k` for every valid keyring. The private
        halves are included — this is the file form, not a shareable one (see `public_jwk`).
        """
        jwks: list[dict[str, Any]] = [
            {"k": _encode_key(key), "kid": kid, "kty": "oct"} for kid, key in self.keys.items()
        ] + [
            _okp_jwk("X25519", kid, key) for kid, key in self.recipients.items()
        ] + [
            _okp_jwk("Ed25519", kid, key) for kid, key in self.signers.items()
        ]
        for jwk in jwks:
            if jwk["kid"] in self.scopes:
                jwk["flenc_scopes"] = sorted(self.scopes[jwk["kid"]])
        doc: dict = {"keys": sorted(jwks, key=lambda jwk: jwk["kid"]), "primary": self.primary}
        if self.primaries:
            doc["flenc_primaries"] = dict(self.primaries)
        if self.signing:
            doc["flenc_signing"] = dict(self.signing)
        return json.dumps(doc, indent=2, sort_keys=True) + "\n"

    def public_jwk(self, kid: str) -> dict:
        """The public half of entry `kid` as one JWK — never the private half.

        What `flechtctl keyring export` prints: hand it to the party that should be
        able to seal (an X25519 recipient) or to check signatures (an Ed25519
        signer). The keyring itself is unchanged. An oct entry has no public half.
        """
        if kid in self.keys:
            raise KeyringError(f"key {kid!r} is a symmetric (oct) key and has no public half")
        if kid in self.recipients:
            jwk = _okp_jwk("X25519", kid, X25519Key(public=self.recipients[kid].public))
        elif kid in self.signers:
            jwk = _okp_jwk("Ed25519", kid, Ed25519Key(public=self.signers[kid].public))
        else:
            raise UnknownKeyError(kid)
        jwk["flenc_scopes"] = sorted(self.scopes[kid])
        return jwk

    def with_jwk(self, jwk: dict) -> "Keyring":
        """A new keyring with one more entry, parsed from a single JWK.

        The private half is kept if the JWK carries one, and so are its scopes.
        A kid already held raises `KeyringError`; the constructor's validation runs.
        """
        kid, material, bound = _parse_jwk(jwk, held=frozenset(self.kids()))
        return replace(
            self,
            keys={**self.keys, kid: material} if isinstance(material, bytes) else self.keys,
            recipients={**self.recipients, kid: material} if isinstance(material, X25519Key) else self.recipients,
            scopes={**self.scopes, kid: bound} if bound is not None else self.scopes,
            signers={**self.signers, kid: material} if isinstance(material, Ed25519Key) else self.signers,
        )

    def key_for(self, kid: str) -> bytes:
        """Return the oct key bytes for `kid`, or raise `UnknownKeyError`.

        A recipient or signing kid is unknown here: this lookup serves `dir`
        tokens only.
        """
        try:
            return self.keys[kid]
        except KeyError:
            raise UnknownKeyError(kid) from None

    def recipient_for(self, kid: str) -> X25519Key:
        """Return the X25519 recipient key for `kid`, or raise `UnknownKeyError`."""
        try:
            return self.recipients[kid]
        except KeyError:
            raise UnknownKeyError(kid) from None

    def signer_for(self, kid: str) -> Ed25519Key:
        """Return the Ed25519 signing key for `kid`, or raise `UnknownKeyError`."""
        try:
            return self.signers[kid]
        except KeyError:
            raise UnknownKeyError(kid) from None

    def primary_pair(self) -> tuple[str, bytes]:
        """Return `(primary_kid, primary_key_bytes)` — what `dir` encryption stamps and uses.

        Raises `KeyringError` when the primary is a recipient key, which has no
        symmetric bytes to hand out.
        """
        if self.primary in self.recipients:
            raise KeyringError(f"primary {self.primary!r} is a recipient key, not a symmetric one")
        return self.primary, self.keys[self.primary]

    def sealing_kid(self, scope: str) -> str:
        """The kid that seals a value of `scope`: `primaries[scope]`, else `primary`.

        Raises `KeyringError` when that kid is bound and `scope` is outside its
        binding — an unscoped value (`scope == ""`) is outside every binding.
        """
        kid = self.primaries.get(scope, self.primary)
        bound = self.scopes.get(kid)
        if bound is not None and scope not in bound:
            what = f"scope {scope!r}" if scope else "an unscoped value"
            raise KeyringError(f"key {kid!r} is bound to scopes {sorted(bound)} and cannot seal {what}")
        return kid

    def signing_kid(self, scope: str) -> str:
        """The kid that signs a value of `scope`, or raise `KeyringError` when none is configured."""
        try:
            return self.signing[scope]
        except KeyError:
            raise KeyringError(f"no signing key for scope {scope!r} in the active keyring") from None

    def kids(self) -> list[str]:
        """Loaded key ids of every kind, sorted — the labels for the startup keyring gauge."""
        return sorted([*self.keys, *self.recipients, *self.signers])

of classmethod 

of(
    keys: dict[str, bytes],
    *,
    primary: str,
    primaries: dict[str, str] | None = None,
    recipients: dict[str, X25519Key] | None = None,
    scopes: Mapping[str, Set[str]] | None = None,
    signers: dict[str, Ed25519Key] | None = None,
    signing: dict[str, str] | None = None
) -> Keyring

Build a keyring from raw key bytes — the programmatic / test entry point.

Source code in flechtctl/src/flechtctl/keyring.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
@classmethod
def of(
    cls,
    keys: dict[str, bytes],
    *,
    primary: str,
    primaries: dict[str, str] | None = None,
    recipients: dict[str, X25519Key] | None = None,
    scopes: Mapping[str, Set[str]] | None = None,
    signers: dict[str, Ed25519Key] | None = None,
    signing: dict[str, str] | None = None,
) -> "Keyring":
    """Build a keyring from raw key bytes — the programmatic / test entry point."""
    return cls(
        keys=dict(keys),
        primary=primary,
        primaries=dict(primaries or {}),
        recipients=dict(recipients or {}),
        # `__post_init__` normalizes every bound to a frozenset.
        scopes=dict(scopes or {}),  # pyright: ignore[reportArgumentType]
        signers=dict(signers or {}),
        signing=dict(signing or {}),
    )

from_json classmethod 

from_json(text: str) -> Keyring

Parse an RFC 7517 JWK Set with a primary extension member.

Accepts oct keys with an unpadded-base64url k, and OKP keys with crv X25519 (recipients) or Ed25519 (signers), a public x and an optional private d; every key needs a kid. A key's optional flenc_scopes member (a non-empty array of scope names) binds it — required for OKP keys. Two optional top-level members map scopes to kids: flenc_primaries (the sealing key per scope) and flenc_signing (the signing key per scope). Unknown members (top-level and per-key) are ignored, per JWK. This is the deployment entry point — the application reads the document from wherever it lives (a mounted secret, a file, an env var it reads itself) and hands the text in.

Source code in flechtctl/src/flechtctl/keyring.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
@classmethod
def from_json(cls, text: str) -> "Keyring":
    """Parse an RFC 7517 JWK Set with a `primary` extension member.

    Accepts oct keys with an unpadded-base64url `k`, and OKP keys with
    `crv` `X25519` (recipients) or `Ed25519` (signers), a public `x` and an
    optional private `d`; every key needs a `kid`. A key's optional
    `flenc_scopes` member (a non-empty array of scope names) binds it —
    required for OKP keys. Two optional top-level members map scopes to
    kids: `flenc_primaries` (the sealing key per scope) and `flenc_signing`
    (the signing key per scope). Unknown members (top-level and per-key)
    are ignored, per JWK. This is the deployment entry point — the
    application reads the document from wherever it lives (a mounted
    secret, a file, an env var it reads itself) and hands the text in.
    """
    if is_encrypted_keyring(text):
        raise KeyringError(
            "keyring document is password-protected — open it with flechtctl (decrypt_keyring) "
            "or write a plain copy with flechtctl keyring unprotect"
        )
    try:
        doc = json.loads(text)
    except json.JSONDecodeError as e:
        raise KeyringError(f"keyring document is not valid JSON: {e}") from e
    if not isinstance(doc, dict):
        raise KeyringError(f"keyring document must be a JSON object, got {type(doc).__name__}")
    raw_keys = doc.get("keys", [])
    if not isinstance(raw_keys, list):
        raise KeyringError("keyring 'keys' must be a JSON array")
    keys: dict[str, bytes] = {}
    recipients: dict[str, X25519Key] = {}
    scopes: dict[str, frozenset[str]] = {}
    signers: dict[str, Ed25519Key] = {}
    for jwk in raw_keys:
        kid, material, bound = _parse_jwk(jwk, held=frozenset([*keys, *recipients, *signers]))
        if bound is not None:
            scopes[kid] = bound
        if isinstance(material, X25519Key):
            recipients[kid] = material
        elif isinstance(material, Ed25519Key):
            signers[kid] = material
        else:
            keys[kid] = material
    if "primary" not in doc:
        raise KeyringError("keyring document has no top-level 'primary' member")
    return cls(
        keys=keys,
        primary=doc["primary"],
        primaries=_parse_scope_map(doc, "flenc_primaries"),
        recipients=recipients,
        scopes=scopes,
        signers=signers,
        signing=_parse_scope_map(doc, "flenc_signing"),
    )

to_json 

to_json() -> str

The inverse of from_json: this keyring as an RFC 7517 JWK Set document.

Keys sorted by kid, members sorted, two-space indent, newline-terminated; Keyring.from_json(k.to_json()) == k for every valid keyring. The private halves are included — this is the file form, not a shareable one (see public_jwk).

Source code in flechtctl/src/flechtctl/keyring.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def to_json(self) -> str:
    """The inverse of `from_json`: this keyring as an RFC 7517 JWK Set document.

    Keys sorted by `kid`, members sorted, two-space indent, newline-terminated;
    `Keyring.from_json(k.to_json()) == k` for every valid keyring. The private
    halves are included — this is the file form, not a shareable one (see `public_jwk`).
    """
    jwks: list[dict[str, Any]] = [
        {"k": _encode_key(key), "kid": kid, "kty": "oct"} for kid, key in self.keys.items()
    ] + [
        _okp_jwk("X25519", kid, key) for kid, key in self.recipients.items()
    ] + [
        _okp_jwk("Ed25519", kid, key) for kid, key in self.signers.items()
    ]
    for jwk in jwks:
        if jwk["kid"] in self.scopes:
            jwk["flenc_scopes"] = sorted(self.scopes[jwk["kid"]])
    doc: dict = {"keys": sorted(jwks, key=lambda jwk: jwk["kid"]), "primary": self.primary}
    if self.primaries:
        doc["flenc_primaries"] = dict(self.primaries)
    if self.signing:
        doc["flenc_signing"] = dict(self.signing)
    return json.dumps(doc, indent=2, sort_keys=True) + "\n"

public_jwk 

public_jwk(kid: str) -> dict

The public half of entry kid as one JWK — never the private half.

What flechtctl keyring export prints: hand it to the party that should be able to seal (an X25519 recipient) or to check signatures (an Ed25519 signer). The keyring itself is unchanged. An oct entry has no public half.

Source code in flechtctl/src/flechtctl/keyring.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def public_jwk(self, kid: str) -> dict:
    """The public half of entry `kid` as one JWK — never the private half.

    What `flechtctl keyring export` prints: hand it to the party that should be
    able to seal (an X25519 recipient) or to check signatures (an Ed25519
    signer). The keyring itself is unchanged. An oct entry has no public half.
    """
    if kid in self.keys:
        raise KeyringError(f"key {kid!r} is a symmetric (oct) key and has no public half")
    if kid in self.recipients:
        jwk = _okp_jwk("X25519", kid, X25519Key(public=self.recipients[kid].public))
    elif kid in self.signers:
        jwk = _okp_jwk("Ed25519", kid, Ed25519Key(public=self.signers[kid].public))
    else:
        raise UnknownKeyError(kid)
    jwk["flenc_scopes"] = sorted(self.scopes[kid])
    return jwk

with_jwk 

with_jwk(jwk: dict) -> Keyring

A new keyring with one more entry, parsed from a single JWK.

The private half is kept if the JWK carries one, and so are its scopes. A kid already held raises KeyringError; the constructor's validation runs.

Source code in flechtctl/src/flechtctl/keyring.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def with_jwk(self, jwk: dict) -> "Keyring":
    """A new keyring with one more entry, parsed from a single JWK.

    The private half is kept if the JWK carries one, and so are its scopes.
    A kid already held raises `KeyringError`; the constructor's validation runs.
    """
    kid, material, bound = _parse_jwk(jwk, held=frozenset(self.kids()))
    return replace(
        self,
        keys={**self.keys, kid: material} if isinstance(material, bytes) else self.keys,
        recipients={**self.recipients, kid: material} if isinstance(material, X25519Key) else self.recipients,
        scopes={**self.scopes, kid: bound} if bound is not None else self.scopes,
        signers={**self.signers, kid: material} if isinstance(material, Ed25519Key) else self.signers,
    )

key_for 

key_for(kid: str) -> bytes

Return the oct key bytes for kid, or raise UnknownKeyError.

A recipient or signing kid is unknown here: this lookup serves dir tokens only.

Source code in flechtctl/src/flechtctl/keyring.py 
324
325
326
327
328
329
330
331
332
333
def key_for(self, kid: str) -> bytes:
    """Return the oct key bytes for `kid`, or raise `UnknownKeyError`.

    A recipient or signing kid is unknown here: this lookup serves `dir`
    tokens only.
    """
    try:
        return self.keys[kid]
    except KeyError:
        raise UnknownKeyError(kid) from None

recipient_for 

recipient_for(kid: str) -> X25519Key

Return the X25519 recipient key for kid, or raise UnknownKeyError.

Source code in flechtctl/src/flechtctl/keyring.py 
335
336
337
338
339
340
def recipient_for(self, kid: str) -> X25519Key:
    """Return the X25519 recipient key for `kid`, or raise `UnknownKeyError`."""
    try:
        return self.recipients[kid]
    except KeyError:
        raise UnknownKeyError(kid) from None

signer_for 

signer_for(kid: str) -> Ed25519Key

Return the Ed25519 signing key for kid, or raise UnknownKeyError.

Source code in flechtctl/src/flechtctl/keyring.py 
342
343
344
345
346
347
def signer_for(self, kid: str) -> Ed25519Key:
    """Return the Ed25519 signing key for `kid`, or raise `UnknownKeyError`."""
    try:
        return self.signers[kid]
    except KeyError:
        raise UnknownKeyError(kid) from None

primary_pair 

primary_pair() -> tuple[str, bytes]

Return (primary_kid, primary_key_bytes) — what dir encryption stamps and uses.

Raises KeyringError when the primary is a recipient key, which has no symmetric bytes to hand out.

Source code in flechtctl/src/flechtctl/keyring.py 
349
350
351
352
353
354
355
356
357
def primary_pair(self) -> tuple[str, bytes]:
    """Return `(primary_kid, primary_key_bytes)` — what `dir` encryption stamps and uses.

    Raises `KeyringError` when the primary is a recipient key, which has no
    symmetric bytes to hand out.
    """
    if self.primary in self.recipients:
        raise KeyringError(f"primary {self.primary!r} is a recipient key, not a symmetric one")
    return self.primary, self.keys[self.primary]

sealing_kid 

sealing_kid(scope: str) -> str

The kid that seals a value of scope: primaries[scope], else primary.

Raises KeyringError when that kid is bound and scope is outside its binding — an unscoped value (scope == "") is outside every binding.

Source code in flechtctl/src/flechtctl/keyring.py 
359
360
361
362
363
364
365
366
367
368
369
370
def sealing_kid(self, scope: str) -> str:
    """The kid that seals a value of `scope`: `primaries[scope]`, else `primary`.

    Raises `KeyringError` when that kid is bound and `scope` is outside its
    binding — an unscoped value (`scope == ""`) is outside every binding.
    """
    kid = self.primaries.get(scope, self.primary)
    bound = self.scopes.get(kid)
    if bound is not None and scope not in bound:
        what = f"scope {scope!r}" if scope else "an unscoped value"
        raise KeyringError(f"key {kid!r} is bound to scopes {sorted(bound)} and cannot seal {what}")
    return kid

signing_kid 

signing_kid(scope: str) -> str

The kid that signs a value of scope, or raise KeyringError when none is configured.

Source code in flechtctl/src/flechtctl/keyring.py 
372
373
374
375
376
377
def signing_kid(self, scope: str) -> str:
    """The kid that signs a value of `scope`, or raise `KeyringError` when none is configured."""
    try:
        return self.signing[scope]
    except KeyError:
        raise KeyringError(f"no signing key for scope {scope!r} in the active keyring") from None

kids 

kids() -> list[str]

Loaded key ids of every kind, sorted — the labels for the startup keyring gauge.

Source code in flechtctl/src/flechtctl/keyring.py 
379
380
381
def kids(self) -> list[str]:
    """Loaded key ids of every kind, sorted — the labels for the startup keyring gauge."""
    return sorted([*self.keys, *self.recipients, *self.signers])

flechtctl.keyring.X25519Key dataclass

An X25519 recipient key: public seals, private opens.

private=None makes a seal-only key — what a writer holds.

Source code in flechtctl/src/flechtctl/keyring.py 
 95
 96
 97
 98
 99
100
101
102
103
@dataclass(frozen=True)
class X25519Key:
    """An X25519 recipient key: `public` seals, `private` opens.

    `private=None` makes a seal-only key — what a writer holds.
    """

    public: bytes
    private: bytes | None = field(default=None, repr=False)

flechtctl.keyring.Ed25519Key dataclass

An Ed25519 signing key: private (the 32-byte seed) signs, public verifies.

private=None makes a verify-only key — what a reader holds.

Source code in flechtctl/src/flechtctl/keyring.py 
84
85
86
87
88
89
90
91
92
@dataclass(frozen=True)
class Ed25519Key:
    """An Ed25519 signing key: `private` (the 32-byte seed) signs, `public` verifies.

    `private=None` makes a verify-only key — what a reader holds.
    """

    public: bytes
    private: bytes | None = field(default=None, repr=False)

flechtwerk.secrets.encrypt_value 

encrypt_value(attribute: Attribute[V], value: V) -> str

Encrypt value for attribute and return the flenc:jwe: token.

The token a typed write (config[attribute] = value) would store, for writers that build their JSON without a Record — a form backend, say. Raises SecretError if attribute is not an encrypted attribute.

Source code in src/flechtwerk/secrets.py 
260
261
262
263
264
265
266
267
def encrypt_value[V](attribute: Attribute[V], value: V) -> str:
    """Encrypt ``value`` for ``attribute`` and return the ``flenc:jwe:`` token.

    The token a typed write (``config[attribute] = value``) would store, for
    writers that build their JSON without a ``Record`` — a form backend, say.
    Raises ``SecretError`` if ``attribute`` is not an encrypted attribute.
    """
    return _require_encrypted(attribute).encode(value)

flechtwerk.secrets.sign_value 

sign_value(attribute: Attribute[V], value: V) -> str

Sign value for attribute and return the flsig:jws: token.

The counterpart of encrypt_value for a SIGNED attribute — for an ENCRYPTED(SIGNED(...)) attribute use encrypt_value. Raises SecretError if attribute is not a signed attribute.

Source code in src/flechtwerk/secrets.py 
270
271
272
273
274
275
276
277
278
279
280
def sign_value[V](attribute: Attribute[V], value: V) -> str:
    """Sign ``value`` for ``attribute`` and return the ``flsig:jws:`` token.

    The counterpart of `encrypt_value` for a ``SIGNED`` attribute — for an
    ``ENCRYPTED(SIGNED(...))`` attribute use `encrypt_value`. Raises ``SecretError`` if ``attribute`` is
    not a signed attribute.
    """
    codec = attribute.codec
    if not isinstance(codec, _SignedCodec):
        raise SecretError(f"{attribute!r} is not a signed attribute (its codec is not SIGNED)")
    return codec.encode(value)

flechtwerk.secrets.reencrypt 

reencrypt(token: str, attribute: Attribute[V]) -> str

Re-encrypt token under the scope's current sealing key (decrypt-with-named-kid).

The building block for a rotation sweep and for cross-environment promotion: it decrypts with the key the token names, then re-encrypts the same payload under the keyring's sealing key for the attribute's scope (primaries[scope], else primary) — which moves a token from dir to a recipient key, from one recipient key to another, and back. Opening a recipient token needs that recipient's private half. The exact inner bytes are preserved, never round-tripped through the decoded value. Also promotes an unscoped token to the attribute's scope (the ratchet's upgrade path); it cannot strip a scope (an unscoped attribute's decrypt rejects a scoped token). Raises if attribute is not encrypted, or SecretDecryptError if the token cannot be read.

Source code in src/flechtwerk/secrets.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def reencrypt[V](token: str, attribute: Attribute[V]) -> str:
    """Re-encrypt ``token`` under the scope's current sealing key (decrypt-with-named-kid).

    The building block for a rotation sweep and for cross-environment
    promotion: it decrypts with the key the token names, then re-encrypts the
    same payload under the keyring's sealing key for the attribute's scope
    (``primaries[scope]``, else ``primary``) — which moves a token from ``dir``
    to a recipient key, from one recipient key to another, and back. Opening a
    recipient token needs that recipient's private half. The exact inner bytes
    are preserved, never round-tripped through the decoded value. Also promotes an unscoped token to
    the attribute's ``scope`` (the ratchet's upgrade path); it cannot strip a
    scope (an unscoped attribute's decrypt rejects a scoped token). Raises if
    ``attribute`` is not encrypted, or ``SecretDecryptError`` if the token
    cannot be read.
    """
    scope = _require_encrypted(attribute).scope
    return _encrypt_payload(scope, _decrypt_to_payload(scope, token))

flechtwerk.secrets.scan_config_topics async 

scan_config_topics(
    consumer: AIOKafkaConsumer,
    topics: Iterable[str],
    attributes: Iterable[Attribute],
) -> AsyncIterator[ScanEntry]

Yield a ScanEntry for each secret-bearing value across topics.

The engine behind the migration and rotation topic scans: for every surviving record it reports which secret attributes carry ciphertext (with the kid), which still carry plaintext (kid=None, error=None), and which carry a flenc value whose kid could not be read (error set). Reads each topic to its end group-less on the given consumer, exactly as the config bootstrap does. Config topics are small by contract, so results are gathered then yielded.

Since the scan is the authoritative gate before destructive steps (removing a key, ending plaintext tolerance), it must produce a COMPLETE report: a topic that does not exist raises SecretError rather than silently contributing zero records — a missing topic must not read as a clean "no plaintext / no old-kid" result — and a single corrupt token is reported as a ScanEntry with error set rather than aborting the whole scan. That extends to record-level decode: an undecodable key or value becomes a ScanEntry with an empty attribute. The scan has no stage, so it never consults Stage.on_invalid_message — a diagnostic tool must report, not adopt a runtime policy that could crash it.

Source code in src/flechtwerk/secrets.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
async def scan_config_topics(
    consumer: AIOKafkaConsumer,
    topics: Iterable[str],
    attributes: Iterable[Attribute],
) -> AsyncIterator[ScanEntry]:
    """Yield a ``ScanEntry`` for each secret-bearing value across ``topics``.

    The engine behind the migration and rotation topic scans: for every
    surviving record it reports which secret attributes carry ciphertext (with
    the ``kid``), which still carry plaintext (``kid=None``, ``error=None``),
    and which carry a ``flenc`` value whose ``kid`` could not be read
    (``error`` set). Reads each topic to its end group-less on the given
    consumer, exactly as the config bootstrap does. Config topics are small by
    contract, so results are gathered then yielded.

    Since the scan is the authoritative gate before destructive steps (removing
    a key, ending plaintext tolerance), it must produce a COMPLETE report: a
    topic that does not exist raises `SecretError` rather than silently
    contributing zero records — a missing topic must not read as a clean "no
    plaintext / no old-kid" result — and a single corrupt token is reported as
    a `ScanEntry` with ``error`` set rather than aborting the whole scan. That
    extends to record-level decode: an undecodable key or value becomes a
    `ScanEntry` with an empty ``attribute``. The scan has no stage, so it never
    consults `Stage.on_invalid_message` — a diagnostic tool must report, not
    adopt a runtime policy that could crash it.
    """
    names = [a.name for a in attributes]
    topics = list(topics)
    found: list[ScanEntry] = []

    def report_record(msg, wire_key: str, error: ValueError) -> None:
        log.warning("Undecodable config record at %s/%d (key %s): %s",
                    msg.topic, msg.offset, wire_key, error)
        found.append(ScanEntry(
            attribute="",
            error=str(error),
            kid=None,
            partition=msg.partition,
            topic=msg.topic,
            wire_key=wire_key,
        ))

    async def collect(msg) -> None:
        if is_tombstone(msg.value):
            return
        try:
            wire_key = decode_key(msg.key)
        except ValueError as e:
            # An undecodable key has no readable identity, so the report falls
            # back to the raw bytes' repr — lossy-decoding it would invent an
            # identity, which is exactly what the runtime path refuses to do.
            report_record(msg, repr(msg.key), e)
            return
        try:
            raw = decode_record(msg.value, Config).raw
        except ValueError as e:
            report_record(msg, wire_key, e)
            return
        for name in names:
            value = raw.get(name)
            if value is None:
                # Absent, or present-but-null — no secret value either way, so
                # not a plaintext leak (`kid=None` is reserved for a value that
                # IS present and unencrypted).
                continue
            kid: str | None = None
            error: str | None = None
            if is_encrypted(value):
                try:
                    kid = kid_of(value)
                except SecretFormatError as e:
                    # A corrupt ciphertext token: report it, don't abort the
                    # scan. The scan is the gate before destructive steps, so a
                    # complete report across every record matters more than
                    # failing fast on the first bad one.
                    log.warning("Unreadable %s token at %s/%d key %r: %s",
                                SCHEME, msg.topic, msg.offset, wire_key, e)
                    error = str(e)
            found.append(ScanEntry(
                attribute=name,
                error=error,
                kid=kid,
                partition=msg.partition,
                topic=msg.topic,
                wire_key=wire_key,
            ))

    tps = await topic_partitions(consumer, topics)
    present = {tp.topic for tp in tps}
    if missing := [t for t in topics if t not in present]:
        raise SecretError(f"cannot scan — config topic(s) not found: {sorted(missing)}")
    await read_to_end(consumer, tps, collect)
    for entry in found:
        yield entry

flechtctl

The standalone flechtctl distribution — keyrings and secret tokens over an explicit keyring, plus the flechtctl command. It depends on joserfc only and never imports Flechtwerk. See the guide Keyrings and Secret Tokens with flechtctl. The entries below are documented at their defining modules; flechtctl re-exports every name, and flechtwerk.keyring / flechtwerk.secrets re-export most of them.

Keyring files

flechtctl.keyring.keyring_from_jwk 

keyring_from_jwk(jwk: dict) -> Keyring

A one-entry keyring whose primary is that entry — a new keyring file made from an imported JWK.

An Ed25519 entry cannot be primary, so the constructor's own validation rejects it.

Source code in flechtctl/src/flechtctl/keyring.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
def keyring_from_jwk(jwk: dict) -> Keyring:
    """A one-entry keyring whose `primary` is that entry — a new keyring file made from an imported JWK.

    An Ed25519 entry cannot be primary, so the constructor's own validation rejects it.
    """
    kid, material, bound = _parse_jwk(jwk, held=frozenset())
    return Keyring(
        keys={kid: material} if isinstance(material, bytes) else {},
        primary=kid,
        recipients={kid: material} if isinstance(material, X25519Key) else {},
        scopes={kid: bound} if bound is not None else {},
        signers={kid: material} if isinstance(material, Ed25519Key) else {},
    )

flechtctl.keyring.is_encrypted_keyring 

is_encrypted_keyring(text: str) -> bool

Whether text is a password-protected keyring: a compact JWE whose cty is jwk-set+json.

Never raises — anything else (a plain JWK Set, garbage, an ordinary token) is False.

Source code in flechtctl/src/flechtctl/keyring.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
def is_encrypted_keyring(text: str) -> bool:
    """Whether `text` is a password-protected keyring: a compact JWE whose `cty` is `jwk-set+json`.

    Never raises — anything else (a plain JWK Set, garbage, an ordinary token) is `False`.
    """
    try:
        segments = text.strip().split(".")
        if len(segments) != 5:
            return False
        header = json.loads(_decode_key(segments[0], "header"))
        return isinstance(header, dict) and header.get("cty") == "jwk-set+json"
    except (AttributeError, TypeError, ValueError):  # JSONDecodeError and binascii.Error ⊂ ValueError
        return False

flechtctl.tokens.encrypt_keyring 

encrypt_keyring(keyring: Keyring, password: str) -> str

The keyring as a password-protected document: an encrypted JWK Set (RFC 7517 §7).

A compact JWE over keyring.to_json(), PBES2-HS512+A256KW + A256GCM, 210 000 iterations, a fresh salt per call. Raises KeyringError for a password shorter than 14 bytes.

Source code in flechtctl/src/flechtctl/tokens.py 
461
462
463
464
465
466
467
468
469
470
def encrypt_keyring(keyring: Keyring, password: str) -> str:
    """The keyring as a password-protected document: an encrypted JWK Set (RFC 7517 §7).

    A compact JWE over ``keyring.to_json()``, ``PBES2-HS512+A256KW`` +
    ``A256GCM``, 210 000 iterations, a fresh salt per call. Raises
    ``KeyringError`` for a password shorter than 14 bytes.
    """
    key = _password_key(password)
    header = {"alg": _KEYRING_ALG, "cty": _KEYRING_CTY, "enc": _ENC, "p2c": _KEYRING_ITERATIONS}
    return jwe.encrypt_compact(header, keyring.to_json().encode("utf-8"), key, registry=_KEYRING_REGISTRY)

flechtctl.tokens.decrypt_keyring 

decrypt_keyring(text: str, password: str) -> Keyring

Open a password-protected keyring document.

Any failure to open it — malformed, a different algorithm, a wrong password — raises KeyringError chaining the cause.

Source code in flechtctl/src/flechtctl/tokens.py 
473
474
475
476
477
478
479
480
481
482
483
484
485
486
def decrypt_keyring(text: str, password: str) -> Keyring:
    """Open a password-protected keyring document.

    Any failure to open it — malformed, a different algorithm, a wrong
    password — raises ``KeyringError`` chaining the cause.
    """
    message = "cannot open the password-protected keyring — wrong password or damaged file"
    try:
        obj = jwe.decrypt_compact(text.strip(), _password_key(password), registry=_KEYRING_REGISTRY)
        assert obj.plaintext is not None, "a decrypted JWE always carries its plaintext"
        plaintext = obj.plaintext.decode("utf-8")
    except Exception as e:
        raise KeyringError(message) from e
    return Keyring.from_json(plaintext)

Values and tokens

flechtctl.tokens.encrypt 

encrypt(
    keyring: Keyring, value: object, *, scope: str = ""
) -> str

Encrypt a JSON value (as canonical JSON) and return the flenc:jwe: token.

Source code in flechtctl/src/flechtctl/tokens.py 
422
423
424
def encrypt(keyring: Keyring, value: object, *, scope: str = "") -> str:
    """Encrypt a JSON value (as canonical JSON) and return the ``flenc:jwe:`` token."""
    return encrypt_bytes(keyring, canonical_json(value), scope=scope)

flechtctl.tokens.decrypt 

decrypt(
    keyring: Keyring, token: str, *, scope: str = ""
) -> object

Decrypt a flenc token and parse its payload as JSON.

Source code in flechtctl/src/flechtctl/tokens.py 
427
428
429
def decrypt(keyring: Keyring, token: str, *, scope: str = "") -> object:
    """Decrypt a ``flenc`` token and parse its payload as JSON."""
    return json.loads(decrypt_bytes(keyring, token, scope=scope))

flechtctl.tokens.sign 

sign(keyring: Keyring, value: object, *, scope: str) -> str

Sign a JSON value (as canonical JSON) and return the flsig:jws: token.

Source code in flechtctl/src/flechtctl/tokens.py 
432
433
434
def sign(keyring: Keyring, value: object, *, scope: str) -> str:
    """Sign a JSON value (as canonical JSON) and return the ``flsig:jws:`` token."""
    return sign_bytes(keyring, canonical_json(value), scope=scope)

flechtctl.tokens.verify 

verify(
    keyring: Keyring, token: str, *, scope: str
) -> object

Verify a flsig token and parse its payload as JSON.

Source code in flechtctl/src/flechtctl/tokens.py 
437
438
439
def verify(keyring: Keyring, token: str, *, scope: str) -> object:
    """Verify a ``flsig`` token and parse its payload as JSON."""
    return json.loads(verify_bytes(keyring, token, scope=scope))

flechtctl.tokens.reencrypt 

reencrypt(
    keyring: Keyring, token: str, *, scope: str = ""
) -> str

Re-encrypt token under the scope's current sealing key (decrypt-with-named-kid).

The building block for a rotation sweep and for cross-environment promotion: it decrypts with the key the token names, then re-encrypts the same payload under the keyring's sealing key for scope (primaries[scope], else primary) — which moves a token from dir to a recipient key, from one recipient key to another, and back. Opening a recipient token needs that recipient's private half. The exact inner bytes are preserved, never round-tripped through the decoded value. Also promotes an unscoped token to scope (the ratchet's upgrade path); it cannot strip a scope. Raises SecretDecryptError if the token cannot be read.

Source code in flechtctl/src/flechtctl/tokens.py 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
def reencrypt(keyring: Keyring, token: str, *, scope: str = "") -> str:
    """Re-encrypt ``token`` under the scope's current sealing key (decrypt-with-named-kid).

    The building block for a rotation sweep and for cross-environment
    promotion: it decrypts with the key the token names, then re-encrypts the
    same payload under the keyring's sealing key for ``scope``
    (``primaries[scope]``, else ``primary``) — which moves a token from ``dir``
    to a recipient key, from one recipient key to another, and back. Opening a
    recipient token needs that recipient's private half. The exact inner bytes
    are preserved, never round-tripped through the decoded value. Also promotes an
    unscoped token to ``scope`` (the ratchet's upgrade path); it cannot strip a
    scope. Raises ``SecretDecryptError`` if the token cannot be read.
    """
    return encrypt_bytes(keyring, decrypt_bytes(keyring, token, scope=scope), scope=scope)

flechtctl.tokens.encrypt_bytes 

encrypt_bytes(
    keyring: Keyring, payload: bytes, *, scope: str = ""
) -> str

Encrypt raw payload bytes under the scope's sealing key, stamping scope if set.

The keyring picks the kid (primaries[scope], else primary) and refuses a scope outside that key's binding. An oct kid seals dir; an X25519 recipient kid seals ECDH-ES+A256KW to its public half.

Source code in flechtctl/src/flechtctl/tokens.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def encrypt_bytes(keyring: Keyring, payload: bytes, *, scope: str = "") -> str:
    """Encrypt raw payload bytes under the scope's sealing key, stamping ``scope`` if set.

    The keyring picks the kid (``primaries[scope]``, else ``primary``) and
    refuses a scope outside that key's binding. An oct kid seals ``dir``; an
    X25519 recipient kid seals ``ECDH-ES+A256KW`` to its public half.
    """
    kid = keyring.sealing_kid(scope)
    if kid in keyring.recipients:
        recipient = keyring.recipient_for(kid)
        if recipient.private is not None:
            _okp_private_key(kid, "X25519", recipient.public, recipient.private)
        key = OKPKey.import_key({"crv": "X25519", "kty": "OKP", "x": _b64url(recipient.public)})
        protected = {"alg": _RECIPIENT_ALG, "enc": _ENC, "kid": kid}
        registry = _RECIPIENT_REGISTRY
    else:
        key = OctKey.import_key(keyring.key_for(kid), {"kid": kid})
        protected = {"alg": _ALG, "enc": _ENC, "kid": kid}
        registry = _REGISTRY
    if scope:
        protected[_FLENC_SCOPE] = scope
    token = jwe.encrypt_compact(protected, payload, key, registry=registry)
    return PREFIX + token

flechtctl.tokens.decrypt_bytes 

decrypt_bytes(
    keyring: Keyring,
    token: str,
    *,
    scope: str = "",
    topic: str | None = None,
    wire_key: str | None = None
) -> bytes

Decrypt a flenc token to raw payload bytes, enforcing the scope ratchet.

Selects the key by the token's kid — a recipient kid opens with its private half under the ECDH-ES+A256KW allowlist, any other kid is an oct key under the dir allowlist. Then, comparing the codec's scope to the token's stamped flenc_scope:

  • scope set: reject only when the token carries a different scope (relocation). An unscoped token is accepted (upgrade / fallback) — a key-less attacker can neither mint nor strip a scope (it's the AEAD's associated data), so this is equivalent to strict once every token is scoped.
  • scope empty: reject when the token carries a scope (downgrade blocked — a scoped token may not be read by an unscoped codec).

Finally, a key bound to scopes in the keyring rejects a token whose stamped scope (possibly none) is outside its binding. That check runs after decryption, because only then is the stamped header authenticated.

Every token failure surfaces as SecretDecryptError; a keyring whose recipient halves do not match raises KeyringError.

Source code in flechtctl/src/flechtctl/tokens.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def decrypt_bytes(keyring: Keyring, token: str, *, scope: str = "", topic: str | None = None,
                  wire_key: str | None = None) -> bytes:
    """Decrypt a ``flenc`` token to raw payload bytes, enforcing the scope ratchet.

    Selects the key by the token's ``kid`` — a recipient kid opens with its
    private half under the ``ECDH-ES+A256KW`` allowlist, any other kid is an oct
    key under the ``dir`` allowlist. Then, comparing the codec's ``scope`` to the token's stamped
    ``flenc_scope``:

    - ``scope`` set: reject only when the token carries a *different* scope
      (relocation). An unscoped token is accepted (upgrade / fallback) — a
      key-less attacker can neither mint nor strip a scope (it's the AEAD's
      associated data), so this is equivalent to strict once every token is
      scoped.
    - ``scope`` empty: reject when the token *carries* a scope (downgrade
      blocked — a scoped token may not be read by an unscoped codec).

    Finally, a key bound to scopes in the keyring rejects a token whose
    stamped scope (possibly none) is outside its binding. That check runs
    after decryption, because only then is the stamped header authenticated.

    Every token failure surfaces as ``SecretDecryptError``; a keyring whose
    recipient halves do not match raises ``KeyringError``.
    """
    compact = _compact(token)
    kid = _read_kid(compact)
    if kid in keyring.recipients:
        recipient = keyring.recipient_for(kid)
        if recipient.private is None:
            raise SecretDecryptError(
                scope=scope, kid=kid, topic=topic, wire_key=wire_key,
                reason=f"keyring holds only the public half of {kid!r}: it can seal but not open",
            )
        key = _okp_private_key(kid, "X25519", recipient.public, recipient.private)
        registry = _RECIPIENT_REGISTRY
    else:
        try:
            key_bytes = keyring.key_for(kid)
        except UnknownKeyError as e:
            raise SecretDecryptError(scope=scope, kid=kid, topic=topic, wire_key=wire_key) from e
        key = OctKey.import_key(key_bytes, {"kid": kid})
        registry = _REGISTRY
    try:
        obj = jwe.decrypt_compact(compact, key, registry=registry)
    except Exception as e:
        raise SecretDecryptError(scope=scope, kid=kid, topic=topic, wire_key=wire_key) from e
    stamped = obj.protected.get(_FLENC_SCOPE)  # str | None
    if scope:
        if stamped is not None and stamped != scope:
            raise SecretDecryptError(
                scope=scope, kid=kid, topic=topic, wire_key=wire_key,
                reason=f"scope mismatch: token scoped {stamped!r}, read as {scope!r}",
            )
    elif stamped is not None:
        raise SecretDecryptError(
            scope=scope, kid=kid, topic=topic, wire_key=wire_key,
            reason=f"token is scoped {stamped!r} but this codec declares no scope (downgrade blocked)",
        )
    bound = keyring.scopes.get(kid)
    if bound is not None and stamped not in bound:
        raise SecretDecryptError(
            scope=scope, kid=kid, topic=topic, wire_key=wire_key,
            reason=f"key {kid!r} is bound to scopes {sorted(bound)}; token stamped {stamped!r}",
        )
    assert obj.plaintext is not None, "a decrypted JWE always carries its plaintext"
    return obj.plaintext

flechtctl.tokens.sign_bytes 

sign_bytes(
    keyring: Keyring, payload: bytes, *, scope: str
) -> str

Sign raw payload bytes with the keyring's signing key for scope; return the flsig:jws: token.

scope is required: a signature always names the compartment it vouches for. The keyring's signing[scope] names the key; none configured raises KeyringError.

Source code in flechtctl/src/flechtctl/tokens.py 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def sign_bytes(keyring: Keyring, payload: bytes, *, scope: str) -> str:
    """Sign raw payload bytes with the keyring's signing key for ``scope``; return the ``flsig:jws:`` token.

    ``scope`` is required: a signature always names the compartment it vouches
    for. The keyring's ``signing[scope]`` names the key; none configured raises
    ``KeyringError``.
    """
    _require_scope("sign", scope)
    kid = keyring.signing_kid(scope)
    signer = keyring.signer_for(kid)
    key = _okp_private_key(kid, "Ed25519", signer.public, signer.private)
    protected = {"alg": _SIGNATURE_ALG, _FLENC_SCOPE: scope, "kid": kid}
    token = jws.serialize_compact(protected, payload, key, registry=_SIGNATURE_REGISTRY)
    return SIGNATURE_PREFIX + token

flechtctl.tokens.verify_bytes 

verify_bytes(
    keyring: Keyring, token: str, *, scope: str
) -> bytes

Verify a flsig token to raw payload bytes, enforcing scope and binding.

Selects the signer by the token's kid and pins Ed25519. The token's stamped flenc_scope must equal scope — there is no ratchet, as no unscoped signature exists — and scope must be within the signer's binding. Every failure surfaces as SecretSignatureError.

Source code in flechtctl/src/flechtctl/tokens.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def verify_bytes(keyring: Keyring, token: str, *, scope: str) -> bytes:
    """Verify a ``flsig`` token to raw payload bytes, enforcing scope and binding.

    Selects the signer by the token's ``kid`` and pins ``Ed25519``. The token's
    stamped ``flenc_scope`` must **equal** ``scope`` — there is no ratchet, as
    no unscoped signature exists — and ``scope`` must be within the signer's
    binding. Every failure surfaces as ``SecretSignatureError``.
    """
    _require_scope("verify", scope)
    compact = _signature_compact(token)
    kid = _read_kid(compact)
    try:
        signer = keyring.signer_for(kid)
    except UnknownKeyError as e:
        raise SecretSignatureError(scope=scope, kid=kid) from e
    key = OKPKey.import_key({"crv": "Ed25519", "kty": "OKP", "x": _b64url(signer.public)})
    try:
        obj = jws.deserialize_compact(compact, key, registry=_SIGNATURE_REGISTRY)
    except Exception as e:
        raise SecretSignatureError(scope=scope, kid=kid) from e
    stamped = obj.protected.get(_FLENC_SCOPE)
    if stamped != scope:
        raise SecretSignatureError(
            scope=scope, kid=kid, reason=f"scope mismatch: signature scoped {stamped!r}, read as {scope!r}",
        )
    bound = keyring.scopes[kid]
    if scope not in bound:
        raise SecretSignatureError(
            scope=scope, kid=kid, reason=f"signer {kid!r} is bound to scopes {sorted(bound)}",
        )
    return obj.payload

flechtctl.tokens.read_header 

read_header(token: str) -> dict

The decoded protected header of a flenc or flsig token — unauthenticated.

No key is involved, so nothing here is verified: the header is only trustworthy once decrypt / verify has succeeded. For inspection and tooling. Raises SecretFormatError for any other value or a malformed header.

Source code in flechtctl/src/flechtctl/tokens.py 
227
228
229
230
231
232
233
234
235
236
237
238
def read_header(token: str) -> dict:
    """The decoded protected header of a ``flenc`` or ``flsig`` token — **unauthenticated**.

    No key is involved, so nothing here is verified: the header is only
    trustworthy once `decrypt` / `verify` has succeeded. For inspection and
    tooling. Raises ``SecretFormatError`` for any other value or a malformed header.
    """
    if is_encrypted(token):
        return _read_header(_compact(token))
    if is_signed(token):
        return _read_header(_signature_compact(token))
    raise SecretFormatError("not a flenc or flsig token")

flechtctl.tokens.is_encrypted 

is_encrypted(value: object) -> bool

True if value is ciphertext-form — a string under the flenc: scheme.

Classification is by scheme, not by exact token: a plaintext config value that merely looks like a URL under a different scheme is not ciphertext-form. The envelope segment is validated later, at decode.

Source code in flechtctl/src/flechtctl/tokens.py 
141
142
143
144
145
146
147
148
def is_encrypted(value: object) -> bool:
    """True if ``value`` is ciphertext-form — a string under the ``flenc:`` scheme.

    Classification is by scheme, not by exact token: a plaintext config value
    that merely looks like a URL under a *different* scheme is not
    ciphertext-form. The envelope segment is validated later, at decode.
    """
    return isinstance(value, str) and value.startswith(_SCHEME_PREFIX)

flechtctl.tokens.is_signed 

is_signed(value: object) -> bool

True if value is signed-form — a string under the flsig: scheme.

Source code in flechtctl/src/flechtctl/tokens.py 
151
152
153
def is_signed(value: object) -> bool:
    """True if ``value`` is signed-form — a string under the ``flsig:`` scheme."""
    return isinstance(value, str) and value.startswith(_SIGNATURE_SCHEME_PREFIX)

flechtctl.tokens.kid_of 

kid_of(token: str) -> str

Return the kid a flenc token was encrypted under (no key needed).

A scan/rotation building block: reports which key a surviving record still depends on. Raises SecretFormatError for a non-flenc value.

Source code in flechtctl/src/flechtctl/tokens.py 
206
207
208
209
210
211
212
213
214
def kid_of(token: str) -> str:
    """Return the ``kid`` a ``flenc`` token was encrypted under (no key needed).

    A scan/rotation building block: reports which key a surviving record still
    depends on. Raises ``SecretFormatError`` for a non-``flenc`` value.
    """
    if not is_encrypted(token):
        raise SecretFormatError("not a flenc token")
    return _read_kid(_compact(token))

flechtctl.tokens.signer_of 

signer_of(token: str) -> str

Return the kid of the key that signed a flsig token (no key needed).

Raises SecretFormatError for a non-flsig value.

Source code in flechtctl/src/flechtctl/tokens.py 
217
218
219
220
221
222
223
224
def signer_of(token: str) -> str:
    """Return the ``kid`` of the key that signed a ``flsig`` token (no key needed).

    Raises ``SecretFormatError`` for a non-``flsig`` value.
    """
    if not is_signed(token):
        raise SecretFormatError("not a flsig token")
    return _read_kid(_signature_compact(token))

flechtctl.tokens.canonical_json 

canonical_json(value: object) -> bytes

Canonical JSON bytes: compact, sorted keys, UTF-8, NaN rejected.

Matches the framework's canonical JSON, so encrypted / signed RECORD/DICT payloads are deterministic.

Source code in flechtctl/src/flechtctl/tokens.py 
261
262
263
264
265
266
267
268
269
def canonical_json(value: object) -> bytes:
    """Canonical JSON bytes: compact, sorted keys, UTF-8, NaN rejected.

    Matches the framework's canonical JSON, so encrypted / signed RECORD/DICT
    payloads are deterministic.
    """
    return json.dumps(
        value, allow_nan=False, ensure_ascii=False, separators=(",", ":"), sort_keys=True,
    ).encode("utf-8")

Key generation

flechtctl.tokens.generate_oct_key 

generate_oct_key() -> bytes

A fresh 32-byte AES-256 key, for the keys of a keyring.

Source code in flechtctl/src/flechtctl/tokens.py 
499
500
501
def generate_oct_key() -> bytes:
    """A fresh 32-byte AES-256 key, for the `keys` of a keyring."""
    return secrets.token_bytes(KEY_BYTES)

flechtctl.tokens.generate_x25519_key 

generate_x25519_key() -> X25519Key

A fresh X25519 recipient key pair (both halves present).

Source code in flechtctl/src/flechtctl/tokens.py 
511
512
513
514
def generate_x25519_key() -> X25519Key:
    """A fresh X25519 recipient key pair (both halves present)."""
    public, private = _generate_okp("X25519")
    return X25519Key(public=public, private=private)

flechtctl.tokens.generate_ed25519_key 

generate_ed25519_key() -> Ed25519Key

A fresh Ed25519 signing key pair (both halves present).

Source code in flechtctl/src/flechtctl/tokens.py 
517
518
519
520
def generate_ed25519_key() -> Ed25519Key:
    """A fresh Ed25519 signing key pair (both halves present)."""
    public, private = _generate_okp("Ed25519")
    return Ed25519Key(public=public, private=private)

Errors

flechtctl.errors.SecretError

Bases: Exception

Base class for secret encode/decode failures.

Source code in flechtctl/src/flechtctl/errors.py 
13
14
class SecretError(Exception):
    """Base class for secret encode/decode failures."""

flechtctl.errors.SecretFormatError

Bases: SecretError

A value is not a well-formed flenc token (unknown envelope, bad header).

Source code in flechtctl/src/flechtctl/errors.py 
17
18
class SecretFormatError(SecretError):
    """A value is not a well-formed ``flenc`` token (unknown envelope, bad header)."""

flechtctl.errors.SecretDecryptError

Bases: SecretError

A flenc token could not be decrypted or failed scope verification.

Carries what the codec knows — the codec's scope (may be empty) and the token's kid — plus topic / wire_key when a caller with that context (the scan helper) raises. The hook an application uses to quarantine a single config under the "only catch what you can remedy" rule. An unknown kid and a GCM tag failure are distinct incident signatures (missing key vs. wrong bytes for a known key); the __cause__ preserves which one.

Source code in flechtctl/src/flechtctl/errors.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class SecretDecryptError(SecretError):
    """A ``flenc`` token could not be decrypted or failed scope verification.

    Carries what the codec knows — the codec's ``scope`` (may be empty) and the
    token's ``kid`` — plus ``topic`` / ``wire_key`` when a caller with that
    context (the scan helper) raises. The hook an application uses to quarantine
    a single config under the "only catch what you can remedy" rule. An unknown
    ``kid`` and a GCM tag failure are distinct incident signatures (missing key
    vs. wrong bytes for a known key); the ``__cause__`` preserves which one.
    """

    def __init__(self, *, scope: str = "", kid: str | None = None,
                 topic: str | None = None, wire_key: str | None = None,
                 reason: str | None = None) -> None:
        self.scope = scope
        self.kid = kid
        self.topic = topic
        self.wire_key = wire_key
        detail = ["cannot decrypt secret value"]
        if scope:
            detail.append(f"scope={scope!r}")
        if kid is not None:
            detail.append(f"kid={kid!r}")
        if wire_key is not None:
            detail.append(f"key={wire_key!r}")
        if topic is not None:
            detail.append(f"topic={topic!r}")
        if reason is not None:
            detail.append(reason)
        super().__init__("; ".join(detail))

flechtctl.errors.PlaintextSecretError

Bases: SecretError

A secret value is plaintext but the codec's read_plaintext is off.

The read-time paste-guard firing: the record must be re-produced encrypted (and the pasted value treated as compromised). Flipping read_plaintext back on to silence it is the anti-pattern — it re-opens the hole for that field.

Source code in flechtctl/src/flechtctl/errors.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class PlaintextSecretError(SecretError):
    """A secret value is plaintext but the codec's ``read_plaintext`` is off.

    The read-time paste-guard firing: the record must be re-produced encrypted
    (and the pasted value treated as compromised). Flipping ``read_plaintext``
    back on to silence it is the anti-pattern — it re-opens the hole for that
    field.
    """

    def __init__(self, *, scope: str = "", topic: str | None = None, wire_key: str | None = None) -> None:
        self.scope = scope
        self.topic = topic
        self.wire_key = wire_key
        detail = ["secret value is plaintext but read_plaintext is off for this attribute"]
        if scope:
            detail.append(f"scope={scope!r}")
        if wire_key is not None:
            detail.append(f"key={wire_key!r}")
        if topic is not None:
            detail.append(f"topic={topic!r}")
        super().__init__("; ".join(detail))

flechtctl.errors.SecretSignatureError

Bases: SecretError

A flsig token could not be verified, or vouches for a different scope.

Shaped like SecretDecryptError: the codec's scope, the token's signer kid, topic / wire_key when a caller has them, and a reason. An unknown signer keeps UnknownKeyError as __cause__.

Source code in flechtctl/src/flechtctl/errors.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class SecretSignatureError(SecretError):
    """A ``flsig`` token could not be verified, or vouches for a different scope.

    Shaped like `SecretDecryptError`: the codec's ``scope``, the token's signer
    ``kid``, ``topic`` / ``wire_key`` when a caller has them, and a ``reason``.
    An unknown signer keeps `UnknownKeyError` as ``__cause__``.
    """

    def __init__(self, *, scope: str = "", kid: str | None = None,
                 topic: str | None = None, wire_key: str | None = None,
                 reason: str | None = None) -> None:
        self.scope = scope
        self.kid = kid
        self.topic = topic
        self.wire_key = wire_key
        detail = ["cannot verify signed value"]
        if scope:
            detail.append(f"scope={scope!r}")
        if kid is not None:
            detail.append(f"kid={kid!r}")
        if wire_key is not None:
            detail.append(f"key={wire_key!r}")
        if topic is not None:
            detail.append(f"topic={topic!r}")
        if reason is not None:
            detail.append(reason)
        super().__init__("; ".join(detail))

flechtctl.errors.UnsignedSecretError

Bases: SecretError

A SIGNED value is unsigned but the codec's read_unsigned is off.

Source code in flechtctl/src/flechtctl/errors.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class UnsignedSecretError(SecretError):
    """A ``SIGNED`` value is unsigned but the codec's ``read_unsigned`` is off."""

    def __init__(self, *, scope: str = "", topic: str | None = None, wire_key: str | None = None) -> None:
        self.scope = scope
        self.topic = topic
        self.wire_key = wire_key
        detail = ["value is unsigned but read_unsigned is off for this attribute"]
        if scope:
            detail.append(f"scope={scope!r}")
        if wire_key is not None:
            detail.append(f"key={wire_key!r}")
        if topic is not None:
            detail.append(f"topic={topic!r}")
        super().__init__("; ".join(detail))

flechtctl.keyring.KeyringError

Bases: Exception

Base class for keyring problems (construction, install, key lookup).

Source code in flechtctl/src/flechtctl/keyring.py 
46
47
class KeyringError(Exception):
    """Base class for keyring problems (construction, install, key lookup)."""

flechtctl.keyring.UnknownKeyError

Bases: KeyringError

A token names a kid absent from the active keyring.

In an incident this is the "the key is missing here" signature, distinct from a GCM tag failure (same kid bound to different bytes).

Source code in flechtctl/src/flechtctl/keyring.py 
50
51
52
53
54
55
56
57
58
59
class UnknownKeyError(KeyringError):
    """A token names a `kid` absent from the active keyring.

    In an incident this is the "the key is missing here" signature, distinct
    from a GCM tag failure (same `kid` bound to different bytes).
    """

    def __init__(self, kid: str) -> None:
        super().__init__(f"no key {kid!r} in the active keyring")
        self.kid = kid