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, 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 
 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
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,
    ) -> "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``). 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
        if enrich_config is not None:
            instance.enrich_config = enrich_config
        if extract_state_key is not None:
            instance.extract_state_key = extract_state_key
        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.

        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
) -> 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). The ABC discipline still applies to every other construction path — Transformer() and any abstract subclass remain uninstantiable.

Source code in src/flechtwerk/transformer.py 
 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
@classmethod
def of(
        cls,
        *,
        input_topics: list[str],
        transform: TransformFn,
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | 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``). 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
    if enrich_config is not None:
        instance.enrich_config = enrich_config
    if extract_state_key is not None:
        instance.extract_state_key = extract_state_key
    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.

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 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@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.

    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
) -> 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 and extract_state_key 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 
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
def transformer(
        *,
        input_topics: list[str],
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | 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`` and ``extract_state_key`` 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,
            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.

Source code in src/flechtwerk/extractor.py 
 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
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
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.
    """

    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,
    ) -> "Extractor":
        """Build an Extractor from a poll function and config topics.

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

        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`` 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
        if enrich_config is not None:
            instance.enrich_config = enrich_config
        if extract_state_key is not None:
            instance.extract_state_key = extract_state_key
        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.
          Tombstoned, suspended, disowned (rebalanced-away), and rewritten
          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 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``.
        """

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
) -> Extractor

Build an Extractor from a poll function and config topics.

enrich_config and extract_state_key are optional overrides; omit them to use the defaults (no enrichment, extract_state_key returns the Kafka message key).

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

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

    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`` 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
    if enrich_config is not None:
        instance.enrich_config = enrich_config
    if extract_state_key is not None:
        instance.extract_state_key = extract_state_key
    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. Tombstoned, suspended, disowned (rebalanced-away), and rewritten 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 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 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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.
      Tombstoned, suspended, disowned (rebalanced-away), and rewritten
      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 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 
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
@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
) -> 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 and extract_state_key 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 
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
def extractor(
        *,
        config_topics: list[str],
        enrich_config: EnrichConfigFn | None = None,
        extract_state_key: ExtractStateKeyFn | 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`` and ``extract_state_key`` 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,
            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. 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 
 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
@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. 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
    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 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.

flechtwerk.IncomingMessage dataclass

A message read from Kafka.

Source code in src/flechtwerk/types.py 
51
52
53
54
55
56
57
58
59
60
@dataclass(frozen=True, slots=True)
class IncomingMessage:
    """A message read from Kafka."""

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

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 
 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
146
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 
61
62
63
64
65
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 
72
73
74
75
76
77
78
79
80
81
82
83
84
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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
102
103
104
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 
106
107
108
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 
110
111
112
113
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 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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

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.

Source code in src/flechtwerk/module.py 
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
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.
    """

    @classmethod
    def of(
        cls,
        *,
        application_id: str,
        bootstrap_servers: str,
        client_id: str,
        stage: Extractor | Transformer,
        compression_type: CompressionType | None = "zstd",
        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 — so it must be unique
        per instance and stable across restarts (production K8s passes the
        pod name). ``compression_type`` defaults to ``"zstd"`` because
        Flechtwerk outputs JSON everywhere (encode_json) and JSON
        compresses ~13×; pass ``None`` to disable. ``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
        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.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",
    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 — so it must be unique per instance and stable across restarts (production K8s passes the pod name). compression_type defaults to "zstd" because Flechtwerk outputs JSON everywhere (encode_json) and JSON compresses ~13×; pass None to disable. 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 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 
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
@classmethod
def of(
    cls,
    *,
    application_id: str,
    bootstrap_servers: str,
    client_id: str,
    stage: Extractor | Transformer,
    compression_type: CompressionType | None = "zstd",
    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 — so it must be unique
    per instance and stable across restarts (production K8s passes the
    pod name). ``compression_type`` defaults to ``"zstd"`` because
    Flechtwerk outputs JSON everywhere (encode_json) and JSON
    compresses ~13×; pass ``None`` to disable. ``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
    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.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 
270
271
272
273
274
275
276
277
278
@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). Malformed values decode to an empty Config with a warning, so they flow into the caller's validation instead of masquerading as a missing key.

From a stage's perspective the store is read-only: query it with get() (and in / len). put/delete exist for the config machinery alone — calling 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 
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
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). Malformed
    values decode to an empty `Config` with a warning, so they flow into
    the caller's validation instead of masquerading as a missing key.

    From a stage's perspective the store is **read-only**: query it with
    `get()` (and ``in`` / ``len``). `put`/`delete` exist for the config
    machinery alone — calling 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._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()
        store._raw = {key: encode_json(value) for key, value in entries.items()}
        return store

    def get(self, key: str) -> Config | None:
        """Return the latest config for ``key``, or None if absent."""
        raw = self._raw.get(key)
        return None if raw is None else Config(decode_event(raw, key))

    def put(self, key: str, config: Record) -> None:
        self._raw[key] = encode_json(config)

    def delete(self, key: str) -> None:
        self._raw.pop(key, None)

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 
73
74
75
76
77
78
@classmethod
def of(cls, entries: dict[str, Record]) -> "ConfigStore":
    """Build a pre-seeded store — the test-side entry point."""
    store = cls()
    store._raw = {key: encode_json(value) for key, value in entries.items()}
    return store

get 

get(key: str) -> Config | None

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

Source code in src/flechtwerk/configs.py 
80
81
82
83
def get(self, key: str) -> Config | None:
    """Return the latest config for ``key``, or None if absent."""
    raw = self._raw.get(key)
    return None if raw is None else Config(decode_event(raw, key))

flechtwerk.MqttBrokerConfig dataclass

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

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 broker-only: the identity of the instance's persistent MQTT session is the module-wide client_id (see Flechtwerk.of), which configured_stage injects onto the stage alongside these settings.

Source code in src/flechtwerk/module.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass(frozen=True)
class MqttBrokerConfig:
    """Shared MQTT connection settings — one broker serves the whole platform.

    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 broker-only: the identity of the instance's persistent MQTT
    session is the module-wide ``client_id`` (see ``Flechtwerk.of``), which
    ``configured_stage`` injects onto the stage alongside these settings.
    """
    broker: str
    port: int
    password: str = ""
    qos: int = 1
    username: str = ""

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 for attributes. See the Encrypted Secrets concept page for usage and the full spec (wire format, rotation, migration, post-quantum posture).

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 
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
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: object) -> str:
        # sort_keys + allow_nan=False match the framework's canonical JSON
        # (kafka.encode_json), so encrypted RECORD/DICT payloads are
        # deterministic and NaN is rejected rather than emitted.
        payload = json.dumps(
            inner.encode(value), allow_nan=False, ensure_ascii=False,
            separators=(",", ":"), sort_keys=True,
        ).encode("utf-8")
        return _encrypt_payload(scope, payload)

    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.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: every key is 32 bytes and primary names a held key.

Source code in src/flechtwerk/keyring.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
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
@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: every key is 32 bytes and `primary`
    names a held key.
    """

    keys: dict[str, bytes]
    primary: str

    def __post_init__(self) -> None:
        if not self.keys:
            raise KeyringError("a keyring needs at least one key")
        for kid, key in self.keys.items():
            if not kid:
                raise KeyringError("a key id must be a non-empty string")
            if len(key) != KEY_BYTES:
                raise KeyringError(f"key {kid!r} is {len(key)} bytes, expected {KEY_BYTES} (AES-256)")
        if self.primary not in self.keys:
            raise KeyringError(f"primary {self.primary!r} is not among the keys {sorted(self.keys)}")

    @classmethod
    def of(cls, keys: dict[str, bytes], *, primary: str) -> "Keyring":
        """Build a keyring from raw key bytes — the programmatic / test entry point."""
        return cls(keys=dict(keys), primary=primary)

    @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 a `kid`; 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.
        """
        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] = {}
        for jwk in raw_keys:
            if not isinstance(jwk, dict):
                raise KeyringError(f"each JWK must be a JSON object, got {type(jwk).__name__}")
            if jwk.get("kty") != "oct":
                raise KeyringError(f"unsupported key type {jwk.get('kty')!r} — only oct (symmetric) keys are used")
            kid = jwk.get("kid")
            if not isinstance(kid, str) or not kid:
                raise KeyringError(f"every JWK needs a non-empty string kid, got {kid!r}")
            if kid in keys:
                raise KeyringError(f"duplicate kid {kid!r} in the keyring")
            if "k" not in jwk:
                raise KeyringError(f"JWK {kid!r} has no 'k' (key material)")
            try:
                keys[kid] = _decode_key(jwk["k"])
            except (TypeError, ValueError) as e:  # binascii.Error ⊂ ValueError
                raise KeyringError(f"JWK {kid!r} has malformed base64url 'k': {e}") from e
        if "primary" not in doc:
            raise KeyringError("keyring document has no top-level 'primary' member")
        return cls(keys=keys, primary=doc["primary"])

    def key_for(self, kid: str) -> bytes:
        """Return the key bytes for `kid`, or raise `UnknownKeyError`."""
        try:
            return self.keys[kid]
        except KeyError:
            raise UnknownKeyError(kid) from None

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

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

of classmethod 

of(keys: dict[str, bytes], *, primary: str) -> Keyring

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

Source code in src/flechtwerk/keyring.py 
113
114
115
116
@classmethod
def of(cls, keys: dict[str, bytes], *, primary: str) -> "Keyring":
    """Build a keyring from raw key bytes — the programmatic / test entry point."""
    return cls(keys=dict(keys), primary=primary)

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 a kid; 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 src/flechtwerk/keyring.py 
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
@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 a `kid`; 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.
    """
    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] = {}
    for jwk in raw_keys:
        if not isinstance(jwk, dict):
            raise KeyringError(f"each JWK must be a JSON object, got {type(jwk).__name__}")
        if jwk.get("kty") != "oct":
            raise KeyringError(f"unsupported key type {jwk.get('kty')!r} — only oct (symmetric) keys are used")
        kid = jwk.get("kid")
        if not isinstance(kid, str) or not kid:
            raise KeyringError(f"every JWK needs a non-empty string kid, got {kid!r}")
        if kid in keys:
            raise KeyringError(f"duplicate kid {kid!r} in the keyring")
        if "k" not in jwk:
            raise KeyringError(f"JWK {kid!r} has no 'k' (key material)")
        try:
            keys[kid] = _decode_key(jwk["k"])
        except (TypeError, ValueError) as e:  # binascii.Error ⊂ ValueError
            raise KeyringError(f"JWK {kid!r} has malformed base64url 'k': {e}") from e
    if "primary" not in doc:
        raise KeyringError("keyring document has no top-level 'primary' member")
    return cls(keys=keys, primary=doc["primary"])

key_for 

key_for(kid: str) -> bytes

Return the key bytes for kid, or raise UnknownKeyError.

Source code in src/flechtwerk/keyring.py 
158
159
160
161
162
163
def key_for(self, kid: str) -> bytes:
    """Return the key bytes for `kid`, or raise `UnknownKeyError`."""
    try:
        return self.keys[kid]
    except KeyError:
        raise UnknownKeyError(kid) from None

primary_pair 

primary_pair() -> tuple[str, bytes]

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

Source code in src/flechtwerk/keyring.py 
165
166
167
def primary_pair(self) -> tuple[str, bytes]:
    """Return `(primary_kid, primary_key_bytes)` — what encryption stamps and uses."""
    return self.primary, self.keys[self.primary]

kids 

kids() -> list[str]

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

Source code in src/flechtwerk/keyring.py 
169
170
171
def kids(self) -> list[str]:
    """Loaded key ids, sorted — the labels for the startup keyring gauge."""
    return sorted(self.keys)

flechtwerk.secrets.encrypt_value 

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

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

The write-side boundary for producers and form backends — the primary defense that keeps malformed tokens off the topic. Raises SecretError if attribute is not an encrypted attribute.

Source code in src/flechtwerk/secrets.py 
338
339
340
341
342
343
344
345
def encrypt_value[V](attribute: Attribute[V], value: V) -> str:
    """Encrypt ``value`` for ``attribute`` and return the ``flenc:jwe:`` token.

    The write-side boundary for producers and form backends — the primary
    defense that keeps malformed tokens off the topic. Raises ``SecretError``
    if ``attribute`` is not an encrypted attribute.
    """
    return _require_encrypted(attribute).encode(value)

flechtwerk.secrets.reencrypt 

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

Re-encrypt token under the current primary 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 primary — preserving the exact inner bytes without round-tripping 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 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def reencrypt[V](token: str, attribute: Attribute[V]) -> str:
    """Re-encrypt ``token`` under the current primary 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 primary — preserving the exact inner bytes without
    round-tripping 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.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 src/flechtwerk/secrets.py 
161
162
163
164
165
166
167
168
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)

flechtwerk.secrets.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 src/flechtwerk/secrets.py 
204
205
206
207
208
209
210
211
212
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))

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.

Source code in src/flechtwerk/secrets.py 
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
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
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.
    """
    names = [a.name for a in attributes]
    topics = list(topics)
    found: list[ScanEntry] = []

    async def collect(msg) -> None:
        if is_tombstone(msg.value):
            return
        wire_key = decode_key(msg.key)
        raw = decode_event(msg.value, f"{msg.topic}/{msg.offset}").raw
        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