Pipelines¶
The pipelines, memory ingestion, and shutdown handling. The pipelines and ingestors are importable from sci_etl_core.
AsyncETLPipeline(
extractor: AsyncExtractor,
relevance_filter: AsyncRelevanceFilter,
entity_extractor: AsyncEntityExtractor[E],
exporter: AsyncExporter[E],
state_manager: AsyncStateManager,
*,
max_concurrency: int = 6,
sleep: Callable[[float], Awaitable[None]] = sleep,
closeables: Iterable[SupportsAclose] = (),
memory_ingestor: MemoryIngestor | None = None,
shutdown: ShutdownSignal | None = None,
on_event: Callable[[PipelineEvent], None] | None = None,
usage_sources: Iterable[UsageReporter] = (),
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 written to the exporter, and only once they are
durable is it marked processed, so a record that fails is retried on the
next run. E is the entity type the extractor produces and the exporter
accepts. :meth:run describes paging, limits, failures, and aborts.
Wire the pipeline's collaborators together.
The five collaborators are positional; every other argument is keyword-only.
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. sleep is awaited
between pages. Log lines go to the sci_etl_core.pipeline_async
logger.
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
¶
closeables: list[SupportsAclose]
Resources whose aclose the owning facade should await on teardown.
classmethod
¶
from_config(
pipeline: PipelineConfig,
extractor: AsyncExtractor,
relevance_filter: AsyncRelevanceFilter,
entity_extractor: AsyncEntityExtractor[E],
exporter: AsyncExporter[E],
state_manager: AsyncStateManager,
*,
max_concurrency: int | None = None,
sleep: Callable[[float], Awaitable[None]] = sleep,
closeables: Iterable[SupportsAclose] = (),
memory_ingestor: MemoryIngestor | None = None,
shutdown: ShutdownSignal | None = None,
on_event: Callable[[PipelineEvent], None] | None = None,
usage_sources: Iterable[UsageReporter] = (),
clock: Callable[[], float] = monotonic,
) -> AsyncETLPipeline[E]
Build a pipeline whose max_concurrency comes from the pipeline config section.
The other arguments are the constructor's; max_concurrency
overrides the section's value when given. Pass the section's
:meth:~sci_etl_core.config.PipelineConfig.run_arguments to
:meth:run.
async
¶
run(
query: str,
*,
page_size: int = _DEFAULT_PAGE_SIZE,
sleep_between: float = 0.0,
total_limit: int | None = None,
start_index: int | None = None,
newest_first: bool = False,
max_attempts: int | None = 3,
) -> int
Process listing pages until the ceiling is reached or the listing ends.
Every argument after query is keyword-only. total_limit
defaults to page_size.
Paging resumes from the cursor saved by the state manager unless
start_index is given. Pass 0 to rescan a listing from its first
page: processed records are skipped by id, so a rescan costs listing
requests but never reprocesses a record. A start_index above 0
needs an extractor that pages by offset
(:class:~sci_etl_core.extractors.async_base.OffsetListing).
Pass newest_first=True for a listing that puts new submissions
first, such as arXiv's; it too needs an OffsetListing extractor.
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 cursor.
The saved cursor only moves past pages whose every record was settled.
Once a record fails, or is deferred because total_limit was reached,
the cursor 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. 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 or
ends the listing.
A page with no next cursor, or with no entries, ends the listing, and
the run completes. An OffsetListing extractor then saves the offset
past the last entry, so entries appended later are found by the next
run; any other extractor saves no cursor, so the next run starts from
the first page. A page the source marks truncated, because it
stopped at its own result cap, completes the run too, reports the cap
in :attr:~sci_etl_core.observability.RunMetrics.listing_truncated and
one log line, and saves no cursor, so the next run pages the reachable
results again instead of stopping at the cap.
:attr:~sci_etl_core.models.PipelineMetadata.truncated records it
until a run reaches the end of the listing without a cap. When the
source rejects the cursor with
:class:~sci_etl_core.exceptions.StaleCursorError, the run restarts
the listing from its first page, once per run.
With max_attempts set, a record's failed attempts are counted
through the state manager's record_failure, and a listed record
whose attempts reached max_attempts in earlier runs is skipped as
quarantined: it is not processed, counts as settled, and is counted in
:attr:~sci_etl_core.observability.RunMetrics.quarantined. Failures
are counted only from pages that processed a record, or from stalled
pages that a later page of the run cleared, so an outage that fails
every record never quarantines one. max_attempts=None counts
nothing.
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 cursor does not move past the page that
was cut short, and :class:PipelineInterrupted is raised.
The exporter is opened before the first listing request, and a fault
there aborts the run before any request. Each processed record,
including one with no entities, is written to the exporter. A record
is marked processed only once its entities are durable: right after
write on an exporter with durable_writes = True, and after the
page's flush otherwise. A write fault fails the record and
counts one attempt. A flush fault leaves the records it would have
covered unsettled, counts no attempt against them, and makes the page a
stalled page.
However a run ends, the exporter is flushed and closed, and then state
is flushed through the state manager's flush. A fault in either
after the run itself failed is logged, so it never hides the original
error.
A page made up entirely of processed or quarantined 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 the end of the listing.
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
|
The exporter could not open, a listing page could not be fetched or parsed, the source rejected the cursor again after a restart, or records kept failing with none processed: on a second page before any progress, or on the last page of the listing. 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 |
ETLPipeline(
extractor: AsyncExtractor,
relevance_filter: AsyncRelevanceFilter,
entity_extractor: AsyncEntityExtractor[Any],
exporter: AsyncExporter[Any],
state_manager: AsyncStateManager,
*,
max_concurrency: int = 6,
sleep: Callable[[float], Awaitable[None]] = sleep,
closeables: Iterable[SupportsAclose] = (),
memory_ingestor: MemoryIngestor | None = None,
shutdown: ShutdownSignal | None = None,
on_event: Callable[[PipelineEvent], None] | None = None,
usage_sources: Iterable[UsageReporter] = (),
clock: Callable[[], float] = monotonic,
run_timeout: float = inf,
)
Synchronous facade for :class:AsyncETLPipeline.
The only blocking entrypoint. It takes the arguments of
:class:AsyncETLPipeline, and all collaborators injected into it 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,
extractor: AsyncExtractor,
relevance_filter: AsyncRelevanceFilter,
entity_extractor: AsyncEntityExtractor[Any],
exporter: AsyncExporter[Any],
state_manager: AsyncStateManager,
*,
max_concurrency: int | None = None,
sleep: Callable[[float], Awaitable[None]] = sleep,
closeables: Iterable[SupportsAclose] = (),
memory_ingestor: MemoryIngestor | None = None,
shutdown: ShutdownSignal | None = None,
on_event: Callable[[PipelineEvent], None] | None = None,
usage_sources: Iterable[UsageReporter] = (),
clock: Callable[[], float] = monotonic,
run_timeout: float = inf,
) -> ETLPipeline
Build a pipeline configured as :meth:AsyncETLPipeline.from_config describes.
run(
query: str,
*,
page_size: int = 100,
sleep_between: float = 0.0,
total_limit: int | None = None,
start_index: int | None = None,
newest_first: bool = False,
max_attempts: int | None = 3,
) -> int
Run :meth:AsyncETLPipeline.run with the same arguments 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)
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>, at WARNING on the sci_etl_core.ingest_async logger. 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
|
|
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,
)
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,
quarantined: int = 0,
listing_truncated: bool = False,
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. quarantined
counts listed records skipped because they failed max_attempts times,
and listing_truncated is True when the source stopped the listing
at its result cap.
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
¶
RunStarted(
*,
query: str,
start_index: int | None,
total_limit: int,
newest_first: bool,
cursor: str | None = None,
)
A run began.
cursor is the listing cursor of its first request, None for the
first page. start_index is the same position as a listing offset when
the extractor pages by offset, and None otherwise.
dataclass
¶
PageFetched(
*,
offset: int | None,
entries: int,
new_records: int,
cursor: str | None = None,
truncated: bool = False,
)
A listing page arrived with entries entries, new_records of them not yet processed.
cursor is the cursor the page was requested with, None for the
first page, and offset its listing offset when the extractor pages by
offset, or None. truncated is True when the source stopped the
listing at its result cap on this page.
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 | None,
duration_seconds: float,
metrics: RunMetrics = RunMetrics(),
cursor: str | None = None,
)
Every record of a page finished; metrics is the run so far.
cursor and offset locate the page as in :class:PageFetched.
class-attribute
instance-attribute
¶
metrics: RunMetrics = field(default_factory=RunMetrics)
dataclass
¶
RunFinished(*, metrics: RunMetrics)
The run ended, however it ended; metrics.outcome says how.