Contributing to sci-etl-core¶

Thanks for your interest in improving sci-etl-core! New extractors, parsers, exporters, embedding backends, and documentation fixes are all genuinely welcome — first-time contributors included. This guide gets you productive quickly.

By participating you agree to abide by our Code of Conduct.

Development Setup¶

git clone https://github.com/xueromll/sci-etl-core.git
cd sci-etl-core

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate

pip install -e ".[full,dev]"

Python 3.11+ is required; the codebase uses X | Y unions and slots=True, kw_only=True dataclasses.
Why [full]: the suite exercises every bundled component.
Not needed for tests: sentence-transformers and a real SQL driver are mocked or stubbed.

How the Codebase Works — Read This First¶

Async is the only implementation. Every component is async. Don't add synchronous twins of components; there is no code generator. Blocking implementations reach a pipeline through the adapters in _adapters.py.
One blocking entrypoint. ETLPipeline (pipeline.py) runs AsyncETLPipeline through _sync_bridge.run_sync. Discuss new blocking APIs in an issue first.
Keep the event loop free. Run CPU-bound or blocking work with asyncio.to_thread. Parsers, pandas, sqlite3, and file IO all follow this pattern.
Inject collaborators. Clients, parsers, sleep, and logger are constructor arguments so tests can substitute them. Accept logger: Callable[[str], None] | None and an injectable sleep wherever timing or retries are involved.
Never swallow cancellation. Before catching broad exceptions, catch and re-raise asyncio.CancelledError (see llm/relevance_async.py).
Signal failures with exceptions. Use the hierarchy in exceptions.py. Don't return None or an empty result for a transport failure — the pipeline relies on UpstreamError and MalformedResponseError to tell faults apart from the end of the data.
Be safe under concurrency. The pipeline calls relevance filters, entity extractors, an exporter's write, and mark_processed concurrently. Serialize shared writes with an asyncio.Lock, and write files with _atomic_io.atomic_write_text. AsyncCsvExporter and AsyncFileStateManager are good models.
Log through the module logger. Each module logs with _logger = logging.getLogger(__name__), at WARNING for anything that changes what a run produces or costs, ERROR for a sink or store fault the run reports, and INFO for routine notes. Never configure handlers or levels; that is the application's job.
Keep optional imports lazy. Each package __init__.py maps public names to their modules in _EXPORTS and loads them on first access, so importing one component never requires another's optional dependencies. Register new public names there and in the matching TYPE_CHECKING imports (a test checks that they agree), and add any new third-party dependency to an extra in pyproject.toml.
No domain-specific constants in the core; keep it field-agnostic.

Running Tests¶

The suite is offline — no network, live LLM, or embedding service required.

pytest                                                # everything offline
pytest tests/async                                    # async components
pytest tests/contract                                 # ABC conformance
pytest --cov=sci_etl_core --cov-report=term-missing   # coverage report

Coverage stays at 100%. pytest --cov=sci_etl_core fails below 100% (configured in pyproject.toml), so new and changed code needs tests that cover it. Use # pragma: no cover only for lines that can't run on the test platform, such as OS-specific imports. CI also measures branch coverage and reports it in the job summary; it isn't gated yet.
Async tests use an explicit @pytest.mark.asyncio marker (auto mode isn't configured), with AsyncMock or the mocker fixture.
No real network or backoff waits. Mock HTTP, LLM, and embedding clients, and inject sleep=AsyncMock() to skip backoff delays.
Contract tests. Every public ABC implementation has a case in tests/contract/test_abc_conformance.py; add one for each new implementation.
Property tests. Put invariants — especially ones two backends must share, such as the in-memory and SQLite embedding stores — in Hypothesis tests in tests/test_properties.py.
Public-surface snapshot. tests/api/public_surface.txt records every stable class, field, method, and function signature, and tests/api/test_public_surface.py fails when the code no longer matches it. When you change the public API on purpose, regenerate the snapshot, commit the diff, and add a CHANGELOG.md entry for it:

python tests/api/update_surface.py

tests/api/surface.py lists the stable names. Every name that sci-etl-cli or udg-catalogue imports or subclasses must be among them. tests/api/consumer_surface.txt lists those names. Regenerate it when a consumer's pin changes, reading the catalogue from its fetched remote branch rather than a local checkout that may be behind:

git -C ../udg-catalogue fetch origin
python tests/api/scan_consumers.py sci-etl-cli=../sci-etl-cli udg-catalogue=../udg-catalogue@origin/main

- Live tests. tests/live runs one small query against each bundled source and checks the records, metadata, and full text it returns. The tests are deselected by default. The Live workflow runs them nightly and is never a required check. Run them locally with pytest -m live tests/live. Keys for a higher rate limit are optional and read from NCBI_API_KEY, SEMANTIC_SCHOLAR_API_KEY, and OPENALEX_MAILTO. Without its key, a source that keeps answering 429 is skipped instead of failed. - Throughput benchmark. python benchmarks/run_throughput.py runs every bundled pipeline exporter through AsyncETLPipeline at 1,000, 10,000, and 50,000 records and writes the timings to benchmarks/results/<version>.json. An exporter counts as linear when its time per record at the largest size is at most 1.5 times its time at the smallest. --sizes, --repeats, and --exporters narrow a run. - Downstream tests. The Downstream workflow runs the sci-etl-cli and udg-catalogue suites against every push and pull request, with this checkout installed in place of the release each project pins, and fails on any DeprecationWarning. Each job installs the extras the project's own requirement names. To run the CLI suite locally:

git clone https://github.com/xueromll/sci-etl-cli.git ../sci-etl-cli
pip install -e ".[config,async,arxiv,html,llm,pdf]"
pip install --no-deps -e ../sci-etl-cli
pip install click rich pytest pytest-asyncio pytest-mock pytest-cov
cd ../sci-etl-cli && python -m pytest -W error::DeprecationWarning

And the udg-catalogue suite, which also installs the CPU build of PyTorch:

git clone https://github.com/xueromll/udg-catalogue.git ../udg-catalogue
export PIP_EXTRA_INDEX_URL=https://download.pytorch.org/whl/cpu
pip install -e ".[config,async,arxiv,html,llm,pdf,processors,cluster,embeddings,embeddings-local,search]" \
    -r ../udg-catalogue/requirements/app.txt -r ../udg-catalogue/requirements/dev.txt
cd ../udg-catalogue && python -m pytest -W error::DeprecationWarning

Code Style¶

PEP 8 naming: snake_case functions and variables, CapWords classes, UPPER_CASE constants.
Type hints on every public signature. The package ships py.typed; keep it accurate.
Small, single-responsibility functions. Prefer composition and dependency injection over branching on hard-coded logic.
Docstrings for contracts and intent. Document Raises: and non-obvious decisions; don't write comments that restate the code.
American English for identifiers and docstrings.

ruff and mypy run in CI on every push and pull request, configured in pyproject.toml. ruff checks the ASYNC, UP, RUF, and PT rule sets besides the pycodestyle, Pyflakes, isort, and bugbear rules. mypy checks the core contracts (models, exceptions, observability, pipeline_async, and every async_base module) with stricter flags: no untyped definitions, no bare generics, no returned Any, and no implicit re-exports. Run both before opening a PR:

pip install -e ".[full,dev,lint]"
ruff check .
mypy

ruff check --fix . sorts imports and applies the other safe fixes. Formatting isn't enforced, so keep unrelated reformatting out of your PR. Record user-facing changes under the unreleased version in CHANGELOG.md.

Documentation¶

The documentation site is built with MkDocs and Material for MkDocs from mkdocs.yml and the docs/ folder, and published to GitHub Pages.

pip install -e ".[docs]"
git clone https://github.com/xueromll/sci-etl-cli.git ../sci-etl-cli
pip install --no-deps -e ../sci-etl-cli
mkdocs serve                                          # preview at http://127.0.0.1:8000
mkdocs build --strict                                 # the check CI runs

Where pages live. Guides are Markdown files under docs/, listed in the nav of mkdocs.yml. CHANGELOG.md, MIGRATION.md, ROADMAP.md, this guide, SECURITY.md, and CODE_OF_CONDUCT.md stay at the repository root; the pages under docs/project/ render them, and links between them are rewritten for the site.
The CLI section comes from the docs/ folder and nav of the sci-etl-cli repository. The build looks for a checkout beside this one, or at the path in SCI_ETL_CLI_DIR; without one it leaves the section out, which --strict reports as a failure. Change CLI pages in that repository.
API reference. The pages under docs/reference/ are generated from docstrings. A new public module needs a module.path entry on one of them; tests/test_docs.py fails until it has one.
Code examples. tests/test_docs.py also checks that every Python example under docs/ compiles and that every name it imports from sci_etl_core exists, so renaming a public name means updating the examples too.
Publishing. The docs workflow deploys master as the dev version and each v* tag as its minor version, for example 0.3, with the latest alias. A tag build uses the CLI pages from the latest sci-etl-cli release, or from its default branch when that release has no mkdocs.yml. Nothing needs to be published by hand.

Releasing¶

Development versions. Right after a release, master moves to the next development version, such as 0.7.0.dev0 after 0.6.0, so a build from master never claims to be a release.
Release gate. Pushing a v* tag runs the CI suite and the Downstream workflow on the tagged commit. The distribution is built and published only when both pass, so a release that breaks sci-etl-cli or udg-catalogue is never published. The tag must match the version in pyproject.toml.
Deprecations. From 0.6.0 on, a deprecated name keeps working for at least two minor releases before it is removed.
Before tagging. Date the unreleased section of CHANGELOG.md, update the supported versions in SECURITY.md, and rerun python benchmarks/run_throughput.py so benchmarks/results/ holds the new version.
Pinned actions. Workflows pin every action to a commit SHA, with the tag in a trailing comment such as # v5. To move to a newer release, resolve its SHA with git ls-remote https://github.com/actions/checkout refs/tags/v5 and update the pin and the comment together.

Commit Format¶

Use Conventional Commits:

<type>(<scope>): <short summary>

Common types: feat, fix, docs, test, refactor, perf, chore. Mark breaking changes with ! (e.g. refactor(core)!: ...).

Examples:

feat(extractors): add PubMed async extractor
fix(exporters): serialize concurrent CSV upserts
docs(readme): document ETLPipeline event-loop behavior

Keep the summary imperative and under ~72 characters. Reference issues in the body (Closes #123).

Pull Request Process¶

Open an issue first for anything non-trivial, so we can agree on the approach.
Branch from master, e.g. feat/pubmed-extractor.
Write tests alongside your change; keep coverage at 100%.
Run the test suite locally (plus lint and type checks if you use them).
Update the docs for user-facing changes:
the pages under docs/ for usage; keep README.md a short overview
MIGRATION.md for a breaking change, under the release that ships it
docs/guide/migrating-a-pipeline.md when the change affects moving an existing pipeline onto the library
ROADMAP.md when you ship a listed item
the pull request description for any breaking change, with what users need to update
Fill in the pull request template.
Keep PRs focused — one logical change per PR is easiest to review.

A maintainer will review promptly. Expect a friendly, constructive exchange; requested changes are about the code, never about you.

Adding a New Component¶

Most contributions plug into an existing abstract base class:

Component	Subclass	Implement	Contract
Extractor	`AsyncExtractor`	`async fetch_page(query, cursor, page_size) -> ListingPage`, `async fetch_full_text`; `cursor_for_offset` when the cursors are decimal offsets	Return every entry it can read: the pipeline skips processed ids. Count entries without an id in `entries`, return `next_cursor=None` on the last page, and `truncated=True` on the page that reaches the source's own result cap. Raise `UpstreamError` when the source can't be reached after retries, `ExtractionError` when it rejects a request outright, `MalformedResponseError` for an unreadable listing, and `StaleCursorError` for a cursor the source no longer accepts.
Parser	`Parser` (optionally `TableParser`)	`extract_text(content: bytes) -> str`	Synchronous; callers run it in a worker thread. Raise `ParsingError` for bytes it can't read.
LLM client	`AsyncLLMClient`	`async complete_json(system_prompt, user_content, timeout)`; optionally `complete_structured(..., schema, timeout)` and `invalidate(..., schema=None)`	Return the parsed JSON object; raise `LLMError` on failure or when the body isn't a JSON object. Override `complete_structured` when the provider supports JSON-schema output; the default calls `complete_json`.
LLM response cache	`AsyncLLMResponseCache`	`async get(key)`, `async set(key, response)`, `async clear()`	`get` returns `None` for a missing key. Return copies, so a caller changing a response can't change the cache. Raise `LLMCacheError` for a storage fault; `CachingLLMClient` logs it and calls the LLM instead. Add `aclose()` if you hold connections.
Relevance filter	`AsyncRelevanceFilter`	`async is_relevant(record)`	Re-raise `CancelledError`.
Entity extractor	`AsyncEntityExtractor[E]`	`async extract(text) -> Sequence[E]`; `extract_record(record, text)` when the record is needed	The pipeline calls `extract_record`, whose default calls `extract`. Raise on failure instead of returning `[]`, so the record is retried. An extractor that needs the record sets `requires_record = True`; a wrapper delegates to the inner `extract_record`.
Exporter	`AsyncExporter[E]`	`async write(record, entities)`; optionally `open`, `flush`, `aclose`, and `durable_writes`	Take the destination in the constructor. `write` runs for every processed record, including one with no entities, possibly concurrently, and must be idempotent. With `durable_writes = False`, `flush` must make every earlier write durable.
State manager	`AsyncStateManager`	`load_processed_ids`, `mark_processed`, `load_metadata`, `save_metadata`	Concurrency-safe. Override `record_failure` and `failure_counts` to support quarantine; the defaults track nothing. Record a schema version and refuse a file from a newer release. Override `flush()` if you buffer; add `aclose()` if you hold connections.
Embedder	`AsyncEmbedder`	`async embed(texts) -> list[list[float]]`	One vector per input, same order; raise `EmbeddingError`.
Vector store	`AsyncEmbeddingStore`	`add`, `delete_record`, `query`, `count`	Replace chunks with the same `(record_id, chunk_index)`; `delete_record` removes all of a record's chunks. Override `replace_record` (delete then add by default) if your backend can do both atomically. Never return hits with non-finite scores. Match `InMemoryEmbeddingStore` for `top_k`, `min_score`, and `exclude_record_id`. Serialize use of a shared connection, and raise `EmbeddingStoreError`. Implement `iter_records` (passages in `chunk_index` order, records in `record_id` order, no vectors) if the text index should be backfillable from your store.
Chunker	`TextChunker`	`chunk(text) -> list[str]`	Ordered passages covering the text.
Text search store	`AsyncTextSearchStore`	`facet_keys` property, `index`, `delete_record`, `search`, `filter_ids`, `get_documents`, `facet_counts`, `count`	Take parsed queries, never text. Reject a query `search` can't rank with `require_rankable`. Order equal scores by `record_id`, and apply filters before `limit`. Raise `ValueError` before any I/O for a filter or facet key outside `facet_keys` or two filters on one key (`validate_filters`, `validate_facet_keys`). Accept `MetadataFilter` and `RangeFilter`, compare ranges with `tag_in_range`, and fill `TextHit.snippets` for every field with a highlighted match. Override `range_counts` (one `filter_ids` per range by default) if you can count in one read. Match `InMemoryTextSearchStore`'s results, and raise `SearchStoreError`.
Edge source	`AsyncEdgeSource`	`kind` property, `async neighbours(record_ids, limit)`	Map every requested id, even one with no neighbors, to up to `limit` `(record_id, weight)` pairs, best first, where a higher weight means more related. Never close the stores you were given.
Processor	`Processor`	`process(frame) -> DataFrame`	Don't mutate the input frame.
Validator	`RecordValidator`	`is_valid(record) -> bool`; optionally `validate(record) -> ValidationResult`	Operates on one entity dict. Override `validate` to name the field and rule of each rejection; it must reject exactly what `is_valid` rejects.

Register the new class in its subpackage's _EXPORTS map and TYPE_CHECKING imports. A new module also needs an entry on its page under docs/reference/. If it's a primary user-facing class, add it to sci_etl_core/__init__.py the same way. Add a case for it to tests/contract/test_abc_conformance.py.

Happy hacking — and thank you for contributing!