extended_data.workflows.sync

Data sync primitives for local structured artifacts.

Module Contents

Classes

DataSyncResult

Result from a local data sync operation.

Functions

sync_value_to_file

Sync a value to a local file only when the rendered bytes change.

sync_file_to_file

Read a structured file and sync its decoded value to another file.

_sync_encoding

_render_sync_value

_read_existing_text

API

class extended_data.workflows.sync.DataSyncResult

Result from a local data sync operation.

source: extended_data.containers.ExtendedString = None
destination: extended_data.containers.ExtendedString = None
changed: bool = None
dry_run: bool = False
bytes_written: int = 0
encoding: extended_data.containers.ExtendedString = 'field(...)'
output_path: pathlib.Path | None = None
metadata: extended_data.containers.ExtendedDict = 'field(...)'
to_dict() extended_data.containers.ExtendedDict

Return this result as promoted Extended Data.

extended_data.workflows.sync.sync_value_to_file(value: Any, file_path: extended_data.io.files.FilePath, *, source: str = 'memory', encoding: str | None = None, charset: str = 'utf-8', allow_empty: bool = False, dry_run: bool = False, tld: pathlib.Path | None = None, metadata: collections.abc.Mapping[str, Any] | None = None) extended_data.workflows.sync.DataSyncResult

Sync a value to a local file only when the rendered bytes change.

extended_data.workflows.sync.sync_file_to_file(source_path: extended_data.io.files.FilePath, destination_path: extended_data.io.files.FilePath, *, suffix: str | None = None, encoding: str | None = None, charset: str = 'utf-8', errors: str = 'strict', allow_empty: bool = False, dry_run: bool = False, tld: pathlib.Path | None = None) extended_data.workflows.sync.DataSyncResult

Read a structured file and sync its decoded value to another file.

extended_data.workflows.sync._sync_encoding(file_path: extended_data.io.files.FilePath, encoding: str | None) str
extended_data.workflows.sync._render_sync_value(value: Any, *, allow_encoding: str) str
extended_data.workflows.sync._read_existing_text(path: pathlib.Path, *, charset: str) str | None