Writers and transactions

Artifact writer helpers for lightweight catalog workflows.

Writers bridge Catalog.add_artifact and concrete storage code. A writer is any object that satisfies ogcat.hooks.ArtifactWriter: it exposes a write(context, source, target) method, receives an ogcat.hooks.OperationContext, and materialises an ogcat.models.ArtifactLocator from an ogcat.hooks.OperationSource.

This module provides small adapters for common examples. source_writer wraps a function that accepts an OperationSource and target Path. memory_writer and path_writer adapt narrower functions for in-memory payloads and local files. UnzipArtifactWriter is a concrete directory writer used by tutorials and tests. UnzipSingleFileArtifactWriter extracts one zip member to a file target. These helpers register rollback actions through the operation context before writing so partially-created targets can be cleaned up if the catalog operation fails.

Writer protocol

Writer classes do not need to inherit from a concrete base class. They satisfy the structural ogcat.hooks.ArtifactWriter protocol by providing a write(context, source, target) method.

class ogcat.hooks.ArtifactWriter(*args, **kwargs)

Bases: Protocol

Plugin-facing writer that materialises artifact data before record write.

write(context, source, target)

Write artifact data from source to target.

Return type:

None

Writer helpers

ogcat.writers.memory_source(data, *, kind='memory', descriptor=None, metadata=None)

Build an operation source carrying an in-memory Python object.

Parameters:

• data (object) – In-memory object to pass to a memory writer.

• kind (str) – Source kind used for writer validation.

• descriptor (str | None) – Optional human-readable source description.

• metadata (MetadataDict | None) – Optional JSON-compatible source metadata.

Return type:

OperationSource

Returns:

An operation source with payload set to data.

ogcat.writers.path_source(path, *, kind='path', descriptor=None, metadata=None)

Build an operation source carrying a local path.

Parameters:

• path (str | Path) – Local source path.

• kind (str) – Source kind used for writer validation.

• descriptor (str | None) – Optional human-readable source description. Defaults to the resolved source path string.

• metadata (MetadataDict | None) – Optional JSON-compatible source metadata.

Return type:

OperationSource

Returns:

An operation source with path set to the resolved local path.

ogcat.writers.source_writer(write_function, *, target_kind, source_kind=None)

Wrap a function that writes from an OperationSource to a target path.

Return type:

FunctionArtifactWriter

ogcat.writers.memory_writer(write_function, *, target_kind, source_kind='memory')

Wrap a function that writes an in-memory payload to a target path.

Return type:

FunctionArtifactWriter

ogcat.writers.path_writer(write_function, *, target_kind, source_kind='path')

Wrap a function that writes from a source path to a target path.

Return type:

FunctionArtifactWriter

FunctionArtifactWriter

class ogcat.writers.FunctionArtifactWriter(write_function, target_kind, source_kind=None, write_mode='write')

Artifact writer adapter around a small Python function.

The wrapped function receives an OperationSource and a local target path. It should write either a single file or a directory, matching target_kind. If it returns metadata, the metadata is merged into context.derived_metadata.

Parameters:

• write_function (Callable[[OperationSource, Path], MetadataDict | None]) – Function that writes artifact data.

• target_kind (Literal['file', 'directory']) – Whether the target locator should be treated as a file or directory.

• source_kind (str | None) – Optional required source kind.

Attributes

FunctionArtifactWriter.write_function

Callable that writes from an OperationSource to a target path and may return MetadataDict derived metadata.

FunctionArtifactWriter.target_kind

Target artifact kind, either "file" or "directory".

FunctionArtifactWriter.source_kind

Optional required source kind. When set, write rejects sources whose OperationSource.kind does not match.

Methods

FunctionArtifactWriter.write(context, source, target)

Write artifact data and register rollback for the created target.

Return type:

None

CopyArtifactWriter

class ogcat.writers.CopyArtifactWriter(source_kind='local_file')

Artifact writer that copies a local source path to a storage target.

write_mode: ClassVar[Literal['copy', 'move', 'write', 'reference']] = 'copy'

write(context, source, target)

Copy a local source path to the target and register rollback.

Return type:

None

MoveArtifactWriter

class ogcat.writers.MoveArtifactWriter(source_kind='local_file')

Artifact writer that moves a local source path to a storage target.

write_mode: ClassVar[Literal['copy', 'move', 'write', 'reference']] = 'move'

write(context, source, target)

Move a local source path to the target and register rollback.

Return type:

None

UnzipArtifactWriter

class ogcat.writers.UnzipArtifactWriter(source_kind='zip_file')

Example writer that safely extracts a zip file into a directory target.

Attributes

UnzipArtifactWriter.source_kind

Required source kind for zip extraction. Defaults to "zip_file".

Methods

UnzipArtifactWriter.write(context, source, target)

Extract a zip source into the target directory.

Return type:

None

UnzipSingleFileArtifactWriter

class ogcat.writers.UnzipSingleFileArtifactWriter(member_name=None, source_kind='zip_file')

Extract a single zip member to a file target.

Parameters:

• member_name (str | None) – Optional archive member to extract. When omitted, the archive must contain exactly one non-directory member.

• source_kind (str) – Required operation source kind.

Attributes

UnzipSingleFileArtifactWriter.member_name

Optional archive member to extract. When omitted, the archive must contain exactly one non-directory member.

UnzipSingleFileArtifactWriter.source_kind

Required source kind for zip extraction. Defaults to "zip_file".

Methods

UnzipSingleFileArtifactWriter.write(context, source, target)

Extract one zip source member to the target file.

Return type:

None

Transactions

Best-effort catalog transaction helpers.

The TinyDB-backed implementation uses compensating rollback actions. This is a unit-of-work helper, not a true database transaction or ACID boundary.

UnitOfWork coordinates record writes with cleanup callbacks registered by artifact writers and hooks. Staged work starts in PLANNED state, moves to STAGED when a record or rollback action is registered, and becomes COMMITTED only after all catalog and artifact work has succeeded. If the operation exits early, rollback actions run in reverse order and failures are captured as RollbackFailure entries while preserving the original exception.

class ogcat.UnitOfWork(repository, *, operation_id=None)

Best-effort unit of work for catalog record staging and cleanup.

Parameters:

• repository (CatalogRepository) – Repository used for staged catalog record writes.

• operation_id (str | None) – Optional externally supplied operation id. If omitted, a random id is generated for audit/debug correlation.

register_rollback(action, *, description=None)

Register an action to run if the unit of work rolls back.

Parameters:

• action (RollbackAction | Callable[[], None]) – A rollback action object or a no-argument callable.

• description (str | None) – Human-readable cleanup description for callable actions. Ignored for action objects that already carry one.

Return type:

RollbackAction

Returns:

The registered rollback action.

insert_staged_record(record)

Insert a staged record and delete it if the unit of work rolls back.

Return type:

CatalogRecord

stage_record(record)

Alias for insert_staged_record.

Return type:

CatalogRecord

commit()

Commit staged work and discard rollback actions.

Return type:

None

rollback(*, original_exception=None)

Run registered rollback actions in reverse order.

Parameters:

original_exception (BaseException | None) – Optional exception that triggered rollback. When supplied, rollback failure summaries are added as exception notes so the original failure remains visible to callers.

Return type:

None

class ogcat.OperationState(value)

Bases: StrEnum

Lifecycle state for a best-effort catalog operation.

PLANNED = 'planned'

STAGED = 'staged'

COMMITTED = 'committed'

ROLLED_BACK = 'rolled_back'

FAILED = 'failed'