Step reference

All steps live in lcmd_db.core.lib.importers.steps.

Fetch

Parse

Derive

Load

Multi-CSV joins

read_csvs_merged handles the common case where a dataset spans several CSVs keyed by a shared identifier. Each CSVSpec controls one input.

from lcmd_db.core.lib.importers.steps import CSVSpec, read_csvs_merged

parse = read_csvs_merged(
    specs=[
        CSVSpec(path="main.csv"),
        CSVSpec(
            path="extra.csv",
            columns=["id", "extra_col"],            # avoid pulling junk columns
            rename={"extra_col": "renamed_col"},    # resolve name collisions
        ),
        CSVSpec(
            path="srs.csv",
            marker="_in_srs",                        # boolean column flagging this CSV's rows
        ),
    ],
    on="id",
    how="left",     # outer is uncommon; left preserves the main CSV's rows
)

The marker column is the canonical way to drive get_participants_fn in reactions — it's purpose-built and avoids null-checking substantive columns.

The ImportContext

Every step receives a single ImportContext. The fields you'll touch:

Writing a custom step

The pipeline is just composed Step[InState, OutState] values. When the built-ins don't fit, write your own:

import polars as pl

from lcmd_db.core.lib.importers import HasRows, ImportContext, ParseError, Step

def select_unique(column: str) -> Step[HasRows, HasRows]:
    """Replace ctx.rows with the unique values of one column, renamed to 'smiles'."""
    async def _step(ctx: ImportContext) -> ImportContext:
        assert ctx.rows is not None
        if column not in ctx.rows.columns:
            raise ParseError(f"Column {column!r} not found")
        ctx.rows = (
            ctx.rows.select(pl.col(column))
            .drop_nulls()
            .unique()
            .rename({column: "smiles"})
        )
        if ctx.limit is not None:
            ctx.rows = ctx.rows.head(ctx.limit)
        return ctx

    return Step(_step)

A few rules the type system enforces:

• Step[Empty, …] runs first (no upstream state), Step[HasRows, …] requires ctx.rows to exist
• >> only composes when the right-hand input matches the left-hand output
• An async step that hits the disk should defer with asyncio.to_thread — the runner overlaps I/O across pipelines

Error modes

By default, per-row failures are collected — bad rows are dropped, the rest ship. Switch to fail-fast with on_error="raise":

my_subset = Subset(
    name="MySubset",
    ...,
    on_error="raise",   # default: "collect"
)

Use raise when:

• You want CI to fail loudly on a bad row
• The dataset is small enough that any drop is significant
• You're iterating on the pipeline and want the first failure as the signal

Use collect (the default) when:

• The dataset is large and partial success is useful
• A few rows have known issues but you want the rest

Either way, every error is logged with row index and stage. The ProcessingResult returned by import_subset carries the activity log; the command also prints a per-stage summary at the end.

Grouping subsets into a dataset

Several subsets that share citations and metadata can be grouped into a dataset. A dataset is a declarative Dataset dataclass instance — the same functional shape as a Subset. It lists its members by referencing their Subset specs directly, so there is no database query and the grouping is explicit. name, description, and subsets are required; citations, keywords, and metadata are optional (metadata defaults to {"source": "dataset_registry"}):

from lcmd_db.apps.datasets.lib.citations.registry import Author, Citation
from lcmd_db.core.lib.datasets import Dataset
from lcmd_db.registry.subsets.my_subset import my_subset

my_dataset = Dataset(
    name="MyDataset",
    description="A short paragraph about the dataset.",
    subsets=(my_subset,),
    citations=[
        Citation(
            citation_key="smith2024example",
            type="article-journal",
            title="The paper title",
            authors=[Author(family="Smith", given="Jane")],
            year=2024,
            doi="10.1234/example",
        ),
    ],
    keywords=["example", "tutorial"],
    metadata={"is_default": True},
)

Then register it (this is a separate registry from the subset one):

from lcmd_db.core.lib.datasets import dataset_registry as registry

from .my_dataset import my_dataset

registry.register("my_dataset", my_dataset)

The subsets tuple holds the subset specs themselves, so a typo'd import fails at startup rather than silently dropping a member, and the declared order becomes the subset order within the dataset. Datasets are materialized by just manage create_dataset --all. Reference example: apps/backend/lcmd_db/registry/datasets/oscar.py.