State

Every name on this page is importable from sci_etl_core.state.

sci_etl_core.state.async_base

AsyncStateManager

Bases: ABC

Contract for the durable record of which records a run has settled.

A pipeline loads the processed ids and metadata once per run, marks each settled record as processed, possibly concurrently, saves metadata after every page, and calls :meth:flush when the run ends, however it ends.

load_processed_ids abstractmethod async

load_processed_ids() -> set[str]

Return the set of record ids already processed.

mark_processed abstractmethod async

mark_processed(record_id: str) -> None

Persist a record id as processed.

load_metadata abstractmethod async

load_metadata() -> PipelineMetadata

Return the last saved pipeline metadata.

save_metadata abstractmethod async

save_metadata(metadata: PipelineMetadata) -> None

Persist pipeline metadata, stamping the current run time.

flush async

flush() -> None

Force buffered state to durable storage before termination.

Backends that commit on every mutation need no action, so the default is a no-op.

sci_etl_core.state.async_file_state

AsyncFileStateManager

AsyncFileStateManager(
    processed_ids_file: str | Path,
    metadata_file: str | Path,
)

Bases: AsyncStateManager

Plain-file state backend hardened against concurrent access.

In-process callers are serialized by an :class:asyncio.Lock; other processes are excluded by an OS-level advisory lock held for the whole of every read or append. Metadata is published by atomic rename, so an interrupted write cannot truncate it. Blocking file work runs in a worker thread to keep the loop responsive.

A file that cannot be read raises OSError rather than reading as empty, because an empty processed-id set would silently reprocess every record.

load_processed_ids async

load_processed_ids() -> set[str]

Return the processed ids.

Raises:

Type	Description
OSError	The ids file exists but cannot be read.

mark_processed async

mark_processed(record_id: str) -> None

Record an id as processed.

An empty or whitespace-only id carries nothing to track and is ignored.

Raises:

Type

Description

ValueError

record_id has leading or trailing whitespace, which the file cannot store faithfully: lines are stripped when read, so the id would come back changed, never match the record again, and the record would be reprocessed on every run. Also raised when record_id contains a line boundary, meaning any character :meth:str.splitlines splits on (\n, \r, \x0b, \x0c, \x1c-\x1e, \x85, \u2028, \u2029). The file stores one id per line, so writing it would be read back as several distinct ids, silently marking records that were never seen.

load_metadata async

load_metadata() -> PipelineMetadata

Return the saved metadata.

Content that is not valid metadata, such as a hand-edited file, yields the defaults: resuming from offset 0 rescans the listing, and ids already processed are still skipped, so nothing is lost.

Raises:

Type	Description
OSError	The metadata file exists but cannot be read.

save_metadata async

save_metadata(metadata: PipelineMetadata) -> None

sci_etl_core.state.sqlite_async

AsyncSqliteStateManager

AsyncSqliteStateManager(
    database_path: str | Path, timeout: float = 30.0
)

Bases: AsyncStateManager

Transactional state backend built on the standard-library sqlite3.

Every mutation is an autocommitted statement guarded by SQLite's own cross-process locking, so concurrent runs cannot interleave a half-written record. Blocking calls run in a worker thread and an :class:asyncio.Lock keeps the shared connection single-user, holding it until the thread finishes even if the awaiting task is cancelled.

The database file and its parent folder are created on first use. timeout is how many seconds a statement waits for another process's lock. A :class:sqlite3.Error propagates unwrapped. Use each instance from one event loop.

load_processed_ids async

load_processed_ids() -> set[str]

mark_processed async

mark_processed(record_id: str) -> None

load_metadata async

load_metadata() -> PipelineMetadata

save_metadata async

save_metadata(metadata: PipelineMetadata) -> None

flush async

flush() -> None

aclose async

aclose() -> None

Close the shared connection; a later call transparently reopens it.

sci_etl_core.state.base

StateManager

Bases: ABC

Blocking counterpart of :class:~sci_etl_core.state.async_base.AsyncStateManager.

Wrap an implementation in :class:~sci_etl_core._adapters.SyncStateManagerAdapter to use it in a pipeline. The blocking contract has no flush.

load_processed_ids abstractmethod

load_processed_ids() -> set[str]

Return the set of record ids already processed.

mark_processed abstractmethod

mark_processed(record_id: str) -> None

Persist a record id as processed.

load_metadata abstractmethod

load_metadata() -> PipelineMetadata

Return the last saved pipeline metadata.

save_metadata abstractmethod

save_metadata(metadata: PipelineMetadata) -> None

Persist pipeline metadata, stamping the current run time.