extended_data.workflows

Tier 3 workflow composition over Extended Data primitives and containers.

Submodules

• extended_data.workflows.sync

Package Contents

Classes

WorkflowStep	A named transformation in a data workflow.
WorkflowResult	The completed value and audit trail for a data workflow.
DataWorkflow	Compose file decoding, transformations, and exports as a Tier 3 primitive.

Functions

list_data_transform_steps	Return the named transform steps supported by DataWorkflow.
data_transform_action	Return a workflow action for one named Tier 2 transform step.
_data_transform_method	Return the best method for a transform step on the current data shape.
_deep_merge_action	Return a typed workflow action that deep-merges mapping values.
_coerce_step	Normalize supported step declarations to WorkflowStep.
_decode_step_name	Return a stable step name for decode-only workflows.

Data

WorkflowAction
StepLike
DATA_TRANSFORM_STEPS

API

extended_data.workflows.WorkflowAction: TypeAlias = None

extended_data.workflows.StepLike: TypeAlias = 'WorkflowStep | tuple[str, WorkflowAction] | WorkflowAction'

extended_data.workflows.DATA_TRANSFORM_STEPS: collections.abc.Mapping[str, str] = 'MappingProxyType(...)'

extended_data.workflows.list_data_transform_steps() → tuple[str, ...]

Return the named transform steps supported by DataWorkflow.

extended_data.workflows.data_transform_action(step: str) → extended_data.workflows.WorkflowAction

Return a workflow action for one named Tier 2 transform step.

extended_data.workflows._data_transform_method(data: Any, step: str, method_name: str) → Any

Return the best method for a transform step on the current data shape.

class extended_data.workflows.WorkflowStep

A named transformation in a data workflow.

name: str = None

action: extended_data.workflows.WorkflowAction = None

class extended_data.workflows.WorkflowResult(value: Any, steps: collections.abc.Iterable[str] = (), output_path: pathlib.Path | None = None, metadata: collections.abc.Mapping[str, Any] | None = None)

The completed value and audit trail for a data workflow.

Initialization

Store workflow metadata as promoted detached data.

value: Any = None

steps: tuple[str, ...] = None

output_path: pathlib.Path | None = None

_metadata: extended_data.containers.ExtendedDict = 'field(...)'

property metadata: extended_data.containers.ExtendedDict

Return a detached promoted copy of workflow metadata.

as_builtin() → Any

Return the workflow value lowered to built-in Python containers.

as_extended() → Any

Return a detached workflow value promoted to Extended Data containers.

to_export_safe(*, export_to_yaml: bool = False) → Any

Return the workflow value converted to export-safe primitive data.

wrap_for_export(allow_encoding: bool | str = True, **format_opts: Any) → str

Return the workflow value wrapped as an encoded export string.

class extended_data.workflows.DataWorkflow(value: Any, *, steps: collections.abc.Iterable[str] = (), as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None)

Compose file decoding, transformations, and exports as a Tier 3 primitive.

Initialization

Create a workflow from an existing value.

property value: Any

Return the current workflow value.

property steps: tuple[str, ...]

Return the names of executed workflow steps.

property metadata: extended_data.containers.ExtendedDict

Return a detached promoted copy of workflow metadata.

classmethod from_value(value: Any, *, as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None) → extended_data.workflows.DataWorkflow

Start a workflow from an in-memory value.

classmethod from_data_file(artifact: extended_data.io.files.DataFile, *, as_extended: bool = True) → extended_data.workflows.DataWorkflow

Start a workflow from a decoded DataFile artifact.

classmethod decode(file_data: str | memoryview | bytes | bytearray, *, file_path: extended_data.io.files.FilePath | None = None, suffix: str | None = None, as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None) → extended_data.workflows.DataWorkflow

Start a workflow by decoding structured text or bytes.

classmethod from_file(file_path: extended_data.io.files.FilePath, *, suffix: str | None = None, as_extended: bool = True, charset: str = 'utf-8', errors: str = 'strict', tld: pathlib.Path | None = None) → extended_data.workflows.DataWorkflow

Read and decode a local file or URL into a workflow.

then(step: extended_data.workflows.StepLike, *, name: str | None = None, as_extended: bool | None = None) → extended_data.workflows.DataWorkflow

Apply one transformation and return the next workflow state.

run(*steps: extended_data.workflows.StepLike, as_extended: bool | None = None) → extended_data.workflows.DataWorkflow

Apply multiple transformations in order.

transform(*steps: str, as_extended: bool | None = None) → extended_data.workflows.DataWorkflow

Apply named Tier 2 transform steps in order.

merge(*mappings: collections.abc.Mapping[str, Any], name: str = 'merge', as_extended: bool | None = None) → extended_data.workflows.DataWorkflow

Deep-merge mappings into the current workflow value.

merge_file(file_path: extended_data.io.files.FilePath, *, suffix: str | None = None, charset: str = 'utf-8', errors: str = 'strict', tld: pathlib.Path | None = None, name: str | None = None, as_extended: bool | None = None) → extended_data.workflows.DataWorkflow

Read a structured file and deep-merge it into the workflow value.

as_builtin() → extended_data.workflows.DataWorkflow

Return the next workflow state with built-in Python containers.

as_extended() → extended_data.workflows.DataWorkflow

Return the next workflow state with Extended Data containers.

result() → extended_data.workflows.WorkflowResult

Return a completed workflow result without writing an output artifact.

write(file_path: extended_data.io.files.FilePath, *, encoding: str | None = None, charset: str = 'utf-8', allow_empty: bool = False, tld: pathlib.Path | None = None, as_builtin: bool = True) → extended_data.workflows.WorkflowResult

Write the current workflow value and return the completed result.

sync_file(file_path: extended_data.io.files.FilePath, *, encoding: str | None = None, charset: str = 'utf-8', allow_empty: bool = False, dry_run: bool = False, tld: pathlib.Path | None = None) → extended_data.workflows.sync.DataSyncResult

Sync the current workflow value to a local file if it changed.

extended_data.workflows._deep_merge_action(*mappings: collections.abc.Mapping[str, Any]) → extended_data.workflows.WorkflowAction

Return a typed workflow action that deep-merges mapping values.

extended_data.workflows._coerce_step(step: extended_data.workflows.StepLike, *, name: str | None = None) → extended_data.workflows.WorkflowStep

Normalize supported step declarations to WorkflowStep.

extended_data.workflows._decode_step_name(*, file_path: extended_data.io.files.FilePath | None, suffix: str | None) → str

Return a stable step name for decode-only workflows.