angee.workflows_integrate.steps

Vendor-free archive extraction workflow steps.

Archive extractor classes arrive through ANGEE_WORKFLOW_ARCHIVE_EXTRACTOR_CLASSES. Each extractor recognizes a storage.File with a hard boolean result and executes through the target domain's own idempotent ingest surface. The workflow steps only orchestrate that contract: probe emits proposals, gate authors the serializable mapping form, and execute prepares and runs the stock MapStep units.

ARCHIVE_EXTRACTOR_CLASSES_SETTING

Settings mapping from stable archive extractor keys to trusted class paths.

ArchiveExecutionReporter

@dataclass(frozen=True, slots=True)
class ArchiveExecutionReporter()

Workflow-owned progress reporter passed to one extractor execution.

The current engine has one durable progress primitive: the step heartbeat. Extractors call :meth:heartbeat during long ingest work; richer progress remains an engine concern rather than vendor state hidden in this addon. The wrapper is a deliberate capability-narrowing boundary: vendor extractor code receives only this reporter, never the StepImpl/step_run surface — mark_failed, resume_state, and the engine verbs stay workflow-owned.

heartbeat

def heartbeat(*, at: datetime | None = None) -> None

Refresh the mapped step-run heartbeat.

ArchiveExtractor

class ArchiveExtractor(ImplBase, ABC)

Base contract for a settings-registered archive extractor.

Subclasses declare a stable :attr:key, a human :attr:label, target_resource as an app_label.Model string, and subject_resource — the storage container kind the run subject is, a storage.File (an archive to open) or a storage.Drive (a mounted tree to inspect). recognizes(subject) returns a real :class:bool — confidence scores and truthy substitutes are not accepted. execute(subject, target_pk, reporter) must land content via the target domain's own idempotent ingest API and return JSON-safe journal output. Vendor parsing and target-domain identity rules stay on the concrete extractor and its owning addon.

recognizes

@abstractmethod
def recognizes(subject: Any) -> bool

Return whether subject is a container this extractor owns.

subject is the run subject named by :attr:subject_resource. The probe only invokes extractors whose subject_resource matches the run subject, so a file extractor never sees a drive. Recognition must stay bounded: read only a header/prefix of a file, or a metadata lookup plus a bounded manifest probe of a drive — the probe runs every matching extractor against the same subject.

execute

@abstractmethod
def execute(subject: Any, target_pk: str,
            reporter: ArchiveExecutionReporter) -> Any

Idempotently ingest subject into target_pk and return journal output.

archive_extractor_classes

def archive_extractor_classes() -> tuple[type[ArchiveExtractor], ...]

Return configured extractor classes in deterministic stable-key order.

archive_extractor_class

def archive_extractor_class(key: str) -> type[ArchiveExtractor]

Resolve and validate the extractor registered as key.

ArchiveProbeStepImpl

class ArchiveProbeStepImpl(StepImpl)

Probe a workflow run's storage file with every configured extractor.

run

def run(step_run: Any, *, now: datetime) -> StepResult

Return stable extractor proposals or the routable failed outcome.

ArchiveGateStepImpl

class ArchiveGateStepImpl(StepImpl)

Suspend for a fixed-row extractor-to-target mapping decision.

v1 renders one shared rows template, so every recognized extractor must declare the same target resource; heterogeneous archives route down the failed edge until per-resource row grouping ships.

validate_config

@classmethod
def validate_config(cls, config: Any) -> None

Validate optional decision action, assignee, and attempt settings.

run

def run(step_run: Any, *, now: datetime) -> StepResult

Author the mapping form from probe output and suspend one decision.

ArchiveExecuteStepImpl

class ArchiveExecuteStepImpl(StepImpl)

Prepare a confirmed decision mapping or execute one stock-map unit.

A workflow uses this implementation twice: mode=prepare follows the archive gate and turns its completed decision into a plain mapping list; the built-in map step consumes that list and targets a second step with mode=unit. This keeps decision lookup outside the generic map engine and preserves its existing per-unit partial-failure accounting.

validate_config

@classmethod
def validate_config(cls, config: Any) -> None

Require an explicit prepare/unit execution mode.

run

def run(step_run: Any, *, now: datetime) -> StepResult

Prepare confirmed mappings or execute the mapped extractor unit.