Hooks & Lifecycle

This reference covers the complete BaseIngestor hook surface. Template plugins inherit all of these hooks; plugins that own the whole batch pipeline subclass BaseIngestor directly and implement _process_batch.

DirectZarr plugins add the index contract on the template page. See Plugin Templates and Index Specification.

Firecube does not yet expose a public storage-writer protocol for custom output code. A new external custom pipeline cannot be implemented using only stable, typed public storage APIs. Do not import the concrete storage session from an internal module as a workaround.

Name	Description
BaseIngestor	Orchestrating base class for all Firecube ingestors.
BaseIngestor.discover_source_files	Discover the source items to ingest (Hook).
BaseIngestor._process_batch	Process a single batch. Called in worker threads for parallel execution.
BaseIngestor.resolve_output_uri	Resolve the canonical output URI (Dataset Directory) for this run.
BaseIngestor.filter_item	Decide whether one discovered item proceeds to batching (Hook).
BaseIngestor.item_size_bytes	Estimate size of an item in bytes.
BaseIngestor.get_batch_groups	Return the logical write groups covered by a batch of items (Hook).
BaseIngestor.prepare_batch_data	Optional hook to prepare data (e.g. load files to DB) before group iteration.
BaseIngestor.cleanup_batch_data	Optional hook to clean up batch data (e.g. drop table rows).
BaseIngestor.batch_setup	Acquire per-batch resources through cooperative setup hooks.
BaseIngestor.batch_teardown	Release resources acquired by successful cooperative batch setup.
BaseIngestor.on_pipeline_start	Called before pipeline execution starts.
BaseIngestor.on_batch_success	Called on successful batch completion (engine-owned bookkeeping).
BaseIngestor.on_batch_failure	Called on batch failure (engine-owned bookkeeping).
BaseIngestor.slice_meta_keys	Name supplied options to include in the logical slice identity.
BaseIngestor.slice_meta	Return canonical slice metadata for this run's resume identity.
BaseIngestor.validation_group	Optional hook to derive a Zarr group path for validate_zarr.
BaseIngestor.catalog_group_info	Optionally annotate or hide one discovered catalog group.
BaseIngestor.default_aggregate_metrics	Default aggregate helper for plugins that do not need custom policy.

Base Class

firecube.ingestor.api.BaseIngestor

Bases: BaseIngestorHookMixin, Ingestor, ABC

Orchestrating base class for all Firecube ingestors.

BaseIngestor is a composition facade. It wires together the runtime services (batching, telemetry, recording, workspace, resume-guard) and exposes a hook surface so that plugin authors only need to implement their domain logic.

run() delegates to: - BatchPlanner — creates PipelineBatch objects from items. - TierConfigurator — splits flat options into EngineConfig / TemplateConfig / PluginConfig options tiers (not storage layers; storage I/O wiring is handled by StorageBinding / StorageSession). - WorkspaceManager — manages per-run temp directories. - ResumeGuard — enforces resume / overwrite safety. - TelemetryService — wraps the telemetry sink with metric limits. - SpanRecorder — writes manifest entries to ChunkManager. - PipelineExecutor — parallel or sequential batch dispatch.

Hooks fall into three categories. The class-variable name is also required.

MUST override (abstract): _process_batch(batch, ctx) -> PipelineResult Core per-batch logic. Called in worker threads for parallel mode. Templates (GenericZarrIngestor, GenericParquetIngestor) implement this; plain plugins may override it directly.

SHOULD override (return empty defaults; rarely correct without change): discover_source_files(ctx) -> Iterable Discovers source files to be batched (paths/URIs/objects) using configured include patterns.

CAN override (optional lifecycle / metadata hooks): filter_item(item, ctx) -> bool Per-item filter applied before batching. Default: keep all. item_size_bytes(item) -> int | None Used for batch size estimation. Default: None. get_batch_groups(items, ctx) -> list[str] Derives logical output groups for a batch. Default: ['default']. slice_meta_keys() -> list[str] Option keys that identify a logical "slice" (used for resume conflict detection). Default: empty list. validation_group(ctx) -> str | None Zarr group path used by the resume-guard's validate_zarr check. on_pipeline_start(ctx, state) Called once before any batch runs. Useful for shared resource setup (e.g. persistent DuckDB schema). on_batch_success(ctx, state, batch, result) Called on the main thread after each successful batch. on_batch_failure(ctx, state, batch, result) Called on the main thread after each failed batch. DO NOT override (framework internals): run(ctx) — top-level orchestration entry point. _create_batches — delegates to BatchPlanner. finalize_pipeline — delegates to PipelineExecutor.

Attributes:

Name	Type	Description
time_dim_name	str	The firecube append/index dimension name written into the Zarr store. Defaults to "timestamp" for backward compatibility; mirrors the PRODUCT_NAME pattern. Not a config-tier field, not exposed via --option — a property of the plugin/product, not a per-run knob. Override on a plugin subclass to match the product's own naming convention (e.g. time_dim_name: ClassVar[str] = "time"), matching whatever dimension the plugin's own hook implementation actually produces. The runtime engine also consumes this name to seed coordinate-array chunks in staged mode. Plugin declaration is authoritative: changing it against an existing store fails loudly with migration guidance rather than silently reinterpreting the store.

Class Declarations

• PRODUCT_NAME: ClassVar[str]: required, non-empty product name; checked when the class is defined.
• time_dim_name: ClassVar[str]: time dimension name, default "timestamp"; a class declaration, not a --option field.
• template_config_class: the template config dataclass whose fields become validated typed options; set by each template.
• plugin_config_class: the plugin's own PluginConfig subclass declaring product-specific options.

Core Hooks

firecube.ingestor.api.BaseIngestor.discover_source_files

discover_source_files(ctx)

Discover the source items to ingest (Hook).

The default implementation requires --input-data (ctx.source) and raises ConfigurationError without it. It searches recursively below ctx.source — a local path or a remote URI reached through the run's storage configuration — and collects .zip, .h5, .nc, .nc4, .hdf, and .he5 files plus local files with unselected or absent suffixes whose content looks like HDF5. Positive input_filters entries add files to that set; entries starting with ! exclude matches, overriding suffix and content selection regardless of list order. Pass --input-filters as one JSON list. Filters match case-sensitively, including for single-file sources. Custom discovery overrides must explicitly apply filters, for example by forwarding them to discover_input_files.

Override this hook when source layout rules cannot be expressed as patterns, or when sources are not file trees at all (catalogs, APIs); overriding removes the --input-data requirement. Returned items flow through filter_item and batching into the build hooks; return path or URI strings unless the plugin's own hooks handle richer item objects end to end.

The engine calls this hook once per run and materialises the returned iterable in full before the first batch is planned; the held items are then reused for batching and never re-discovered. Returning an iterator or generator is fine for large listings and avoids building a second container in the plugin, but it is consumed completely up front, so it cannot defer or stream discovery across batches.

firecube.ingestor.api.BaseIngestor._process_batch abstractmethod

_process_batch(batch, ctx)

Process a single batch. Called in worker threads for parallel execution.

Subclasses must implement this. Templates (GenericZarrIngestor, GenericParquetIngestor) provide a default implementation via their own build_dataset abstract hook.

firecube.ingestor.api.BaseIngestor.resolve_output_uri

resolve_output_uri(ctx, write_mode='staged')

Resolve the canonical output URI (Dataset Directory) for this run.

Batch Shaping

firecube.ingestor.api.BaseIngestor.filter_item

filter_item(item, ctx)

Decide whether one discovered item proceeds to batching (Hook).

Called once per item, after discovery and before batching. Returning False drops the item silently; it never reaches a batch or a plugin hook. Unlike input_filters/ discover_source_files, which control which files discovery finds at all, this hook filters items discovery has already found.

Parameters:

Name	Type	Description	Default
item	Any	One discovered item, in whatever form discovery produced it.	required
ctx	PluginContext	Read-only plugin context.	required

Returns:

Type	Description
bool	True to keep the item, False to drop it. Default:
bool	True (keep all).

Examples:

Drop zero-byte files before they reach batching:

def filter_item(self, item, ctx):
    return Path(item).stat().st_size > 0

firecube.ingestor.api.BaseIngestor.item_size_bytes

item_size_bytes(item)

Estimate size of an item in bytes.

firecube.ingestor.api.BaseIngestor.get_batch_groups

get_batch_groups(items, ctx)

Return the logical write groups covered by a batch of items (Hook).

A group names a logical write destination: for Zarr templates it becomes the group path inside the store, for Parquet it becomes the output partition (subdirectory). Called once per planned batch; build_dataset is then invoked once per group with the full batch item list, and the plugin routes items to the group it is building.

Implementations MUST be deterministic and return a stable, sorted list so grouping stays consistent across runs.

Parameters:

Name	Type	Description	Default
items	Sequence[Any]	Items assigned to this batch.	required
ctx	PluginContext	Read-only plugin context.	required

Returns:

Type	Description
list[str]	Group names for the batch. Default: ["default"].

Examples:

Write two Zarr groups from the same source items. Each group name becomes the group path in the store, and build_dataset selects what belongs in the group it is called for:

def get_batch_groups(self, items, ctx):
    return ["quality", "sst"]

def build_dataset(self, group, items, ctx):
    ds = self._open(items, ctx)
    if group == "sst":
        return ds[["sea_surface_temperature"]]
    return ds[["quality_level"]]

firecube.ingestor.api.BaseIngestor.prepare_batch_data

prepare_batch_data(batch, ctx)

Optional hook to prepare data (e.g. load files to DB) before group iteration.

firecube.ingestor.api.BaseIngestor.cleanup_batch_data

cleanup_batch_data(batch, ctx)

Optional hook to clean up batch data (e.g. drop table rows).

Lifecycle Hooks

Cooperative hooks: overrides of batch_setup and batch_teardown must call super() so mixins in the class hierarchy run their own setup and teardown.

firecube.ingestor.api.BaseIngestor.batch_setup

batch_setup(ctx)

Acquire per-batch resources through cooperative setup hooks.

Overrides must call super().batch_setup(ctx). GenericZarrIngestor, GenericParquetIngestor, and DirectZarrIngestor wrap batch processing in setup and teardown, including dataset construction or write-intent generation. Teardown runs only after successful setup; if setup raises, the acquiring hook must release resources it has already acquired. Direct BaseIngestor subclasses own this boundary themselves.

Parameters:

Name	Type	Description	Default
ctx	PluginContext	Read-only context for the current run.	required

firecube.ingestor.api.BaseIngestor.batch_teardown

batch_teardown(ctx)

Release resources acquired by successful cooperative batch setup.

Overrides must call super().batch_teardown(ctx). Callers that create a BatchResourceRegistry must explicitly drain it; this hook does not automatically discover or manage registry instances.

Parameters:

Name	Type	Description	Default
ctx	PluginContext	Read-only context for the current run.	required

firecube.ingestor.api.BaseIngestor.on_pipeline_start

on_pipeline_start(ctx, state)

Called before pipeline execution starts.

firecube.ingestor.api.BaseIngestor.on_batch_success

on_batch_success(ctx, state, batch, result)

Called on successful batch completion (engine-owned bookkeeping).

firecube.ingestor.api.BaseIngestor.on_batch_failure

on_batch_failure(ctx, state, batch, result)

Called on batch failure (engine-owned bookkeeping).

Validation And Metadata

firecube.ingestor.api.BaseIngestor.slice_meta_keys

slice_meta_keys()

Name supplied options to include in the logical slice identity.

Returns:

Type	Description
list[str]	Option keys consumed by the default slice_meta implementation.
list[str]	Keys absent from ctx.options are omitted, even when a typed
list[str]	config supplies a default. Override slice_meta for derived
list[str]	metadata or identity that includes validated defaults.

firecube.ingestor.api.BaseIngestor.slice_meta

slice_meta(ctx)

Return canonical slice metadata for this run's resume identity.

Parameters:

Name	Type	Description	Default
ctx	PluginContext	Run context containing the supplied options snapshot.	required

Returns:

Type	Description
dict[str, Any]	Canonicalized values for slice_meta_keys() present in
dict[str, Any]	ctx.options. Does not derive values or insert typed defaults.

firecube.ingestor.api.BaseIngestor.validation_group

validation_group(ctx)

Optional hook to derive a Zarr group path for validate_zarr.

firecube.ingestor.api.BaseIngestor.catalog_group_info

catalog_group_info(group, store_uri, storage_config=None)

Optionally annotate or hide one discovered catalog group.

firecube.ingestor.api.BaseIngestor.default_aggregate_metrics staticmethod

default_aggregate_metrics(ctx, state)

Default aggregate helper for plugins that do not need custom policy.

Runtime Context

RuntimeIngestContext is the engine-owned context used to run the pipeline. It is not the context type passed to plugin hooks and must not be reused across runs.

firecube.ingestor.api.RuntimeIngestContext dataclass

Bases: IngestContext

Engine-owned runtime context copied from IngestContext.

This carries internal execution-only state and must never be reused across runs.

Attributes

temp_root property writable

temp_root

Per-run workspace root used for materialization and temporary files.

force_reingest property writable

force_reingest

Whether this run is allowed to overwrite existing slice data.

incremental property writable

incremental

Whether plugin logic should prefer incremental update behavior.

dry_run property writable

dry_run

Whether side-effecting writes should be suppressed when supported.

Methods:

from_ingest_context classmethod

from_ingest_context(
    ctx, *, run_id, temp_root, materializer
)

Build an isolated runtime copy without mutating the caller context.

materialize

materialize(source)

Ensure the source file is available locally and return its path.

If the file is remote (e.g. S3), it will be downloaded to the per-run cache. If it's already local, the path is returned directly.

See Also

• Custom pipeline plugins
• Plugin extensions
• Context & Results
• Index Specification
• Package and Register a Plugin