Pipelines¶
The pipelines, memory ingestion, sync adapters, and shutdown handling. The pipelines, ingestors, and adapters are importable from sci_etl_core.
AsyncETLPipeline(
extractor: AsyncExtractor,
relevance_filter: AsyncRelevanceFilter,
entity_extractor: AsyncEntityExtractor,
exporter: AsyncExporter,
state_manager: AsyncStateManager,
destination: str,
max_concurrency: int = 6,
logger: Callable[[str], None] | None = None,
sleep: Any = sleep,
closeables: Iterable[Any] | None = None,
memory_ingestor: MemoryIngestor | None = None,
shutdown: ShutdownSignal | None = None,
on_event: Callable[[PipelineEvent], None] | None = None,
usage_sources: Iterable[Any] = (),
clock: Callable[[], float] = monotonic,
)
Take listed records through relevance, full text, memory, entity extraction, and export.
Records on a page run concurrently, up to max_concurrency at a time. An
irrelevant record is marked processed at once. A relevant one has its full
text fetched and stored through the optional memory_ingestor, its
entities extracted and exported, and only then is marked processed, so a
record that fails is retried on the next run. :meth:run describes paging,
limits, and aborts.
Wire the pipeline's collaborators together.
memory_ingestor is any :class:~sci_etl_core.ingest_protocol.MemoryIngestor,
such as an AsyncChunkIngestor, an AsyncSearchIndexer, or an
AsyncCompositeIngestor feeding both. An exception it raises that is
in :data:~sci_etl_core.ingest_protocol.MEMORY_FAULTS is logged as
Memory ingest failed for <record_id>: <error>, and the record's
entities are still exported. Any other exception, including
:class:~sci_etl_core.exceptions.SearchQueryError, fails the record.
closeables are closed each time async with pipeline exits, and a
SQLite store reopens when it is used after that. List a store only when
the pipeline owns it, as in a one-shot script; an application that keeps
using its stores after a run closes them itself. logger receives
every log line, and sleep is awaited between pages.
shutdown makes a run stop cleanly on SIGINT or SIGTERM, or when
:meth:~sci_etl_core.signals.ShutdownSignal.request is called: see
:meth:run. The pipeline installs its handlers for the duration of each
run, unless an enclosing shutdown.guard() already has.
on_event receives a :mod:~sci_etl_core.observability event as the
run progresses: :class:~sci_etl_core.observability.RunStarted,
:class:~sci_etl_core.observability.PageFetched, one
:class:~sci_etl_core.observability.RecordFinished per record,
:class:~sci_etl_core.observability.PageFinished, and
:class:~sci_etl_core.observability.RunFinished. It is called on the
event loop, so it must not block. An exception it raises is logged as
Event handler failed: <error> and does not affect the run.
usage_sources are clients with a usage property, such as the
LLM client and embedder, whose tokens used during a run are reported in
:attr:last_run_metrics. clock measures durations.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
property
¶
last_run_metrics: RunMetrics | None
Metrics of the most recent run, however it ended, or None before the first run.
property
¶
shutdown: ShutdownSignal | None
The shutdown signal this pipeline stops on, if any.
property
¶
Resources whose aclose the owning facade should await on teardown.
classmethod
¶
from_config(
pipeline: PipelineConfig, **arguments: Any
) -> AsyncETLPipeline
Build a pipeline whose max_concurrency comes from the pipeline config section.
arguments are the other constructor arguments, the collaborators
among them, and may override max_concurrency. Pass the section's
:meth:~sci_etl_core.config.PipelineConfig.run_arguments to
:meth:run.
async
¶
run(
query: str,
page_size: int | None = None,
sleep_between: float = 0.0,
total_limit: int | None = None,
max_records: int | None = None,
start_index: int | None = None,
newest_first: bool = False,
) -> int
Process listings until the ceiling is reached or the source is exhausted.
Paging resumes from the offset saved by the state manager unless
start_index is given. Pass 0 to rescan a listing whose order has
shifted since the last run: processed records are skipped by id, so a
rescan costs listing requests but never reprocesses a record.
Pass newest_first=True for a listing that puts new submissions
first, such as arXiv's. The run pages from offset 0 until it reaches
the records the previous run saw at the top of the listing, then jumps
to the saved offset moved down by the number of new submissions, less
one page to absorb entries removed from the listing, so the records in
between are not listed again. A run without saved head
records, such as the first one in this mode, rescans from offset 0, as
does a run that never finds them. The head records are saved through
:class:~sci_etl_core.models.PipelineMetadata next to the offset.
The saved offset only moves past pages whose every record was settled.
Once a record fails, or is deferred because total_limit was reached,
the offset stays at the start of that page for the rest of the run, so
the next run revisits the unsettled record instead of skipping it.
total_limit is exact: no more relevant records are processed than
it allows. max_records is a deprecated alias that sets both
page_size and total_limit when they are not given. Records
whose record_id is missing or blank cannot be tracked and are
skipped with a log message. sleep_between is waited between pages,
never after the page that reaches total_limit.
With a shutdown signal, a shutdown request stops the run cleanly.
Records already in flight are finished, records not yet started are
left for the next run, and a pending listing fetch or wait between
pages is cancelled. The saved offset does not move past the page that
was cut short, and :class:PipelineInterrupted is raised.
State is flushed through the state manager's flush whenever a run
ends, however it ends. A flush failure after the run itself failed is
logged, so it never hides the original error.
The only clean exits are an empty listing and reaching total_limit.
A page made up entirely of already-processed records is not the end of
the data, so paging moves past it. Any other interruption raises
:class:PipelineAborted carrying the count processed so far, so a
transport fault can never be mistaken for end-of-data.
A page on which records failed and none was processed is a stall. A single stall is tolerated, because one transient fault on a page of mostly irrelevant records says nothing about the source, and a later page that processes a record clears it.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
PipelineAborted
|
A listing could not be fetched or parsed, or records kept failing with none processed: on a second page before any progress, or on the last page before the listing ended. That signals a systemic fault, such as a rejected API key or an unwritable export, rather than one bad record, so the run stops instead of spending calls on every remaining page. |
PipelineInterrupted
|
A shutdown was requested through |
Synchronous facade for :class:AsyncETLPipeline.
The only blocking entrypoint. All collaborators injected into the
constructor must be async implementations. A full run is unbounded by
default; run_timeout imposes a ceiling when the caller wants one.
property
¶
last_run_metrics: RunMetrics | None
Metrics of the most recent run, as :attr:AsyncETLPipeline.last_run_metrics.
classmethod
¶
from_config(
pipeline: PipelineConfig, **arguments: Any
) -> ETLPipeline
Build a pipeline configured as :meth:AsyncETLPipeline.from_config describes.
Run :meth:AsyncETLPipeline.run and block until it ends.
With a shutdown signal, its handlers are installed on the calling
thread, which must be the main thread for signals to reach them, and
set the flag on the background loop, so Ctrl+C stops the run as
described for the async pipeline.
module-attribute
¶
MEMORY_FAULTS: tuple[type[Exception], ...] = (
EmbeddingError,
EmbeddingStoreError,
SearchStoreError,
)
The storage and embedding faults a memory ingest logs instead of failing the record.
SearchStoreError is listed rather than SearchError, so a SearchQueryError is never mistaken for a
storage fault.
Bases: Protocol
Stores a relevant record's full text in a memory backend.
A memory backend is the vector memory, the text search index, or both
through :class:~sci_etl_core.ingest_async.AsyncCompositeIngestor. The
exceptions in :data:MEMORY_FAULTS are storage or embedding faults: the
pipeline logs them and still exports the record's entities. Any other
exception, including a :class:~sci_etl_core.exceptions.SearchQueryError,
fails the record.
AsyncCompositeIngestor(
*ingestors: MemoryIngestor,
logger: Callable[[str], None] | None = None,
)
Send one record's text to several memory backends concurrently.
A memory fault (:data:~sci_etl_core.ingest_protocol.MEMORY_FAULTS) in one
backend is logged and does not stop the others, so a broken text index
never costs a record its embeddings, and never costs it its entity export.
Any other exception is re-raised after every backend has finished, the
first in ingestor order when several fail. Cancelling the awaiting task
cancels every backend.
Log lines read Memory ingest failed for <record_id> in <IngestorClass>:
<error>. Pass the pipeline's own logger, so memory faults share one
stream. The composite borrows its ingestors and never closes anything.
Combine ingestors, which then run concurrently on every record.
ingest returns the first ingestor's count, so the first ingestor
must not be an AsyncSearchIndexer; pass the chunk ingestor first.
The pipeline ignores the return value.
Raises:
| Type | Description |
|---|---|
ValueError
|
|
SyncExtractorAdapter(extractor: Extractor)
Bases: AsyncExtractor
Expose a synchronous extractor through the async extractor contract.
Per-record full-text retrieval is dispatched to a worker thread so the orchestrator keeps real fan-out; listing calls stay on the loop because they are inherently sequential.
SyncRelevanceFilterAdapter(
relevance_filter: RelevanceFilter,
)
Bases: AsyncRelevanceFilter
SyncEntityExtractorAdapter(
entity_extractor: EntityExtractor,
)
SyncExporterAdapter(exporter: Exporter)
Bases: AsyncExporter
Serialize exports on the loop thread so concurrent writes cannot interleave.
SyncStateManagerAdapter(state_manager: StateManager)
module-attribute
¶
DEFAULT_SIGNALS: tuple[Signals, ...] = tuple(
member
for member in (
getattr(signal, "SIGINT", None),
getattr(signal, "SIGTERM", None),
)
if member is not None
)
ShutdownSignal(
signals: Iterable[Signals] = DEFAULT_SIGNALS,
logger: Callable[[str], None] | None = None,
)
Cooperative shutdown flag driven by OS termination signals.
The first signal sets the flag so the owner can flush pending state. A second signal restores the previous handler and re-raises, letting the default hard termination proceed. Handlers are installed only from the main thread, so the synchronous bridge loop is unaffected, and the exact handler in force beforehand is put back on exit.
Pass one to :class:~sci_etl_core.pipeline_async.AsyncETLPipeline or
:class:~sci_etl_core.pipeline.ETLPipeline as shutdown and the pipeline
installs the handlers for each run and stops cleanly when the flag is set.
guard(
loop: AbstractEventLoop | None = None,
) -> Iterator[ShutdownSignal]
Install handlers for the duration of the block and restore them after.
Guards nest: only the outermost block installs and restores handlers, so
a pipeline given this signal can run inside a caller's own guard.
loop is passed to :meth:install.
install(loop: AbstractEventLoop | None = None) -> None
Install the handlers from the main thread.
Without loop, the handlers serve the running event loop. With
loop, plain OS handlers set the flag on that loop, which may run on
another thread; this lets a thread that blocks while a background loop
does the work, as :class:~sci_etl_core.pipeline.ETLPipeline does,
receive signals for it.
Remove the handlers :meth:install added and restore the ones they replaced.
Structured progress events and run metrics for :class:~sci_etl_core.pipeline_async.AsyncETLPipeline.
module-attribute
¶
RecordOutcome = Literal[
"processed",
"irrelevant",
"deferred",
"failed",
"skipped",
]
module-attribute
¶
RunOutcome = Literal[
"completed",
"aborted",
"interrupted",
"cancelled",
"failed",
]
module-attribute
¶
PipelineEvent = (
RunStarted
| PageFetched
| RecordFinished
| PageFinished
| RunFinished
)
dataclass
¶
RunMetrics(
pages: int = 0,
listed: int = 0,
processed: int = 0,
irrelevant: int = 0,
deferred: int = 0,
failed: int = 0,
skipped: int = 0,
entities_exported: int = 0,
memory_faults: int = 0,
duration_seconds: float = 0.0,
token_usage: TokenUsage | None = None,
outcome: RunOutcome | None = None,
)
Counts and timings for one pipeline run.
listed counts listing entries across every page fetched, and
processed, irrelevant, deferred, failed, and skipped
count records by :data:RecordOutcome. entities_exported counts the
entities handed to the exporter, and memory_faults the memory ingest
faults that were logged without failing their record.
duration_seconds is measured on a monotonic clock. token_usage is
what the pipeline's usage_sources used during the run, or None
when it has none. outcome is None while the run is in progress.
dataclass
¶
dataclass
¶
dataclass
¶
RecordFinished(
record_id: str,
title: str,
outcome: RecordOutcome,
duration_seconds: float,
entities: int = 0,
error: BaseException | None = None,
)
A record left the pipeline for this run.
entities is how many entities were exported for it, and error is
what failed it when outcome is "failed". A "skipped" record had
no record_id to track it by. duration_seconds counts from the moment
the record got a concurrency slot, and is 0 for a skipped record.
dataclass
¶
PageFinished(
offset: int,
duration_seconds: float,
metrics: RunMetrics = RunMetrics(),
)
Every record of the page at offset finished; metrics is the run so far.
class-attribute
instance-attribute
¶
metrics: RunMetrics = field(default_factory=RunMetrics)
dataclass
¶
RunFinished(metrics: RunMetrics)
The run ended, however it ended; metrics.outcome says how.