Pipeline Store

The pipeline store provides persistence for versioned pipeline configurations. All operations are keyed by pipeline name and pipeline version. Use it when you need to store and run pipelines by name (e.g. from the REST API or Web UI), or to version pipeline definitions alongside your contracts.

Overview

from pycharter.pipeline_store import PostgresPipelineStore

store = PostgresPipelineStore("postgresql://...")
store.connect()

# Store a pipeline config (steps + optional settings, variables)
config_id = store.store_config(
    pipeline_name="daily_orders",
    pipeline_version="1.0.0",
    steps=[{"type": "extract", "config": {"url": "https://api.example.com/orders"}}, ...],
    settings={"contract_store": {"type": "postgres", "connection_string": "..."}},
    description="Daily orders ingestion",
)
# Retrieve and run with Pipeline.from_dict()
config = store.get_pipeline_dict("daily_orders", "1.0.0")
# config has keys: name, version, steps, settings?, variables?
store.disconnect()

When to use

• REST API / Web UI — The API uses the pipeline store to persist pipeline configs created in the Pipelines section and to run them by name/version.
• Config-driven pipelines — Store pipeline definitions in the database instead of (or in addition to) YAML files.
• Versioning — Keep multiple versions of the same pipeline (e.g. daily_orders 1.0.0 and 2.0.0).

Store implementations

Class	Backend	Use case
InMemoryPipelineStore	Memory	Testing
SQLitePipelineStore	SQLite	Development, single process
PostgresPipelineStore	PostgreSQL	Production, API
MongoDBPipelineStore	MongoDB	Document-oriented
RedisPipelineStore	Redis	Caching / ephemeral

PipelineStoreClient

Base interface for all pipeline stores. All pipeline operations are keyed by (pipeline_name, pipeline_version).

PipelineStoreClient

PipelineStoreClient(connection_string: str | None = None)

Bases: ABC

Client for storing and retrieving pipeline configurations.

All operations are keyed by (pipeline_name, pipeline_version).

connect abstractmethod

connect() -> None

Establish database connection.

disconnect

disconnect() -> None

Close database connection.

store_config abstractmethod

store_config(pipeline_name: str, pipeline_version: str, steps: list[dict[str, Any]], *, settings: dict[str, Any] | None = None, variables: dict[str, Any] | None = None, description: str | None = None, contract_name: str | None = None, contract_version: str | None = None, created_by: str | None = None) -> str

Store a pipeline configuration (immutable per version).

Parameters:

Name	Type	Description	Default
pipeline_name	str	Pipeline identifier.	required
pipeline_version	str	Pipeline version string.	required
steps	list[dict[str, Any]]	Steps list (the DAG definition).	required
settings	dict[str, Any] | None	Optional pipeline settings dict.	None
variables	dict[str, Any] | None	Optional variable substitution dict.	None
description	str | None	Optional human-readable description.	None
contract_name	str | None	Optional associated data contract name.	None
contract_version	str | None	Optional associated data contract version.	None
created_by	str | None	Optional creator identifier.	None

Returns:

Type	Description
str	Config ID (implementation-defined).

Raises:

Type	Description
ValueError	If (pipeline_name, pipeline_version) already exists.

get_config abstractmethod

get_config(pipeline_name: str, pipeline_version: str) -> dict[str, Any] | None

Retrieve the full stored configuration.

Returns:

Type	Description
dict[str, Any] | None	Dict with keys: pipeline_name, pipeline_version, steps,
dict[str, Any] | None	settings, variables, description, status, created_at, etc.
dict[str, Any] | None	None if not found.

get_pipeline_dict

get_pipeline_dict(pipeline_name: str, pipeline_version: str) -> dict[str, Any] | None

Return a config dict ready for Pipeline.from_dict().

Parameters:

Name	Type	Description	Default
pipeline_name	str	Pipeline identifier.	required
pipeline_version	str	Pipeline version string.	required

Returns:

Type	Description
dict[str, Any] | None	Dict with name, version, steps, and optional
dict[str, Any] | None	settings/variables — or None if not found.

list_configs abstractmethod

list_configs(pipeline_name: str | None = None) -> list[dict[str, Any]]

List stored pipeline configurations.

Parameters:

Name	Type	Description	Default
pipeline_name	str | None	Optional filter by pipeline name.	None

Returns:

Type	Description
list[dict[str, Any]]	List of summary dicts.

list_versions abstractmethod

list_versions(pipeline_name: str) -> list[str]

List all stored versions for a given pipeline.

update_status abstractmethod

update_status(pipeline_name: str, pipeline_version: str, status: str, *, updated_by: str | None = None) -> None

Update the lifecycle status of a stored config.

Raises:

Type	Description
ValueError	If the config does not exist.

update_config abstractmethod

update_config(pipeline_name: str, pipeline_version: str, *, steps: list[dict[str, Any]] | None = None, settings: dict[str, Any] | None = None, variables: dict[str, Any] | None = None, description: str | None = None, updated_by: str | None = None) -> None

Patch mutable fields on a stored config.

Only provided fields are updated; None means "leave unchanged".

Raises:

Type	Description
ValueError	If the config does not exist.

delete_config abstractmethod

delete_config(pipeline_name: str, pipeline_version: str) -> bool

Delete a stored pipeline configuration.

Returns:

Type	Description
bool	True if deleted, False if not found.

Key methods

Method	Description
store_config(pipeline_name, pipeline_version, steps, ...)	Store a pipeline configuration (immutable per version). Optional: settings, variables, description, contract_name, contract_version, created_by.
get_config(pipeline_name, pipeline_version)	Retrieve the full stored config dict.
get_pipeline_dict(pipeline_name, pipeline_version)	Return a dict ready for Pipeline.from_dict() (name, version, steps, settings, variables).
list_configs(pipeline_name=None)	List stored configs, optionally filtered by pipeline name.
list_versions(pipeline_name)	List all stored versions for a pipeline.
update_status(...)	Update lifecycle status of a config.
update_config(...)	Patch mutable fields (steps, settings, variables, description).
delete_config(pipeline_name, pipeline_version)	Delete a stored config.

SQLite and Postgres

from pycharter.pipeline_store import SQLitePipelineStore, PostgresPipelineStore

# SQLite (development)
store = SQLitePipelineStore(connection_string="sqlite:///pycharter.db")
store.connect()

# Postgres (production / API)
store = PostgresPipelineStore(connection_string="postgresql://user:pass@localhost/pycharter")
store.connect()

In-memory

from pycharter.pipeline_store import InMemoryPipelineStore

store = InMemoryPipelineStore()
store.connect()
# Data is lost when the process exits

See also

• Contract Store — Schema registry for data contracts
• Semantic Store — Ontology registry
• Pipeline API — Building and running pipelines
• Building Pipelines