Overview
The Extender class is an abstract base class (ABC) that provides an extensible framework for enhancing and wrapping functions with additional capabilities. It is especially useful for automating and monitoring various operations such as metadata harvesting, messaging integration, and event logging. This class offers a standardized approach to augmenting functions with critical features like performance monitoring, audit trails, and impact analysis.
Extenders are configured on the session: every call that plans (the mlodaAPI(...) constructor, prepare(), run_all(), stream_all(), explain(), and diagnose()) accepts function_extender and snapshots it. run() and stream_run() don't take it and never re-read it: they reuse the set snapshotted when the session was planned (a session is planned exactly once), so changing function_extender afterwards has no effect.
In the following example, we will reuse the previous feature group example and demonstrate how to monitor the execution time of the calculate_feature function using a custom extender.
Monitoring Execution Time
We will create a DokuExtender class to monitor and log the time taken for the calculate_feature function of the feature group to execute.
1. Define the Extender
from typing import Any
import time
from mloda.steward import Extender, ExtenderHook
import logging
logger = logging.getLogger(__name__)
A simple DokuExtender class:
class DokuExtender(Extender):
def wraps(self) -> set[ExtenderHook]:
return {ExtenderHook.FEATURE_GROUP_CALCULATE_FEATURE}
def __call__(self, func: Any, *args: Any, **kwargs: Any) -> Any:
start = time.time()
result = func(*args, **kwargs)
logger.error(f"Time taken: {time.time() - start}")
return result
2. Run the Example with the Extender
We will now run the mlodaAPI call, including our custom DokuExtender to monitor the execution time of the calculate_feature function.
from mloda.user import mloda
from mloda.user import DataAccessCollection
file_path = "tests/test_plugins/feature_group/src/dataset/creditcard_2023_short.csv"
data_access_collection = DataAccessCollection(files={file_path})
feature_list = ["id","V1","V2","V3"]
example_feature_list = [f"ExampleB_{f}" for f in feature_list]
mloda.run_all(
feature_list,
compute_frameworks=["PyArrowTable"],
data_access_collection=data_access_collection,
function_extender={DokuExtender()}
)
Expected Output (Logged Execution Times)
ERROR test_getting_started:test_getting_started.py:29 Time taken: 0.00454258918762207
ERROR test_getting_started:test_getting_started.py:29 Time taken: 0.001033782958984375
3. Summary
With this simple extender, you can easily log and monitor the execution time of any functionality within feature groups. By extending the Extender class, you can wrap additional behavior such as performance monitoring, logging, or auditing around critical functions to enhance observability and traceability in your data processing workflows.
When multiple extenders are provided, they are automatically chained and executed in priority order (lower values first). Each extender's wraps() is read once per run, so the hook set is fixed for the run; wraps() must return ExtenderHook members, otherwise the run fails at setup with a TypeError.
4. Error handling
By default an exception raised inside an extender is breaking: it propagates and fails the feature calculation, just like a bug in any other code. This holds whether one extender or several are registered for a hook, so adding a second extender never changes the error semantics.
An extender that is non-critical (for example observability or telemetry) can opt out by setting raise_on_error = False. When such an extender fails, the error is logged as a warning and the wrapped function still runs, so a failing extender cannot break the calculation:
class MetricsExtender(Extender):
def __init__(self) -> None:
self.raise_on_error = False # failures log a warning instead of breaking
def wraps(self) -> set[ExtenderHook]:
return {ExtenderHook.FEATURE_GROUP_CALCULATE_FEATURE}
def __call__(self, func: Any, *args: Any, **kwargs: Any) -> Any:
result = func(*args, **kwargs)
record_metric(...) # if this raises, the calculation still succeeds
return result
Only the extender's own failure is caught: an exception raised by the wrapped function (or by a downstream breaking extender) always propagates, and the wrapped function is never run twice. Concrete extenders can also expose the flag as a constructor argument (for example MetricsExtender(raise_on_error=True)) to let callers opt back into breaking behavior.
Once the wrapped call has run, an extender's return value is discarded: it can observe or fail the calculation but not substitute a different result. An extender that never calls the wrapped function has no wrapped result to prefer, so its own return value is used instead (this is what keeps a gate's raise-to-refuse or fallback pattern below working).
A gate (an authorization or identity check that refuses a call by raising) sets never_fall_back = True in its class body or __init__: its failure always propagates, whatever raise_on_error says, and the wrapped function never runs as a fallback. Gates always sort outermost on their hook, whatever their priority (priority still orders within the gate tier and within the rest; equal priorities order by the extender class's module and qualified name). An extender that swallows the wrapped call's exception gets it re-raised, and a gate that skips the wrapped call while another gate sits inside it raises GateBypassError. never_fall_back, raise_on_run_complete, raise_on_error and priority are sealed when mloda first builds the run's hook table (planning, including explain()/diagnose()): assigning or deleting them afterwards raises AttributeError, and reassigning them on the class no longer affects the sealed instance. An extender whose on_run_complete must be able to fail the run (e.g. sealing an audit log) sets raise_on_run_complete = True (see section 8). Registry strict mode ("strict", see Plugin Registry) raises EnvironmentPreconditionError for an unregistered gate or raise_on_run_complete extender instead of dropping it, and drops other unregistered extenders.
5. Reading call facts via HookContext
HookContext.current() returns the HookContext for the hook being dispatched (FEATURE_GROUP_CALCULATE_FEATURE, VALIDATE_INPUT_FEATURE, VALIDATE_OUTPUT_FEATURE, FEATURE_GROUP_MATCHED, INPUT_DATA_LOAD, JOIN), or None outside a hook call. It is set only on the thread dispatching the hook: capture it before handing work to another thread, or run that work under contextvars.copy_context().run. A reader load from a thread that lost the context is still dispatched through the INPUT_DATA_LOAD hook (gate or not) when it uses the FeatureSet handed to calculate_feature, a copy of it, or one built in the calculation thread, because each carries a per-run marker; a gate-free run is unaffected by another run's gate. A FeatureSet built inside the hopped thread has no marker and falls back to the process-wide check: under a gate it is refused with GateBypassError (use contextvars.copy_context().run), including while a gated calculation runs elsewhere in the process. A thread that outlives the calculation, or a leftover or unpickled FeatureSet loaded after it, is refused under a gate and loads directly otherwise. Calling Reader.load_data directly bypasses the hook. It carries which hook fired, the feature group's module.qualname and version() (None on JOIN, and on FEATURE_GROUP_MATCHED until the match resolves), the owning plugin's installed distribution version (resolved in the parent process at prepare and cached for the process's life, so a MULTIPROCESSING worker keeps reporting that value even if the package is upgraded on disk afterward; None when no installed distribution owns the module, on FEATURE_GROUP_MATCHED/JOIN, or outside a planned run), the requested feature names, the input feature names the engine resolved for the step at planning time (None for a root step; an injected filter or index feature does not break this, since the step inherits its host's resolved names; a best-effort input_features() re-read only for a FeatureSet assembled outside an execution plan), input_feature_edges (a dict[str, tuple[str, ...]] mapping each output feature name to its declared input names; None for a root step; a feature declaring no inputs is absent; same-named features union-merge into one entry keyed by name; injected features are absent when the engine resolved the step, but a runtime re-read, i.e. a FeatureSet assembled outside a plan or after option defaults materialize, may include them; the mapping is copied on construction like carrier), the compute framework's class name (None on FEATURE_GROUP_MATCHED), and rows_in (None when the framework can't report a count without materializing, e.g. a lazy or SQL-backed frame). rows_in is the framework's current data, so on VALIDATE_OUTPUT_FEATURE it is the freshly calculated output, not the pre-calculation input.
The fields are frozen (assigning raises FrozenInstanceError) except rows_out, output_schema, duration_seconds, status; the dict fields (carrier, declared_attributes, input_feature_edges) are read-only copies (mutating raises TypeError; json.dumps, copy and pickle give a plain dict). Read it after calling func(*args, **kwargs) in your own __call__ to also see rows_out, output_schema, duration_seconds, and status. status reflects only the wrapped call: "success" once it returns, "error" if it raised; a warning-only extender's own failure does not change it. output_schema is tuple[tuple[str, str | None], ...] | None: (column, dtype) pairs sorted by name, dtype a framework-specific human-readable string or None if that column's dtype could not be read (the name is still reported). Populated best-effort from the raw calculate_feature result on FEATURE_GROUP_CALCULATE_FEATURE, without materializing a lazy or SQL-backed result. A dict result (column to values, the interchange shape accepted before transform) is read the same way on every framework, with the Python type name of the first non-None value as dtype. On VALIDATE_OUTPUT_FEATURE, since the hook's return value carries no schema, it's read from the framework's finalized self.data (the native, post-transform/post-filter shape) the same way rows_in is, but only once validate_output_features returns without raising; rows_in is captured before that call, so it stays populated even when the call raises. None on VALIDATE_INPUT_FEATURE and every other hook, and also when the read data is in neither the dict nor the framework's native shape or has no columns, or when reading it fails. On VALIDATE_INPUT_FEATURE/VALIDATE_OUTPUT_FEATURE, rows_out stays None, since those hooks return no data. func is always an instrumentation wrapper around the feature group's method, not the bound classmethod: use Extender.feature_group_name(func) or inspect.unwrap(func) to reach the original. plan_id is the prepared session's id (stable across its runs; the only id set during FEATURE_GROUP_MATCHED, where run_id is None); structure_hash is the session's PlanContext.structure_hash on the calculate, validate, JOIN and INPUT_DATA_LOAD hooks, including in MULTIPROCESSING workers, None on FEATURE_GROUP_MATCHED and outside a planned run; run_id is minted per execution (each run() and each stream). step_uuid (a UUID, like PlanStep.step_uuid) is the compute step's PlanStep.step_uuid on the calculate, validate and INPUT_DATA_LOAD hooks, None on FEATURE_GROUP_MATCHED and JOIN, and for a FeatureSet assembled outside an execution plan; it is minted per prepare() and stays the same across that session's runs, so two steps with the same feature group, feature names and framework stay apart. tenant_id, project_id, and principal carry the server-verified values set via mloda.steward.verified_context() for the scope of a run, None when nothing set them, and are never influenced by a feature's Options. Like HookContext.current(), that scope is thread/task-local: a caller handing the actual prepare()/run()/run_all()/stream_run()/stream_all() call to another thread must copy context (contextvars.copy_context().run(...)) for it to still apply there. carrier is an opaque dict[str, str] | None forwarded from the run call; core never interprets it. Each HookContext copies carrier on construction into a read-only mapping (mutating it raises TypeError; json.dumps, copy and pickle yield a plain dict). worker_index is int | None, set only inside a spawned MULTIPROCESSING worker. declared_attributes is a copy of the feature group's declared_attributes(features) (see Feature Group Matching) on the feature group hooks, None when that call raises or returns a non-mapping or a non-str key.
FEATURE_GROUP_MATCHED, INPUT_DATA_LOAD, and JOIN populate their own extra fields and leave the rest at their defaults. On INPUT_DATA_LOAD, data_access_identity and data_access_format carry a string identity and format for the data-access handle/value: the reader's data_access_identity(data_access) classmethod decides it. The default, in order: sorted key names for a mapping; scheme/host/port/path for a string that fullmatches scheme://host[:port]/path, user info, query and fragment dropped, an abfs/abfss/wasb/wasbs container kept, the path cut back to the last / before its first percent-escape and dropped entirely for a jdbc: scheme, and the type name instead when any other scheme's path has a segment with a : before a =; an existing local path (os.path.exists), as given; else the type name. The kept URI or local path is published as given, key=value segments included, so a reader whose paths carry tokens should override this method. Core's file and document readers publish their path this way; SQLITEReader publishes <path>::<table> when its sqlite value names an existing file and the matched access names a table (the path alone otherwise), never its other credential values. data_access_identity_is_fallback is True when that identity is a fallback (the type name or bare key names), False otherwise, and None on every other hook; a fallback also logs one WARNING per reader class naming the override, and extenders that key datasets by identity should not merge on it. An override must not raise: it runs when an INPUT_DATA_LOAD extender is registered, where its exception aborts the load, and on every PlanStep.data_access_identity read. Extenders must record this field, never the raw data_access argument, which may carry credentials. data_access_dataset_version stays None, since no dataset versioning exists yet. reader_class is the reader class that loads the data, and declared_attributes holds that reader's declarations instead of the feature group's; read the class from reader_class rather than unwrapping func. reader_class is None on every other hook. INPUT_DATA_LOAD brackets the reader's load_data call, and a reader may return an InputDataDescriptor instead of reading (core's CsvReader returns a FileSource): then duration_seconds, status and rows_out (None) describe resolving the descriptor, the compute framework reads the file after calculate_feature returns, and no hook times that read. A gate on this hook still blocks the read, since the framework only reads a descriptor the reader returned, and an extender can tell the cases apart with isinstance(result, InputDataDescriptor) (from mloda.provider). On JOIN, join_type, join_keys and asof_config come from the Link being merged (asof_config is the link's AsOfJoinConfig, exported from mloda.steward; None for non-asof links), and join_left_feature_group / join_right_feature_group name the link's declared left and right feature group classes (module.qualname, the declared class even for a polymorphic link, in declared orientation regardless of which side the merge writes into). On FEATURE_GROUP_MATCHED, plan_feature_count counts completed resolutions (a raising one is not recorded) and plan_node_count deduplicated plan nodes, both before this match; the two are not comparable, since injected filter and index features become plan nodes without a resolution. plan_depth is the recursion depth; all three are running values, not final totals. The hook fires once per resolution attempt, so a feature reached from two parents fires twice.
Core's own failure logs and MlodaRunError scrub exception text through mloda.steward.scrub_credentials, which drops URL user info, query and fragment, masks credential-named key=value and quoted dict-key values (including tuple and set values), bearer and Authorization tokens, and masks the password in user/password@host:port, @host/service and @//host DSNs. An extender that logs a wrapped failure message itself should apply the same function. The exception object raised to the caller is left unchanged, so logging it yourself prints its raw text; a caller holding a DataAccessCollection can call data_access_collection.redact(str(exc)) to also mask its registered credential values before logging.
FEATURE_GROUP_MATCHED reads the mloda.steward.verified_context() scope active while a session is planned (for example prepare(), explain(), diagnose(), and the planning half of run_all()/stream_all()). Every other hook reads the scope active at run-call time (run(), stream_run(), and the run half of run_all()/stream_all(); for the stream calls that is creation, not first iteration).
A session prepared once and run many times reports each hook its own phase's identity, so a match-time gate authorizes the preparer, not the runner. Both are carried on the run's RunContext and the session's PlanContext: a run inside a verified_context() scope reports that scope's identity, and a run outside any scope runs under the preparer's identity (inherited from the PlanContext). on_run_start(run, plan, steps) receives both, so a gate can compare them and refuse a run whose identity differs. If the run identity differs from the plan identity, a never_fall_back extender wraps FEATURE_GROUP_MATCHED, and that matched gate does not itself override on_run_start, the run is refused with GateBypassError; the refusal is lifted only by a gate wrapping FEATURE_GROUP_MATCHED that implements on_run_start. Wrapping prepare() and run() in one scope, or using run_all()/stream_all(), avoids a difference. Strict-mode refusal of an unregistered gate happens before on_plan_start, so no lifecycle hooks fire. RunContext.carrier and RunContext.plugin_versions are read-only, and a framework attached to a run rejects plain assignment and deletion of its run_context, worker_index and function_extender (AttributeError). Its extender set and hook table are read-only copies, so in-place mutation raises too. These guards catch accidental mutation, not deliberate reach such as dict.clear(table) or object.__setattr__: in-process plugins are trusted code, not sandboxed (see Registration-Time Enforcement).
explain() and diagnose() let an exception a breaking extender (raise_on_error = True, or never_fall_back = True) raises propagate, for example a refusal at FEATURE_GROUP_MATCHED; diagnose() still projects it when it is one of the error types it projects. An extender with raise_on_error = False and no never_fall_back is logged and swallowed instead, so resolution falls back.
from mloda.steward import Extender, ExtenderHook, HookContext
class FactsExtender(Extender):
def wraps(self) -> set[ExtenderHook]:
return {ExtenderHook.FEATURE_GROUP_CALCULATE_FEATURE}
def __call__(self, func: Any, *args: Any, **kwargs: Any) -> Any:
result = func(*args, **kwargs)
context = HookContext.current()
if context is not None:
logger.info(f"{context.feature_group_class}: {context.duration_seconds}s, {context.rows_out} rows out")
return result
specialized_from is a tuple of module.qualname strings naming the parent classes the step's feature group replaced via subclass preference, sorted by that string, and () when none. On FEATURE_GROUP_MATCHED it is written once the match returns, like feature_group_class; on compute hooks and INPUT_DATA_LOAD it is the union over the step's features; on JOIN it is ().
To unit-test an extender's __call__ without running the engine, build a HookContext directly and activate() it. Only hook, feature_group_class, feature_group_version and compute_framework_name are required; the last three may be None. plugin_version, input_features and input_feature_edges default to None (a mapping passed in is copied, like carrier), feature_names and specialized_from default to empty tuples, and every other field defaults to None.
from mloda.steward import ExtenderHook, HookContext
context = HookContext(
hook=ExtenderHook.FEATURE_GROUP_CALCULATE_FEATURE,
feature_group_class="my_plugin.MyFeatureGroup",
feature_group_version="1",
compute_framework_name="PyArrowTable",
)
with context.activate():
FactsExtender()(lambda: "result")
6. Discovering Extenders
To list all available extenders and their documentation, use the get_extender_docs() function from mloda.steward.
7. Extender identity under ParallelizationMode.MULTIPROCESSING
A framework that stays resident in the parent process uses the caller's own extender objects, not copies: identity and mutable state on an extender (a client, tracer, connection) hold across every framework built there, including one whose own resolved parallelization_mode is not MULTIPROCESSING inside an overall MULTIPROCESSING-enabled run (e.g. a compute framework that only supports SYNC). For that SYNC-only framework to run at all, the requested parallelization_modes set must also contain the framework's own supported mode, e.g. {ParallelizationMode.SYNC, ParallelizationMode.MULTIPROCESSING}; requesting {ParallelizationMode.MULTIPROCESSING} alone filters the framework out at setup and the run raises.
A parent-resident extender is one shared object, invoked by every framework built there, potentially concurrently under ParallelizationMode.THREADING. It must be thread-safe. Apart from the sealed flags (section 4) its state is unprotected, so keeping it consistent across frameworks and runs is the author's job.
For a framework whose class declares MULTIPROCESSING support, a framework dispatched to a spawned worker gets an independent copy, materialized only when the framework is unpickled inside the worker: the parent only ever holds the pickled snapshot taken at run start. The snapshot is taken once per run, at setup (a stream's first iteration), so for runs executed one after another it carries the parent's state as the previous run's on_run_complete left it, and the worker copy also reflects state set in this run's on_run_start. The extender must be picklable at that point: a live handle that pickle can't round-trip (a client, tracer, connection, lock) must be stripped in __getstate__ and rebuilt in __setstate__ or on first use in __call__. Constructing the handle eagerly in __init__ is fine as long as __getstate__ strips it before pickling. State a worker copy accumulates stays in that worker unless the extender ships it out itself (a file, a socket, an exporter).
Two copies in the same process that resolve to the same underlying handle (e.g. a pooled client) share its lifecycle, not two independently rebuilt handles: an extender exposing close() must make the closed state visible to every copy resolving to that handle, or closing one leaves the others writing to a dead handle.
An extender that accepts an injected sink (client, provider, connection) should trial-pickle it in __getstate__ and drop it if it can't survive: the run then completes with the sink dropped, instead of being rejected outright by mloda's plan-time picklability check. Log the drop once per instance. mloda.steward ships two small primitives for this: pickle_failure_reason(value) (None if value pickles cleanly, else the caught exception's type name) and WarnOncePerInstance, a thread-safe double-checked-locking guard whose warn_once(emit) fires emit at most once per instance and always resets to unwarned on any copy, so a pickled worker copy decides independently whether to warn instead of inheriting the parent's already-fired state:
import logging
from typing import Any
from mloda.steward import Extender, pickle_failure_reason, WarnOncePerInstance
logger = logging.getLogger(__name__)
class MyExtender(Extender):
def __init__(self, sink: Any = None) -> None:
self._sink = sink
self._drop_guard = WarnOncePerInstance()
def __getstate__(self) -> dict[str, Any]:
state = dict(self.__dict__)
reason = pickle_failure_reason(self._sink) if self._sink is not None else None
if reason is not None:
self._drop_guard.warn_once(lambda: logger.warning(f"dropping unpicklable sink ({reason})"))
state["_sink"] = None
return state
No custom __setstate__ is needed for _drop_guard itself: pickle reconstructs it as a fresh, unwarned guard on its own. Plan-time picklability validation and worker-dispatch pickling both pickle the same live extender instance, so this drop-and-warn fires once, in the parent, during plan-time validation; the later dispatch pickle is then silent, since the guard has already fired for that instance.
8. Plan and run lifecycle hooks
Extender has four no-op hooks, all run in the parent on the caller's own extender objects that the request used (registry strict mode drops unregistered ones, see section 4), in every parallelization mode, for every extender even if wraps() returns nothing, in gates first, then ascending priority order:
on_plan_start(plan)andon_plan_complete(plan, outcome)wrap planning:prepare(),explain(),diagnose()and the planning half ofrun_all()/stream_all().planis the session'sPlanContext.plan.structure_hash(see Plan Lock) isNoneinon_plan_startand after a failed plan, and set inon_plan_completeandon_run_start. ThePlanContextpassed to those two hooks is a new instance carrying the hash (it still compares equal to the oneon_plan_startgot), so read the hash from the context the later hook receives, not one kept fromon_plan_start. Unlikeplan_idit is the same across sessions and processes for the same plan, and it is excluded fromPlanContextequality.RunContext.structure_hashandHookContext.structure_hashcarry the same value into each run and its per-calculation hooks.plan.content_hashfollows the same lifecycle; see Plan Lock for what it covers.on_plan_completealso fires when planning raised (diagnose()reports its swallowed errors asfailed).on_run_start(run, plan, steps)fires once perrun()or stream, before any setup or compute.runis theRunContext,stepsa tuple of the session'sPlanSteprecords (computed once per session). An exception it raises refuses the run when the extender'sraise_on_errorornever_fall_backisTrue; otherwise it is logged and the run proceeds. A refusal stops later extenders from seeingon_run_start, but they still geton_run_complete.on_run_complete(run, outcome)fires exactly once per run, after the workers were joined and the runner exited: also when setup, execution, finalizing or anon_run_startrefusal raised.
outcome is a frozen LifecycleOutcome(status, error_type) (exported from mloda.steward with PlanContext and RunContext): status is "succeeded", "failed" (with error_type the exception class name) or "cancelled". A stream closed early, or never iterated and then closed or collected, completes as "cancelled"; if finalizing raises while it closes, "failed" wins. A stream primes at stream_run()/stream_all(), so an on_run_start refusal raises there, not on first iteration. A stream's RunContext.started_at is its creation time, and its run identity is captured at the stream_run() call. An old on_run_complete(self, run_id) signature raises TypeError naming the extender when the extender set is set up.
Exceptions in on_plan_start, on_plan_complete and on_run_complete are logged and never propagated by default: raise_on_error and never_fall_back do not apply, and the remaining extenders are still called. With raise_on_run_complete = True, the first Exception from on_run_complete (in hook order) of a "succeeded" run propagates out of run()/run_all(), or the final next() of an exhausted stream, after all extenders were called; it is logged for a "failed" or "cancelled" run, and later ones are logged. close(), by contrast, runs only on a MULTIPROCESSING worker's copy. A worker that does not exit within graceful_shutdown_timeout (one deadline for all workers) is terminated, possibly mid-close(), and on_run_complete still fires: a sink whose records must not be lost should flush synchronously, not only in close(). run_all/stream_all accept graceful_shutdown_timeout too, forwarding it the same way as run()/stream_run(). Inside close(), CloseContext.current() returns the ambient context for that worker's close, None everywhere else: deadline, reason ("stop" when the worker stops normally, a STOP or finishing its work; "error" for any abnormal exit, a failed command or bootstrap, or an unexpected exception; "parent_gone" when the parent process died), run_id, plan_id, worker_index, carrier, tenant_id, project_id, principal, and remaining() (seconds left, clamped to 0.0). It is one instance shared across every extender closing in that worker, including its carrier, so one extender mutating it is visible to the others; remaining() can overstate by the time between the parent sending STOP and this worker starting to close. Capture the instance and pass it to threads, since a thread started inside close() does not inherit it. Unit tests can build one directly, requiring only deadline (a time.monotonic() timestamp) and reason, and use activate() the same way as HookContext.
run.run_id is minted per execution, the same value as HookContext.run_id in the per-calculation hooks, so records can be correlated; re-running a prepared session gives a new run_id (run.plan_id and HookContext.plan_id stay the session's id). HookContext.current() is None inside these hooks. The hooks are synchronous with no time budget: a blocking hook blocks the caller.
from typing import Any
from mloda.steward import Extender, ExtenderHook, LifecycleOutcome, RunContext
class SealingExtender(Extender):
def __init__(self) -> None:
self.completed_run_ids: list[str | None] = []
def wraps(self) -> set[ExtenderHook]:
return set()
def __call__(self, func: Any, *args: Any, **kwargs: Any) -> Any:
return func(*args, **kwargs)
def on_run_complete(self, run: RunContext, outcome: LifecycleOutcome) -> None:
self.completed_run_ids.append(run.run_id)