Skip to content

PipelineRegistry

Module: flowerpower.pipeline.registry.PipelineRegistry

The registry manages pipeline discovery, creation, deletion, summaries, and hooks.

Initialization

from_filesystem

1
2
3
4
5
6
from_filesystem(
    cls,
    base_dir: str,
    fs: AbstractFileSystem | None = None,
    storage_options: dict | None = None,
) -> PipelineRegistry

Create a registry instance from a filesystem path.

Parameter Type Description
base_dir str Project base directory.
fs AbstractFileSystem \| None fsspec-compatible filesystem.
storage_options dict \| None Storage options. 
1
registry = PipelineRegistry.from_filesystem("/path/to/project")

Methods

new / create_pipeline

1
2
new(self, name: str, overwrite: bool = False) -> None
create_pipeline(self, name: str, overwrite: bool = False) -> None

Create a new pipeline. create_pipeline is an alias for new and is the form used by the manager's creator.

1
2
3
4
from flowerpower.pipeline import PipelineManager

manager = PipelineManager()
manager.creator.create_pipeline("data_transformation")

delete / delete_pipeline

1
2
delete(self, name: str, cfg: bool = True, module: bool = False) -> None
delete_pipeline(self, name: str, cfg: bool = True, module: bool = False) -> None

Delete a pipeline's configuration file, module, or both. delete_pipeline is an alias for delete.

1
manager.registry.delete("old_pipeline", cfg=True, module=True)

get_pipeline

1
2
3
4
5
6
get_pipeline(
    self,
    name: str,
    project_context: FlowerPowerProject,
    reload: bool = False,
) -> Pipeline

Load a Pipeline instance, injecting the project context.

1
pipeline = manager.registry.get_pipeline("my_pipeline", project)

list_pipelines

1
list_pipelines(self) -> list[str]

Return the sorted list of pipeline names.

1
names = manager.registry.list_pipelines()

show_pipelines

1
show_pipelines(self) -> None

Print all available pipelines in a formatted table.

1
manager.registry.show_pipelines()

get_summary

1
2
3
4
5
6
7
get_summary(
    self,
    name: str | None = None,
    cfg: bool = True,
    code: bool = True,
    project: bool = True,
) -> dict[str, dict | str]

Return a detailed summary of one or all pipelines. The summary contains configuration, code, and project sections.

1
2
summary = manager.registry.get_summary("my_pipeline")
print(summary["pipelines"]["my_pipeline"]["cfg"]["name"])

show_summary

1
2
3
4
5
6
7
8
9
show_summary(
    self,
    name: str | None = None,
    cfg: bool = True,
    code: bool = True,
    project: bool = True,
    to_html: bool = False,
    to_svg: bool = False,
) -> str | None

Print or return a formatted summary of one or all pipelines.

1
output = manager.registry.show_summary("my_pipeline", to_html=True)

add_hook

1
2
3
4
5
6
7
add_hook(
    self,
    name: str,
    type: HookType,
    to: str | None = None,
    function_name: str | None = None,
) -> None

Add a hook to a pipeline module.

Parameter Type Description
name str Pipeline name.
type HookType Hook type. Only MQTT_BUILD_CONFIG is supported.
to str \| None Target file or node.
function_name str \| None Function name. 
1
2
3
4
5
6
7
from flowerpower.pipeline.registry import HookType

manager.registry.add_hook(
    name="my_pipeline",
    type=HookType.MQTT_BUILD_CONFIG,
    function_name="build_mqtt_config",
)

clear_cache

1
clear_cache(self, name: str | None = None) -> None

Clear cached pipeline data. If name is provided, only that pipeline's cache is cleared.

1
2
manager.registry.clear_cache("my_pipeline")
manager.registry.clear_cache()  # clear all

Properties

Property Type Description
pipelines list[str] Names of all discovered pipelines.
summary dict Full summary of all pipelines.
project_cfg ProjectConfig Current project configuration.