Exporters

Every name on this page is importable from sci_etl_core.exporters. The claim store exporter is on the Claims page.

sci_etl_core.exporters.async_base

E module-attribute

E = TypeVar('E', contravariant=True)

AsyncExporter

Bases: ABC, Generic[E]

A sink for one run's entities.

:class:~sci_etl_core.pipeline_async.AsyncETLPipeline calls :meth:open once when a run starts, :meth:write once per processed record, possibly for several records concurrently, :meth:flush after each page, and :meth:aclose when the run ends, however it ends. An exporter takes its destination when it is constructed.

Delivery is at least once: a crash can repeat a record, never lose one. With durable_writes = True, :meth:write returns only once the entities are durable, and the pipeline marks the record processed right after it. With durable_writes = False, :meth:write may buffer, and the pipeline marks the page's written records processed only after :meth:flush returns.

durable_writes class-attribute

durable_writes: bool = True

open async

open() -> None

Prepare the sink before the run's first listing request; the default does nothing.

A fault here aborts the run before any listing request.

write abstractmethod async

write(record: RawRecord, entities: Sequence[E]) -> None

Store entities for record.

The pipeline calls it for every processed record, including one with no entities, so a sink can clear rows a re-extracted record no longer has. It must be idempotent: repeating a write with the same arguments, including after one that raised partway, leaves the sink as a single successful write would. Each sink documents which write wins when a repeat carries different entities.

A fault here fails the record, which is retried on the next run and counts one failed attempt.

flush async

flush() -> None

Make every earlier write durable; the default does nothing.

A fault here leaves the records written since the last successful flush unsettled, counts no failed attempt against them, and makes the page a stalled page.

aclose async

aclose() -> None

Finish the run's output and release resources; the default does nothing.

The pipeline calls :meth:flush first. A fault here is raised only when the run itself succeeded.

sci_etl_core.exporters.csv_async

RECORD_COLUMN module-attribute

RECORD_COLUMN = 'record_id'

EXTRA_COLUMN module-attribute

EXTRA_COLUMN = 'extra'

Row module-attribute

Row = tuple[str, ...]

AsyncCsvExporter

AsyncCsvExporter(
    path: str | Path,
    columns: Sequence[str],
    *,
    escape_formulas: bool = True,
)

Bases: AsyncExporter[Any]

Write every entity as one CSV row, tagged with the record it came from.

The table is long: the first column is record_id, then columns in order, then extra, which holds any other keys of the entity as a JSON object, so no extracted value is dropped. Values are written as text, unchanged: nothing is clipped, coerced to a number, or merged across records, so two papers that report one object give two rows and a conflict stays visible. Merge or deduplicate after the run, for example with :class:~sci_etl_core.processors.dedup.DeduplicationStep.

Writing a record replaces all of its rows, so the last write for a record wins, and a record written with no entities has no rows. A repeated write is idempotent.

The CSV file is rendered once per run, when the run closes the exporter. During the run, each :meth:flush appends the page's records to <path>.journal and calls fsync, so durable_writes is False. A reader mid-run sees the previous run's table. A run that crashed leaves the journal behind, and the next :meth:open replays it into the table before anything else happens. Total work per run is linear in the number of rows.

Cells a spreadsheet would read as a formula are written with a leading apostrophe and restored when the file is read back; plain numbers such as -5.361 are written unchanged. Escaping is on unless escape_formulas is False. Entities may be dicts, Pydantic models, dataclass instances, or claims.

Configure the exporter.

Raises:

Type	Description
ValueError	columns is empty, repeats a name, or names record_id or extra.

durable_writes class-attribute instance-attribute

durable_writes = False

path property

path: Path

The CSV file.

journal_path property

journal_path: Path

The journal each flush appends to, deleted once the run's table is rendered.

header property

header: Row

The CSV header: record_id, the configured columns, and extra.

open async

open() -> None

Load the existing table and replay a journal a crashed run left behind.

Raises:

Type	Description
ExportError	The CSV file or journal cannot be read, or the CSV's header differs from :attr:header.

write async

write(record: RawRecord, entities: Sequence[Any]) -> None

Replace the record's rows and queue them for the next :meth:flush.

Raises:

Type	Description
TypeError	An entity is of a type that cannot be exported.

flush async

flush() -> None

Append the queued records to the journal and fsync it.

Raises:

Type	Description
ExportError	The journal could not be written; the records stay queued.

aclose async

aclose() -> None

Render the table, replace the CSV atomically, and delete the journal.

Queued records are journaled first. After a fault the journal stays, and the next :meth:open replays it.

Raises:

Type	Description
ExportError	The journal or the CSV file could not be written.

sci_etl_core.exporters.jsonl_async

AsyncJsonlExporter

AsyncJsonlExporter(path: str | Path, *, fsync: bool = True)

Bases: AsyncExporter[Any]

Append one JSON line per written record to a file.

Each line is an object with the record's record_id, title, and source_url, and its entities as a list, possibly empty. Nothing is merged: two papers that report one object give two lines, so conflicting values survive for review.

Writes are buffered and appended on :meth:flush, which also calls fsync, so durable_writes is False. A record written again, in this run or a later one, appends another line; the last line for a record_id wins, which :func:read_jsonl_export applies. A repeated write therefore changes the file's bytes but not what a reader sees. A record whose last line has no entities holds none, so a re-extraction that finds nothing clears the record.

Entities may be dicts, Pydantic models, dataclass instances, or claims. The file and its parent folder are created when the run opens the exporter.

Configure the exporter; fsync=False skips the fsync after each flush, for tests and scratch runs.

durable_writes class-attribute instance-attribute

durable_writes = False

path property

path: Path

The file lines are appended to.

open async

open() -> None

Open the file for appending.

Raises:

Type	Description
ExportError	The file cannot be opened.

write async

write(record: RawRecord, entities: Sequence[Any]) -> None

Queue the record's line for the next :meth:flush.

Raises:

Type	Description
TypeError	An entity is of a type that cannot be exported.

flush async

flush() -> None

Append the queued lines and fsync; after a fault the lines stay queued and the file is truncated back.

Raises:

Type	Description
ExportError	The lines could not be written.

aclose async

aclose() -> None

Flush any queued lines and close the file; a later :meth:open reopens it.

read_jsonl_export

read_jsonl_export(
    path: str | Path,
) -> dict[str, list[dict[str, Any]]]

Read a file :class:AsyncJsonlExporter wrote, keeping the last line for each record_id.

Records come back in the order of their first line. A blank line is skipped.

Raises:

Type	Description
ExportError	A line is not a JSON object with record_id and entities.