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: VersionedArtifactStore

Versioned store for steps-native pipeline configs.

The document passed to :meth:save / :meth:update carries {steps, settings, variables, description, metadata, contract_name, contract_version}; steps is required on :meth:save.

connect `abstractmethod` ¶

connect() -> None

Establish the backend connection.

disconnect ¶

disconnect() -> None

Close the backend connection.

list_versions ¶

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

Return all versions for name, oldest to newest.

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