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 Parallel Zarr Writes.

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`	Hook for per-batch setup (e.g. DB connections). Cooperatively calls super.
`BaseIngestor.batch_teardown`	Hook for per-batch cleanup. Cooperatively calls super.
`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`	Option keys that define a logical slice for this plugin.
`BaseIngestor.slice_meta`	Return canonical slice metadata for this run.
`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, and .nc files plus extensionless files that look like HDF5. Patterns from the include_patterns engine option add matching files to that set; they do not replace it.

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.

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 include_patterns/ 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)

Hook for per-batch setup (e.g. DB connections). Cooperatively calls super.

firecube.ingestor.api.BaseIngestor.batch_teardown

batch_teardown(ctx)

Hook for per-batch cleanup. Cooperatively calls super.

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

Option keys that define a logical slice for this plugin.

firecube.ingestor.api.BaseIngestor.slice_meta

slice_meta(ctx)

Return canonical slice metadata for this run.

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.