---

name: flowerpower description: Create and manage data pipelines using the FlowerPower framework with Hamilton DAGs and uv. Use when users request creating flowerpower projects, pipelines, Hamilton dataflows, or ask about flowerpower configuration, execution, or CLI commands.

FlowerPower Pipeline Skill

Create and manage data processing pipelines using FlowerPower with Hamilton DAGs.

Quick Start

Copy# Install flowerpower
uv pip install flowerpower

# Initialize project
flowerpower init --name my-project

# Create pipeline
flowerpower pipeline new my_pipeline

# Run pipeline
flowerpower pipeline run my_pipeline

Project Initialization

Use scripts/init_project.py or CLI:

Copy# CLI
flowerpower init --name <project-name>

# Python
from flowerpower import FlowerPowerProject
project = FlowerPowerProject.init(name='my-project')

Creates structure:

Copymy-project/
├── conf/
│   ├── project.yml
│   └── pipelines/
├── pipelines/
└── hooks/

Creating Pipelines

Use scripts/create_pipeline.py or CLI:

Copyflowerpower pipeline new <name>

Creates:

• pipelines/<name>.py - Hamilton functions
• conf/pipelines/<name>.yml - Configuration

Pipeline Module Template

Copyfrom hamilton.function_modifiers import parameterize
from pathlib import Path
from flowerpower.cfg import Config

PARAMS = Config.load(
    Path(__file__).parents[1], pipeline_name="my_pipeline"
).pipeline.h_params

@parameterize(**PARAMS.input_config)
def load_data(source: str) -> dict:
    """Load data from source."""
    return {"source": source}

def process_data(load_data: dict) -> dict:
    """Process loaded data."""
    return {"processed": load_data}

def final_result(process_data: dict) -> str:
    """Return final result."""
    return str(process_data)

Pipeline Config Template

Copyparams:
  input_config:
    source: "data.csv"

run:
  final_vars:
    - final_result
  executor:
    type: threadpool
    max_workers: 4
  retry:
    max_retries: 3
    retry_delay: 1.0

Running Pipelines

Copy# Basic run
flowerpower pipeline run my_pipeline

# With inputs
flowerpower pipeline run my_pipeline --inputs '{"key": "value"}'

# With executor
flowerpower pipeline run my_pipeline --executor threadpool --executor-max-workers 8

# With retries
flowerpower pipeline run my_pipeline --max-retries 3 --retry-delay 2.0

Python API:

Copyfrom flowerpower import FlowerPowerProject

project = FlowerPowerProject.load('.')
result = project.run('my_pipeline')

# With RunConfig
from flowerpower.cfg.pipeline.run import RunConfig
config = RunConfig(inputs={"key": "value"}, final_vars=["output"])
result = project.run('my_pipeline', run_config=config)

CLI Commands

Executor Types

Optional Dependencies

Copyuv pip install flowerpower[io]   # I/O plugins
uv pip install flowerpower[ui]   # Hamilton UI
uv pip install flowerpower[all]  # All extras

Resources

• references/overview.md - Key concepts, architecture, project structure
• references/configuration.md - Complete YAML configuration patterns
• references/hamilton-patterns.md - Hamilton function decorators and patterns

Scripts

• scripts/init_project.py - Initialize new flowerpower project
• scripts/create_pipeline.py - Create new pipeline with template
• scripts/run_pipeline.py - Execute pipeline with options
• scripts/list_pipelines.py - List available pipelines