pipeline

Pipeline framework for data processing workflows.

pipeline.cache

Pipeline caching system using Parquet for fast checkpointing.

Provides hash-based cache invalidation and parquet storage for pipeline step outputs, enabling fast debugging and iteration.

PipelineCache

Manages parquet-based caching for pipeline steps.

Cache structure

.cache/ {step_name}/ {cache_key}/ metadata.json {table_name}.parquet ...

The cache key is a hash of: - Step name - Input data (schema + row count + content hash of all rows) - Step parameters

This ensures cache invalidation when inputs or configuration change.

cache_dir `instance-attribute`

cache_dir = Path(cache_dir)

get_cache_key

get_cache_key(
    step_name: str,
    inputs: dict[str, pl.DataFrame] | None,
    params: dict[str, Any] | None,
) -> str

Generate cache key from step name, inputs, and parameters.

The key is a hash of: - Step name - For each input DataFrame: - Table name - Schema (column names and types) - Row count - Content hash (hash of all row values) - Step parameters (as sorted JSON)

This ensures the cache invalidates when any input data or configuration changes.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Name of the pipeline step	required
`inputs`	`dict[str, pl.DataFrame] \| None`	Input DataFrames (or None for first step)	required
`params`	`dict[str, Any] \| None`	Step parameters from config	required

Returns:

Type	Description
`str`	16-character hex hash string

load

load(
    step_name: str, cache_key: str
) -> dict[str, pl.DataFrame] | None

Load cached step outputs from parquet.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Name of the pipeline step	required
`cache_key`	`str`	Cache key from get_cache_key()	required

Returns:

Type	Description
`dict[str, pl.DataFrame] \| None`	Dictionary of table name -> DataFrame, or None if cache miss

save

save(
    step_name: str, cache_key: str, outputs: dict[str, pl.DataFrame]
) -> None

Save step outputs to parquet cache.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Name of the pipeline step	required
`cache_key`	`str`	Cache key from get_cache_key()	required
`outputs`	`dict[str, pl.DataFrame]`	Dictionary of table name -> DataFrame	required

clear

clear() -> None

Clear the entire cache.

invalidate

invalidate(step_name: str | None = None) -> None

Invalidate cache for a step or all steps.

Parameters:

Name	Type	Description	Default
`step_name`	`str \| None`	Name of step to invalidate, or None for all steps	`None`

list_cached_steps

list_cached_steps() -> list[dict[str, Any]]

List all cached steps with metadata.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts with step info (name, cache_key, tables, sizes)

get_stats

get_stats() -> dict[str, int | float]

Get cache statistics.

Returns:

Type	Description
`dict[str, int \| float]`	Dict with 'loaded', 'missing', 'stale', 'total',
`dict[str, int \| float]`	and 'load_rate' keys

reset_stats

reset_stats() -> None

Reset cache statistics.

pipeline.config

Reading a pipeline config: {{ name }} templates, and what they resolve to.

A config names its data locations through templates over top-level variables, the committed ENVS defaults and the environment. These helpers resolve them and check, before any step runs, that every root a step needs is set and every input path it names exists.

MAX_TEMPLATE_PASSES `module-attribute`

MAX_TEMPLATE_PASSES = 10

ENVS_KEY `module-attribute`

ENVS_KEY = 'ENVS'

resolve_templates

resolve_templates(config: dict[str, Any]) -> dict[str, Any]

Replace every {{ name }} in a loaded config with its value.

The values come from the config's own variables and the environment, as template_namespace gathers them. A root no one set stays a template.

template_namespace

template_namespace(config: dict[str, Any]) -> dict[str, str]

Everything a {{ name }} may refer to, lowest precedence first.

Three sources feed one namespace, which is what keeps this from being a second mechanism bolted beside the templating that already exists:

the config's own top-level strings -- output_dir and friends, so log_file: "{{ output_dir }}/run.log" keeps working;
committed defaults under ENVS, for a root that genuinely is the same everywhere;
the environment, including a .env the runners load, which is where a machine says where it mounts a shared location.

The environment wins, and that is not an override of anything stated as fact: a config says {{ MTC_DATA }}/Data/..., which is true wherever MTC_DATA points. Only the root is machine-specific, and only the root comes from outside. The structure beneath it is shared knowledge and stays in the config where it can be reviewed.

Parameters:

Name	Type	Description	Default
`config`	`dict[str, Any]`	The raw configuration, before substitution.	required

Returns:

Type	Description
`dict[str, str]`	Mapping of template name to value.

check_roots_resolved

check_roots_resolved(
    config: dict[str, Any], namespace: dict[str, str]
) -> None

Fail if a referenced root is declared nowhere, naming it.

Left unresolved, {{ MTC_DATA }} renders literally and surfaces later as a baffling "no such file or directory" containing brace characters. The root is missing, and saying so is more useful than anything downstream can work out.

Raises:

Type	Description
`ValueError`	Naming every unresolved root and where to set it.

shorthands_used_by_steps

shorthands_used_by_steps(raw_config: dict[str, Any]) -> set[str]

Which shorthands the configured steps actually refer to.

Read off the unsubstituted steps, so a shorthand is only in scope when a step names it as {{ key }}. A commented-out parameter is not a reference: YAML drops comments before this sees them, which is what makes the distinction free. pums_dir in bats_2023 is exactly that -- declared, and used only by two commented lines -- so requiring it to exist would refuse to start a run that never reads it.

check_input_paths

check_input_paths(config: dict[str, Any], used: set[str]) -> None

Fail before the first step if an input directory a step needs is missing.

Every one of these is read by some step, but never the first one, so a wrong drive letter surfaces minutes into a run rather than at the start -- and one path at a time: fix, re-run, wait, discover the next. A wrong TM2_shapefile_dir was found this way, eight steps in, after the survey had been read, trips linked, tours built and weights computed.

Only inputs a configured step refers to are checked. Output directories are created on demand, and a path nothing reads is not this function's business.

Parameters:

Name	Type	Description	Default
`config`	`dict[str, Any]`	The fully substituted configuration.	required
`used`	`set[str]`	Shorthands the steps refer to, from :func:`shorthands_used_by_steps`.	required

Raises:

Type	Description
`ValueError`	Naming every missing path at once, and how to redirect them.

pipeline.decoration

Decorators for pipeline steps with automatic validation and caching.

PIPELINE_ARGS `module-attribute`

PIPELINE_ARGS = frozenset(
    {
        "canonical_data",
        "validate_input",
        "validate_output",
        "cache",
        "pipeline_cache",
        "kwargs",
    }
)

settings_models

settings_models(
    func: Callable[..., Any],
) -> dict[str, type[BaseModel]]

The step's arguments annotated with a pydantic model, by argument name.

field_owners

field_owners(func: Callable[..., Any]) -> dict[str, str]

Each settings-model field name, mapped to the argument whose model holds it.

A flat config key is routed by this map, so every name must have one owner.

Raises:

Type	Description
`TypeError`	If a field name is also another model's field, a step argument, a canonical table or a pipeline argument.

step

step(
    *,
    validate_input: bool = True,
    validate_output: bool = False,
    cache: bool = False,
    requires: dict[str, set[str]] | None = None,
    produces: dict[str, set[str]] | None = None
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator for pipeline steps with automatic validation and caching.

This decorator validates canonical data inputs and/or outputs using the Pydantic models defined in data.models. Only parameters/returns that match canonical table names (households, persons, days, unlinked_trips, linked_trips, tours) are validated.

Steps can declare workflow contracts via requires and produces:

requires: mapping of table name to set of field names that must be present before the step executes.
produces: mapping of table name to set of field names that the step is responsible for after execution (newly created or updated).

When caching is enabled, the decorator will check for cached outputs before executing the step function. If valid cache exists, outputs are loaded from parquet files. Otherwise, the step executes and results are cached after successful validation.

The default value is to only validate inputs to avoid duplicate validation. Recommend putting a final_check step at the end of the pipeline to validate all tables after all processing is complete.

Parameters:

Name	Type	Description	Default
`validate_input`	`bool`	Whether to validate inputs. Defaults to True.	`True`
`validate_output`	`bool`	Whether to validate outputs. Defaults to False.	`False`
`cache`	`bool`	Whether to enable caching for this step. Defaults to False.	`False`
`requires`	`dict[str, set[str]] \| None`	Mapping of table name to set of required input field names.	`None`
`produces`	`dict[str, set[str]] \| None`	Mapping of table name to set of produced output field names.	`None`

Example

@step( ... requires={"unlinked_trips": {"day_id", "depart_time"}}, ... produces={"linked_trips": {"linked_trip_id"}}, ... ) ... def link_trips( ... unlinked_trips: pl.DataFrame, ... max_dwell_time: float, # a setting, from the step's params ... ) -> dict[str, pl.DataFrame]: ... # Process trips ... linked_trips = ... ... return {"linked_trips": linked_trips}

@step(validate_input=False) ... def load_data(input_paths: dict) -> dict[str, pl.DataFrame]: ... return { ... "households": households_df, ... "persons": persons_df, ... ... ... }

Returns:

Type	Description
`Callable[[Callable[..., Any]], Callable[..., Any]]`	Decorated function with validation

pipeline.logger

Logging configuration utilities for the pipeline.

setup_logging

setup_logging(
    log_file: Path | str | None,
    console_level: int = logging.INFO,
    file_level: int = logging.DEBUG,
    log_file_mode: str = "a",
) -> logging.Logger

Configure logging to both console and file.

Parameters:

Name	Type	Description	Default
`log_file`	`Path \| str \| None`	Path to the log file	required
`console_level`	`int`	Logging level for console output	`logging.INFO`
`file_level`	`int`	Logging level for file output	`logging.DEBUG`
`log_file_mode`	`str`	File open mode for the log file. Use "a" to append (default) or "w" to overwrite at the start of each run.	`'a'`

Returns:

Type	Description
`logging.Logger`	Configured logger instance

pipeline.pipeline

Pipeline execution module for running data processing steps.

Pipeline

Class to run a data processing pipeline based on a configuration file.

cache `instance-attribute`

cache: PipelineCache | None

config_path `instance-attribute`

config_path = config_path

config `instance-attribute`

config = self._load_config()

data `instance-attribute`

data: CanonicalData = CanonicalData()

steps `instance-attribute`

steps: dict[str, Callable] = {
    (func.__name__): func for func in (steps or [])
}

report_status

report_status() -> None

Report the current pipeline status with ASCII flow diagram.

Shows which steps have cached data available: - ✓ CACHED: Step has valid cache - ✗ NO CACHE: Step caching enabled but no cache exists - ∅ NO CACHE (disabled): Step caching disabled

parse_step_args

parse_step_args(
    step_name: str, step_obj: Callable
) -> dict[str, Any]

Separate the canonical data and parameters.

If argument name matches a canonical table, it is passed from self.data. Else, it is taken from the step configuration "parameters".

A config key the step does not take raises, as does one naming a canonical table: either would otherwise be dropped without a word, and the step would run on its own default instead of the configured value.

An argument annotated with a pydantic model is built from its fields' keys. They sit flat under params:, each routed to the model that owns the name, or in a block named after the argument; not both. The model decides what may be left out: a field with a default may, one without may not, the same rule as a plain argument.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Name of the step.	required
`step_obj`	`Callable`	The step function or class.	required

run

run() -> CanonicalData

Run a data processing pipeline based on a configuration file.

get_data

get_data(table_name: str, step: str | None = None) -> Any

Fetch a table from cached pipeline data. If no cache, just return latest.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Name of the table to fetch (e.g., 'households', 'trips')	required
`step`	`str \| None`	Optional step name to fetch from. If None, uses the last step that has a cache containing this table.	`None`

Returns:

Type	Description
`Any`	The requested DataFrame or data object

Raises:

Type	Description
`ValueError`	If table not found in any cached steps, or if specified step doesn't have cache or doesn't contain table.

Example

pipeline = Pipeline(config_path, steps, caching=True)

Fetch from latest cached step

households = pipeline.get_data("households")

Fetch from specific step

trips = pipeline.get_data("linked_trips", step="link_trips")

pipeline.step_registry

Central registry for step workflow contracts.

Steps declare which fields they require as input and produce as output via the @step() decorator. This registry stores those contracts and provides lookup functions used by validation and documentation.

Models define what a field is (type, constraints, relationships). Steps define when a field must exist (workflow contracts). This registry bridges the two.

STEP_REGISTRY `module-attribute`

STEP_REGISTRY: dict[str, StepContract] = {}

TableContract `dataclass`

Workflow contract for a single table within a step.

requires `class-attribute` `instance-attribute`

requires: frozenset[str] = field(default_factory=frozenset)

produces `class-attribute` `instance-attribute`

produces: frozenset[str] = field(default_factory=frozenset)

StepContract `dataclass`

Workflow contract for a pipeline step.

tables `class-attribute` `instance-attribute`

tables: dict[str, TableContract] = field(default_factory=dict)

is_config_named

is_config_named(field_name: str) -> bool

Whether a contract entry is a pattern like usable_<profile>, not a column.

The pattern documents the column; the real name comes from the step's config, so the step itself checks for it, not the input validation.

register_step

register_step(
    step_name: str,
    requires: dict[str, set[str]] | None = None,
    produces: dict[str, set[str]] | None = None,
) -> None

Register a step's workflow contract in the global registry.

Called automatically by the @step() decorator at import time.

Parameters:

Name	Type	Description	Default
`step_name`	`str`	Name of the step (typically the function name).	required
`requires`	`dict[str, set[str]] \| None`	Mapping of table name to set of required input field names.	`None`
`produces`	`dict[str, set[str]] \| None`	Mapping of table name to set of produced output field names.	`None`

get_required_fields

get_required_fields(table_name: str, step_name: str) -> set[str]

Get fields required by a step for a given table.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Name of the canonical table.	required
`step_name`	`str`	Name of the pipeline step.	required

Returns:

Type	Description
`set[str]`	Set of field names required as input. Empty if step or table
`set[str]`	has no registered contract.

get_produced_fields

get_produced_fields(table_name: str, step_name: str) -> set[str]

Get fields produced by a step for a given table.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Name of the canonical table.	required
`step_name`	`str`	Name of the pipeline step.	required

Returns:

Type	Description
`set[str]`	Set of field names produced as output. Empty if step or table
`set[str]`	has no registered contract.

get_all_required_fields

get_all_required_fields(table_name: str) -> set[str]

Get the union of all required fields for a table across every step.

Useful for checking that raw input data contains every field that any downstream step will need.

Parameters:

Name	Type	Description	Default
`table_name`	`str`	Name of the canonical table.	required

Returns:

Type	Description
`set[str]`	Set of field names required by at least one registered step.

get_step_validation_summary

get_step_validation_summary() -> dict[str, dict[str, list[str]]]

Get a summary of required fields across all steps and tables.

Returns:

Type	Description
`dict[str, dict[str, list[str]]]`	Nested dict: step_name -> table_name -> sorted list of required fields.

pipeline

pipeline.cache

PipelineCache

cache_dir instance-attribute

get_cache_key

load

save

clear

invalidate

list_cached_steps

get_stats

reset_stats

pipeline.config

MAX_TEMPLATE_PASSES module-attribute

ENVS_KEY module-attribute

resolve_templates

template_namespace

check_roots_resolved

shorthands_used_by_steps

check_input_paths

pipeline.decoration

PIPELINE_ARGS module-attribute

settings_models

field_owners

step

pipeline.logger

setup_logging

pipeline.pipeline

Pipeline

cache instance-attribute

config_path instance-attribute

config instance-attribute

data instance-attribute

steps instance-attribute

report_status

parse_step_args

run

get_data

Fetch from latest cached step

Fetch from specific step

pipeline.step_registry

STEP_REGISTRY module-attribute

TableContract dataclass

requires class-attribute instance-attribute

produces class-attribute instance-attribute

StepContract dataclass

tables class-attribute instance-attribute

is_config_named

register_step

get_required_fields

get_produced_fields

get_all_required_fields

get_step_validation_summary

cache_dir `instance-attribute`

MAX_TEMPLATE_PASSES `module-attribute`

ENVS_KEY `module-attribute`

PIPELINE_ARGS `module-attribute`

cache `instance-attribute`

config_path `instance-attribute`

config `instance-attribute`

data `instance-attribute`

steps `instance-attribute`

STEP_REGISTRY `module-attribute`

TableContract `dataclass`

requires `class-attribute` `instance-attribute`

produces `class-attribute` `instance-attribute`

StepContract `dataclass`

tables `class-attribute` `instance-attribute`