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, failure counts, and metadata once per run, marks each settled record as processed, possibly concurrently, records failed attempts after the page that failed them, saves metadata after every page, and calls :meth:flush when the run ends, however it ends.

:meth:record_failure and :meth:failure_counts have defaults that track nothing, so a state manager that does not override them never quarantines a record.

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.

record_failure async

record_failure(record_id: str, error: str) -> int

Count a failed attempt for record_id and return its attempts so far.

error describes the failure. The default tracks nothing and returns 0.

failure_counts async

failure_counts() -> dict[str, int]

Return the attempts per record id that has failed and not been marked processed since.

The default returns an empty mapping.

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

SCHEMA_VERSION module-attribute

SCHEMA_VERSION = 2

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.

The metadata file is JSON with a schema_version; a file without one was written by sci-etl-core 0.4, and its saved offset is read as the cursor. Failed attempts are kept in the metadata file per record, with the last error truncated to 4,096 characters, and :meth:mark_processed clears them.

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 and clear its failed attempts.

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.
StateStoreError	The record had failed attempts and the metadata file was written by a newer sci-etl-core.

record_failure async

record_failure(record_id: str, error: str) -> int

Count a failed attempt and keep error, truncated to 4,096 characters; an empty id is ignored.

Raises:

Type	Description
OSError	The metadata file cannot be read or written.
StateStoreError	The metadata file was written by a newer sci-etl-core.

failure_counts async

failure_counts() -> dict[str, int]

Return the attempts per record id that has failed and not been marked processed since.

Raises:

Type	Description
OSError	The metadata file exists but cannot be read.
StateStoreError	The metadata file was written by a newer sci-etl-core.

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 the first page 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.
StateStoreError	The metadata file was written by a newer sci-etl-core.

save_metadata async

save_metadata(metadata: PipelineMetadata) -> None

Persist metadata, keeping the failed attempts already recorded.

Raises:

Type	Description
OSError	The metadata file cannot be read or written.
StateStoreError	The metadata file was written by a newer sci-etl-core.

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 or one short transaction, 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, and the file records its schema version in PRAGMA user_version. A database written by sci-etl-core 0.4 is upgraded in place, its saved offset becoming the cursor. Failed attempts are stored per record with the last error, truncated to 4,096 characters, and :meth:mark_processed clears them. timeout is how many seconds a statement waits for another process's lock. A :class:sqlite3.Error propagates unwrapped, and a database written by a newer sci-etl-core raises :class:~sci_etl_core.exceptions.StateStoreError. 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

Record an id as processed and clear its failed attempts; an empty id is ignored.

record_failure async

record_failure(record_id: str, error: str) -> int

Count a failed attempt and keep error, truncated to 4,096 characters; an empty id is ignored.

failure_counts async

failure_counts() -> dict[str, int]

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.