<div align="center">
<h1>FlowerPower 🌸 - Build & Orchestrate Data Pipelines</h1>
<h3>Simple Workflow Framework - Hamilton + APScheduler or RQ = FlowerPower</h3>
<img src="./image.png" alt="FlowerPower Logo" width="400" height="300">
</div>
[](https://pypi.org/project/flowerpower/) <!-- Placeholder -->
[](https://github.com/legout/flowerpower/blob/main/LICENSE)
[](https://deepwiki.com/legout/flowerpower)
**FlowerPower** is a Python framework designed for building, configuring, scheduling, and executing data processing pipelines with ease and flexibility. It promotes a modular, configuration-driven approach, allowing you to focus on your pipeline logic while FlowerPower handles the orchestration.
It is leveraging the [Hamilton](https://github.com/DAGWorks-Inc/hamilton) library for defining dataflows in a clean, functional way within your Python pipeline scripts. Pipelines are defined in Python modules and configured using YAML files, making it easy to manage and understand your data workflows.
FlowerPower integrates with job queue systems like [APScheduler](https://github.com/scheduler/apscheduler) and [RQ](https://github.com/rq/rq), enabling you to schedule and manage your pipeline runs efficiently. It also provides a web UI (Hamilton UI) for monitoring and managing your pipelines.
FlowerPower is designed to be extensible, allowing you to easily swap components like job queue backends or add custom I/O plugins. This flexibility makes it suitable for a wide range of data processing tasks, from simple ETL jobs to complex data workflows.
## ✨ Key Features
* **Modular Pipeline Design:** Thanks to [Hamilton](https://github.com/DAGWorks-Inc/hamilton), you can define your data processing logic in Python modules, using functions as nodes in a directed acyclic graph (DAG).
* **Configuration-Driven:** Define pipeline parameters, execution logic, and scheduling declaratively using simple YAML files.
* **Job Queue Integration:** Built-in support for different asynchronous execution models:
* **APScheduler:** For time-based scheduling (cron, interval, date).
* **RQ (Redis Queue):** For distributed task queues.
* **Extensible I/O Plugins:** Connect to various data sources and destinations (CSV, JSON, Parquet, DeltaTable, DuckDB, PostgreSQL, MySQL, MSSQL, Oracle, MQTT, SQLite, and more).
* **Multiple Interfaces:** Interact with your pipelines via:
* **Command Line Interface (CLI):** For running, managing, and inspecting pipelines.
* **Web UI:** A graphical interface for monitoring and managing pipelines and schedules. ([Hamilton UI](https://hamilton.dagworks.io/en/latest/hamilton-ui/ui/))
* **Filesystem Abstraction:** Simplified file handling with support for local and remote filesystems (e.g., S3, GCS).
## 📦 Installation
We recommend using [uv](https://github.com/astral-sh/uv) for installing FlowerPower and managing your project environments. `uv` is an extremely fast Python package installer and resolver.
```bash
# Create and activate a virtual environment (recommended)
uv venv
source .venv/bin/activate # Or .\.venv\Scripts\activate on Windows
# Install FlowerPower
uv pip install flowerpower
# Optional: Install additional dependencies for specific features
uv pip install flowerpower[apscheduler,rq] # Example for APScheduler and RQ
uv pip install flowerpower[io] # Example for I/O plugins (CSV, JSON, Parquet, DeltaTable, DuckDB, PostgreSQL, MySQL, MSSQL, Oracle, SQLite)
uv pip install flowerpower[ui] # Example for Hamilton UI
uv pip install flowerpower[all] # Install all optional dependencies
```
*(Note: Specify required Python versions if known, e.g., Python 3.8+)*
## 🚀 Getting Started
Let's build a simple "Hello World" pipeline.
### 1. Initialize Your Project:
You can quickly set up the standard FlowerPower project structure using the CLI or Python.
**Using the CLI:**
Navigate to your desired parent directory and run:
```bash
flowerpower init --name hello-flowerpower-project
```
**Using Python:**
Alternatively, you can initialize programmatically:
```python
from flowerpower import init_project
# Creates the structure in the current directory
init_project(name='hello-flowerpower-project', job_queue_type='rq') # Or 'apscheduler'
```
This will create a `hello-flowerpower-project` directory with the necessary `conf/` and `pipelines/` subdirectories and default configuration files.
```
hello-flowerpower-project/
├── conf/
│ ├── project.yml
│ └── pipelines/
└── pipelines/
```
Now, navigate into your new project directory:
```bash
cd hello-flowerpower-project
```
**Configure Project (`conf/project.yml`):**
Open `conf/project.yml` and define your project name and choose your job queue backend. Here's an example using RQ:
```yaml
name: hello-flowerpower
job_queue:
type: rq
backend:
type: redis
host: localhost
port: 6379
# ... other redis options
queues:
- default
- high
- low
# adapter: ... # Optional adapter configurations (e.g., Hamilton Tracker, MLflow), see `conf/project.yml` for details
```
### 2. Create Your Pipeline
You can create a new pipeline using the CLI or programmatically.
**Using the CLI:**
```bash
flowerpower pipeline new hello_world
```
**Using Python:**
There is a `PipelineManager` class to manage pipelines programmatically:
```python
from flowerpower.pipeline import PipelineManager
pm = PipelineManager(base_dir='.')
pm.new(name='hello_world') # Creates a new pipeline
```
This will create a new file `hello_world.py` in the `pipelines/` directory and a corresponding configuration file `hello_world.yml` in `conf/pipelines/`.
**Implement Pipeline (`pipelines/hello_world.py`):**
Open `pipelines/hello_world.py` and write your pipeline logic using Python and Hamilton. FlowerPower makes configuration easily accessible.
```python
# FlowerPower pipeline hello_world.py
# Created on 2025-05-03 22:34:09
####################################################################################################
# Import necessary libraries
# NOTE: Remove or comment out imports that are not used in the pipeline
from hamilton.function_modifiers import parameterize
from pathlib import Path
from flowerpower.cfg import Config
####################################################################################################
# Load pipeline parameters. Do not modify this section.
PARAMS = Config.load(
Path(__file__).parents[1], pipeline_name="hello_world"
).pipeline.h_params
####################################################################################################
# Helper functions.
# This functions have to start with an underscore (_).
####################################################################################################
# Pipeline functions
@parameterize(**PARAMS.greeting_message) # Inject 'message' from params
def greeting_message(message: str) -> str:
"""Provides the greeting part."""
return f"{message},"
@parameterize(**PARAMS.target_name) # Inject 'name' from params
def target_name(name: str) -> str:
"""Provides the target name."""
return f"{name}!"
def full_greeting(greeting_message: str, target_name: str) -> str:
"""Combines the greeting and target."""
print(f"Generating greeting: {greeting_message} {target_name}")
return f"{greeting_message} {target_name}"
# You can add more complex Hamilton functions here...
```
**Configure Pipeline (`conf/pipelines/hello_world.yml`):**
Open `conf/pipelines/hello_world.yml` and specify parameters, run configurations, and scheduling for your pipeline.
```yaml
# adapter: ... # Pipeline-specific adapter overrides
params: # Parameters accessible in your Python code
greeting_message:
message: "Hello"
target:
name: "World"
run: # How to execute the pipeline
final_vars: # Specify the desired output(s) from your Hamilton DAG
- full_greeting
# inputs: # Optional: Specify input variables to the pipeline
# message: "Hello"
# config: ... # Runtime configuration overrides for Hamilton
# executor: ... # Execution backend (e.g., threadpool, multiprocessing)
schedule: # Optional: How often to run the pipeline
cron: "0 * * * *" # Run hourly
# interval: # e.g., { "minutes": 15 }
# date: # e.g., "2025-12-31 23:59:59"
```
### 3. Run Your Pipeline 🏃♀️
FlowerPower offers flexibility in how you execute your pipelines:
- **Synchronous Execution:** Run the pipeline directly.
- **Asynchronous Execution:** Use job queues for scheduling, background execution, or distributed processing.
#### 1. Synchronous Execution:
For quick testing or local runs, you can execute your pipeline synchronously. This is useful for debugging or running pipelines in a local environment.
* **Via CLI:**
```bash
# Run the pipeline synchronously
flowerpower pipeline run hello_world --base_dir .
```
* **Via Python:**
```python
from flowerpower.pipeline import PipelineManager
pm = PipelineManager(base_dir='.')
pm.run('hello_world') # Execute the pipeline named 'hello_world'
#### 2. Asynchronous Execution (Job Queues):
For scheduling, background execution, or distributed processing, leverage FlowerPower's job queue integration. Ideal for distributed task queues where workers can pick up jobs.
You have to install the job queue backend you want to use. FlowerPower supports two job queue backends: RQ (Redis Queue) and APScheduler.
```bash
# Install RQ (Redis Queue) or APScheduler
uv pip install flowerpower[rq] # For RQ (Redis Queue)
uv pip install flowerpower[apscheduler] # For APScheduler
```
* **Note:** Ensure you have the required dependencies installed for your chosen job queue backend. For RQ, you need Redis running. For APScheduler, you need a data store (PostgreSQL, MySQL, SQLite, MongoDB) and an event broker (Redis, MQTT, PostgreSQL).
**a) Configuring Job Queue Backends:**
Configuration of the job queue backend is done in your `conf/project.yml`. Currently, FlowerPower supports two job queue backends:
* **RQ (Redis Queue):**
* **Requires:** Access to a running Redis server.
* Configure in `conf/project.yml`:
```yaml
job_queue:
type: rq
backend:
type: redis
host: localhost
port: 6379
... # other redis options
* **APScheduler:**
* **Requires:**
* A **Data Store:** To persist job information (Options: PostgreSQL, MySQL, SQLite, MongoDB).
* An **Event Broker:** To notify workers of scheduled jobs (Options: Redis, MQTT, PostgreSQL).
* Configure in `cong/project.yml`:
```yaml
job_queue:
type: apscheduler
backend:
type: postgresql # or mysql, sqlite, mongodb
host: localhost
port: 5432
user: your_user
password: your_password
database: your_database
... # other database options
event_broker:
type: redis # or mqtt, postgresql
host: localhost
port: 6379
... # other redis options
```
It is possible to override the job queue backend configuration using environment variables, the `settings` module or by monkey patching the backend configuration of the `PipelineManager` or `JobQueueManager` classes. This might be useful for testing or when you want to avoid hardcoding values in your configuration files.
* **Using the `settings` module:**
e.g to override the RQ backend username and password:
```python
from flowerpower import settings
# Override some configuration values. e.g. when using rq
settings.RQ_BACKEND_USERNAME = 'your_username'
settings.RQ_BACKEND_PASSWORD = 'your_password'
```
See the `flowerpower/settings/job_queue.py` file for all available settings.
* **Monkey Patching:**
e.g to override the APScheduler data store username and password:
```python
from flowerpower.pipeline import PipelineManager
pm = PipelineManager(base_dir='.')
pm.project_cfg.job_queue.backend.username = 'your_username'
pm.project_cfg.job_queue.backend.password = 'your_password'
```
* **Using Environment Variables:**
e.g. use a `.env` file or set them in your environment. Here is a list of the available environment variables for the job queue backend configuration:
```
FP_JOB_QUEUE_TYPE
# RQ (Redis Queue) backend
FP_RQ_BACKEND
FP_RQ_BACKEND_USERNAME
FP_RQ_BACKEND_PASSWORD
FP_RQ_BACKEND_HOST
FP_RQ_BACKEND_PORT
# APScheduler data store
FP_APS_BACKEND_DS
FP_APS_BACKEND_DS_USERNAME
FP_APS_BACKEND_DS_PASSWORD
FP_APS_BACKEND_DS_HOST
FP_APS_BACKEND_DS_PORT
# APScheduler event broker
FP_APS_BACKEND_EB
FP_APS_BACKEND_EB_USERNAME
FP_APS_BACKEND_EB_PASSWORD
FP_APS_BACKEND_EB_HOST
FP_APS_BACKEND_EB_PORT
```
**b) Add Job to Queue:**
Run your pipeline using the job queue system. This allows you to schedule jobs, run them in the background, or distribute them across multiple workers.
* **Via CLI:**
```bash
# This will run the pipeline immediately and return the job result (blocking, until the job is done)
flowerpower pipeline run-job hello_world --base_dir .
# Submit the pipeline to the job queue and return the job ID (non-blocking)
flowerpower pipeline add-job hello_world --base_dir .
```
* **Via Python:**
```python
from flowerpower.pipeline import PipelineManager
pm = PipelineManager(base_dir='.')
# submit the pipeline to the job queue and return the job ID (non-blocking)
job_id = pm.add_job('hello_world')
# submit the pipeline to the job queue, runs it immediately and returns the job ID (non-blocking)
result = pm.run_job('hello_world')
```
These commands will add the pipeline to the job queue, allowing it to be executed in the background or at scheduled intervals. The jobs will be processed by one or more workers, depending on your job queue configuration. You have to start the job queue workers separately.
**c) Start Job Queue Workers:**
To process jobs in the queue, you need to start one or more workers.
* **Via CLI:**
```bash
flowerpower job-queue start-worker --base_dir . # Start the job queue worker
```
* **Via Python:**
```python
from flowerpower.job_queue import JobQueueManager
with JobQueueManager(base_dir='.'):
# Start the job queue worker
jqm.start_worker()
```
## Local Development Setup (Docker):
To easily set up required services like Redis, PostgreSQL, or MQTT locally for testing job queues, a basic `docker-compose.yml` file is provided in the `docker/` directory. This file includes configurations for various services useful during development.
```bash
# Navigate to the docker directory and start services
cd docker
docker-compose up -d redis postgres # Example: Start Redis and PostgreSQL
```
*(Note: Review and adapt `docker/docker-compose.yml` for your specific needs. It's intended for development, not production.)*
## ⚙️ Configuration Overview
FlowerPower uses a layered configuration system:
* **`conf/project.yml`:** Defines global settings for your project, primarily the `job_queue` backend (RQ or APScheduler) and configurations for integrated `adapter`s (like Hamilton Tracker, MLflow, etc.).
* **`conf/pipelines/*.yml`:** Each file defines a specific pipeline. It contains:
* `params`: Input parameters for your Hamilton functions.
* `run`: Execution details like target outputs (`final_vars`), Hamilton runtime `config`, and `executor` settings.
* `schedule`: Defines when the pipeline should run automatically (using `cron`, `interval`, or `date`).
* `adapter`: Pipeline-specific overrides for adapter settings.
## 🛠️ Basic Usage
The primary way to interact with pipelines is often through the CLI:
```bash
# Run a pipeline manually
flowerpower pipeline run hello_world --base_dir .
# Add a job to the queue
flowerpower pipeline add-job hello_world --base_dir .
# Schedule a pipeline
flowerpower pipeline schedule hello_world --base_dir . # Schedules like cron, interval, or date are configured in the pipeline config
# And many more commands...
flowerpower --help # List all available commands
```
## 🖥️ UI
The FlowerPower web UI (Hamilton UI) provides a graphical interface for monitoring and managing your pipelines. It allows you to visualize pipeline runs, schedules, and potentially manage configurations.
```bash
# Start the web UI
flowerpower ui
```
## 📖 Documentation
There is not much documentation yet, but you can find some examples in the `examples/` directory. The examples cover various use cases, including:
* Basic pipeline creation and execution.
* Using different job queue backends (RQ and APScheduler).
* Configuring and scheduling pipelines.
There is a first version of documentation in `docs/`. This documentation is generated using [Pocket Flow Tutorial Project](https://github.com/The-Pocket/PocketFlow-Tutorial-Codebase-Knowledge). Although it is not complete and might be wrong in some parts, it can be a good starting point for understanding how to use FlowerPower.
## 📜 License
This project is licensed under the MIT License - see the `LICENSE` file for details. (Placeholder - update with actual license)
Raw data
{
"_id": null,
"home_page": null,
"name": "FlowerPower",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "hamilton, workflow, pipeline, scheduler, apscheduler, dask, ray",
"author": null,
"author_email": "\"Volker L.\" <ligno.blades@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/fb/b1/47b3022931ebaf238462d2f58f22e6d54e7d2711d24cf951fa96de6880e2/flowerpower-0.11.6.19.tar.gz",
"platform": null,
"description": "<div align=\"center\">\n <h1>FlowerPower \ud83c\udf38 - Build & Orchestrate Data Pipelines</h1>\n <h3>Simple Workflow Framework - Hamilton + APScheduler or RQ = FlowerPower</h3>\n <img src=\"./image.png\" alt=\"FlowerPower Logo\" width=\"400\" height=\"300\">\n</div>\n\n\n\n[](https://pypi.org/project/flowerpower/) <!-- Placeholder -->\n[](https://github.com/legout/flowerpower/blob/main/LICENSE)\n[](https://deepwiki.com/legout/flowerpower)\n\n\n**FlowerPower** is a Python framework designed for building, configuring, scheduling, and executing data processing pipelines with ease and flexibility. It promotes a modular, configuration-driven approach, allowing you to focus on your pipeline logic while FlowerPower handles the orchestration.\n\nIt is leveraging the [Hamilton](https://github.com/DAGWorks-Inc/hamilton) library for defining dataflows in a clean, functional way within your Python pipeline scripts. Pipelines are defined in Python modules and configured using YAML files, making it easy to manage and understand your data workflows.\nFlowerPower integrates with job queue systems like [APScheduler](https://github.com/scheduler/apscheduler) and [RQ](https://github.com/rq/rq), enabling you to schedule and manage your pipeline runs efficiently. It also provides a web UI (Hamilton UI) for monitoring and managing your pipelines.\nFlowerPower is designed to be extensible, allowing you to easily swap components like job queue backends or add custom I/O plugins. This flexibility makes it suitable for a wide range of data processing tasks, from simple ETL jobs to complex data workflows.\n\n\n## \u2728 Key Features\n\n* **Modular Pipeline Design:** Thanks to [Hamilton](https://github.com/DAGWorks-Inc/hamilton), you can define your data processing logic in Python modules, using functions as nodes in a directed acyclic graph (DAG).\n* **Configuration-Driven:** Define pipeline parameters, execution logic, and scheduling declaratively using simple YAML files.\n* **Job Queue Integration:** Built-in support for different asynchronous execution models:\n * **APScheduler:** For time-based scheduling (cron, interval, date).\n * **RQ (Redis Queue):** For distributed task queues.\n* **Extensible I/O Plugins:** Connect to various data sources and destinations (CSV, JSON, Parquet, DeltaTable, DuckDB, PostgreSQL, MySQL, MSSQL, Oracle, MQTT, SQLite, and more).\n* **Multiple Interfaces:** Interact with your pipelines via:\n * **Command Line Interface (CLI):** For running, managing, and inspecting pipelines.\n * **Web UI:** A graphical interface for monitoring and managing pipelines and schedules. ([Hamilton UI](https://hamilton.dagworks.io/en/latest/hamilton-ui/ui/))\n* **Filesystem Abstraction:** Simplified file handling with support for local and remote filesystems (e.g., S3, GCS).\n\n## \ud83d\udce6 Installation\n\nWe recommend using [uv](https://github.com/astral-sh/uv) for installing FlowerPower and managing your project environments. `uv` is an extremely fast Python package installer and resolver.\n\n```bash\n# Create and activate a virtual environment (recommended)\nuv venv\nsource .venv/bin/activate # Or .\\.venv\\Scripts\\activate on Windows\n\n# Install FlowerPower\nuv pip install flowerpower\n\n# Optional: Install additional dependencies for specific features\nuv pip install flowerpower[apscheduler,rq] # Example for APScheduler and RQ\nuv pip install flowerpower[io] # Example for I/O plugins (CSV, JSON, Parquet, DeltaTable, DuckDB, PostgreSQL, MySQL, MSSQL, Oracle, SQLite)\nuv pip install flowerpower[ui] # Example for Hamilton UI\nuv pip install flowerpower[all] # Install all optional dependencies\n```\n\n*(Note: Specify required Python versions if known, e.g., Python 3.8+)*\n\n## \ud83d\ude80 Getting Started\n\nLet's build a simple \"Hello World\" pipeline.\n\n### 1. Initialize Your Project:\n\nYou can quickly set up the standard FlowerPower project structure using the CLI or Python.\n\n**Using the CLI:**\n\nNavigate to your desired parent directory and run:\n```bash\nflowerpower init --name hello-flowerpower-project\n```\n\n\n**Using Python:**\n\nAlternatively, you can initialize programmatically:\n```python\nfrom flowerpower import init_project\n\n# Creates the structure in the current directory\ninit_project(name='hello-flowerpower-project', job_queue_type='rq') # Or 'apscheduler'\n```\n\nThis will create a `hello-flowerpower-project` directory with the necessary `conf/` and `pipelines/` subdirectories and default configuration files.\n\n```\nhello-flowerpower-project/\n\u251c\u2500\u2500 conf/\n\u2502 \u251c\u2500\u2500 project.yml\n\u2502 \u2514\u2500\u2500 pipelines/\n\u2514\u2500\u2500 pipelines/\n```\n\nNow, navigate into your new project directory:\n\n```bash\ncd hello-flowerpower-project\n```\n\n**Configure Project (`conf/project.yml`):**\n\nOpen `conf/project.yml` and define your project name and choose your job queue backend. Here's an example using RQ:\n\n```yaml\nname: hello-flowerpower\njob_queue:\n type: rq\n backend:\n type: redis\n host: localhost\n port: 6379\n # ... other redis options\n queues:\n - default\n - high\n - low\n# adapter: ... # Optional adapter configurations (e.g., Hamilton Tracker, MLflow), see `conf/project.yml` for details\n```\n\n### 2. Create Your Pipeline\n\nYou can create a new pipeline using the CLI or programmatically.\n\n**Using the CLI:**\n\n```bash\nflowerpower pipeline new hello_world\n```\n\n**Using Python:**\n\nThere is a `PipelineManager` class to manage pipelines programmatically:\n\n```python\nfrom flowerpower.pipeline import PipelineManager\npm = PipelineManager(base_dir='.')\npm.new(name='hello_world') # Creates a new pipeline\n```\n\nThis will create a new file `hello_world.py` in the `pipelines/` directory and a corresponding configuration file `hello_world.yml` in `conf/pipelines/`.\n\n**Implement Pipeline (`pipelines/hello_world.py`):**\n\nOpen `pipelines/hello_world.py` and write your pipeline logic using Python and Hamilton. FlowerPower makes configuration easily accessible.\n\n```python\n# FlowerPower pipeline hello_world.py\n# Created on 2025-05-03 22:34:09\n\n####################################################################################################\n# Import necessary libraries\n# NOTE: Remove or comment out imports that are not used in the pipeline\n\nfrom hamilton.function_modifiers import parameterize\n\nfrom pathlib import Path\n\nfrom flowerpower.cfg import Config\n\n####################################################################################################\n# Load pipeline parameters. Do not modify this section.\n\nPARAMS = Config.load(\n Path(__file__).parents[1], pipeline_name=\"hello_world\"\n).pipeline.h_params\n\n\n####################################################################################################\n# Helper functions.\n# This functions have to start with an underscore (_).\n\n\n####################################################################################################\n# Pipeline functions\n\n@parameterize(**PARAMS.greeting_message) # Inject 'message' from params\ndef greeting_message(message: str) -> str:\n \"\"\"Provides the greeting part.\"\"\"\n return f\"{message},\"\n\n@parameterize(**PARAMS.target_name) # Inject 'name' from params\ndef target_name(name: str) -> str:\n \"\"\"Provides the target name.\"\"\"\n return f\"{name}!\"\n\ndef full_greeting(greeting_message: str, target_name: str) -> str:\n \"\"\"Combines the greeting and target.\"\"\"\n print(f\"Generating greeting: {greeting_message} {target_name}\")\n return f\"{greeting_message} {target_name}\"\n\n# You can add more complex Hamilton functions here...\n```\n\n**Configure Pipeline (`conf/pipelines/hello_world.yml`):**\n\nOpen `conf/pipelines/hello_world.yml` and specify parameters, run configurations, and scheduling for your pipeline.\n\n```yaml\n# adapter: ... # Pipeline-specific adapter overrides\n\nparams: # Parameters accessible in your Python code\n greeting_message:\n message: \"Hello\"\n target:\n name: \"World\"\n\nrun: # How to execute the pipeline\n final_vars: # Specify the desired output(s) from your Hamilton DAG\n - full_greeting\n # inputs: # Optional: Specify input variables to the pipeline\n # message: \"Hello\"\n # config: ... # Runtime configuration overrides for Hamilton\n # executor: ... # Execution backend (e.g., threadpool, multiprocessing)\n\nschedule: # Optional: How often to run the pipeline\n cron: \"0 * * * *\" # Run hourly\n # interval: # e.g., { \"minutes\": 15 }\n # date: # e.g., \"2025-12-31 23:59:59\"\n```\n### 3. Run Your Pipeline \ud83c\udfc3\u200d\u2640\ufe0f\n\nFlowerPower offers flexibility in how you execute your pipelines:\n - **Synchronous Execution:** Run the pipeline directly.\n - **Asynchronous Execution:** Use job queues for scheduling, background execution, or distributed processing.\n\n#### 1. Synchronous Execution:\n\nFor quick testing or local runs, you can execute your pipeline synchronously. This is useful for debugging or running pipelines in a local environment.\n\n* **Via CLI:**\n ```bash\n # Run the pipeline synchronously\n flowerpower pipeline run hello_world --base_dir .\n ```\n* **Via Python:**\n ```python\n from flowerpower.pipeline import PipelineManager\n pm = PipelineManager(base_dir='.')\n pm.run('hello_world') # Execute the pipeline named 'hello_world' \n\n#### 2. Asynchronous Execution (Job Queues):\n\nFor scheduling, background execution, or distributed processing, leverage FlowerPower's job queue integration. Ideal for distributed task queues where workers can pick up jobs. \n\nYou have to install the job queue backend you want to use. FlowerPower supports two job queue backends: RQ (Redis Queue) and APScheduler.\n```bash\n# Install RQ (Redis Queue) or APScheduler\nuv pip install flowerpower[rq] # For RQ (Redis Queue)\nuv pip install flowerpower[apscheduler] # For APScheduler\n```\n* **Note:** Ensure you have the required dependencies installed for your chosen job queue backend. For RQ, you need Redis running. For APScheduler, you need a data store (PostgreSQL, MySQL, SQLite, MongoDB) and an event broker (Redis, MQTT, PostgreSQL).\n\n**a) Configuring Job Queue Backends:** \n\nConfiguration of the job queue backend is done in your `conf/project.yml`. Currently, FlowerPower supports two job queue backends:\n\n* **RQ (Redis Queue):**\n * **Requires:** Access to a running Redis server.\n * Configure in `conf/project.yml`: \n ```yaml\n job_queue:\n type: rq\n backend:\n type: redis\n host: localhost\n port: 6379\n ... # other redis options\n\n* **APScheduler:**\n * **Requires:**\n * A **Data Store:** To persist job information (Options: PostgreSQL, MySQL, SQLite, MongoDB).\n * An **Event Broker:** To notify workers of scheduled jobs (Options: Redis, MQTT, PostgreSQL).\n * Configure in `cong/project.yml`:\n ```yaml\n job_queue:\n type: apscheduler\n backend:\n type: postgresql # or mysql, sqlite, mongodb\n host: localhost\n port: 5432\n user: your_user\n password: your_password\n database: your_database\n ... # other database options\n event_broker:\n type: redis # or mqtt, postgresql\n host: localhost\n port: 6379\n ... # other redis options\n ```\n \nIt is possible to override the job queue backend configuration using environment variables, the `settings` module or by monkey patching the backend configuration of the `PipelineManager` or `JobQueueManager` classes. This might be useful for testing or when you want to avoid hardcoding values in your configuration files.\n* **Using the `settings` module:**\n e.g to override the RQ backend username and password:\n ```python\n from flowerpower import settings\n \n # Override some configuration values. e.g. when using rq \n settings.RQ_BACKEND_USERNAME = 'your_username'\n settings.RQ_BACKEND_PASSWORD = 'your_password' \n ```\n See the `flowerpower/settings/job_queue.py` file for all available settings.\n\n* **Monkey Patching:**\n e.g to override the APScheduler data store username and password:\n ```python\n from flowerpower.pipeline import PipelineManager\n\n pm = PipelineManager(base_dir='.')\n pm.project_cfg.job_queue.backend.username = 'your_username'\n pm.project_cfg.job_queue.backend.password = 'your_password'\n ```\n* **Using Environment Variables:**\n e.g. use a `.env` file or set them in your environment. Here is a list of the available environment variables for the job queue backend configuration:\n ```\n FP_JOB_QUEUE_TYPE\n\n # RQ (Redis Queue) backend\n FP_RQ_BACKEND\n FP_RQ_BACKEND_USERNAME\n FP_RQ_BACKEND_PASSWORD\n FP_RQ_BACKEND_HOST\n FP_RQ_BACKEND_PORT\n\n # APScheduler data store\n FP_APS_BACKEND_DS\n FP_APS_BACKEND_DS_USERNAME\n FP_APS_BACKEND_DS_PASSWORD\n FP_APS_BACKEND_DS_HOST\n FP_APS_BACKEND_DS_PORT\n\n # APScheduler event broker\n FP_APS_BACKEND_EB\n FP_APS_BACKEND_EB_USERNAME\n FP_APS_BACKEND_EB_PASSWORD\n FP_APS_BACKEND_EB_HOST\n FP_APS_BACKEND_EB_PORT\n ```\n\n\n**b) Add Job to Queue:** \nRun your pipeline using the job queue system. This allows you to schedule jobs, run them in the background, or distribute them across multiple workers.\n\n* **Via CLI:**\n ```bash\n # This will run the pipeline immediately and return the job result (blocking, until the job is done)\n flowerpower pipeline run-job hello_world --base_dir . \n\n # Submit the pipeline to the job queue and return the job ID (non-blocking)\n flowerpower pipeline add-job hello_world --base_dir . \n ```\n* **Via Python:**\n \n ```python\n from flowerpower.pipeline import PipelineManager\n pm = PipelineManager(base_dir='.')\n\n # submit the pipeline to the job queue and return the job ID (non-blocking)\n job_id = pm.add_job('hello_world') \n\n # submit the pipeline to the job queue, runs it immediately and returns the job ID (non-blocking)\n result = pm.run_job('hello_world')\n ```\n\nThese commands will add the pipeline to the job queue, allowing it to be executed in the background or at scheduled intervals. The jobs will be processed by one or more workers, depending on your job queue configuration. You have to start the job queue workers separately.\n\n\n**c) Start Job Queue Workers:** \nTo process jobs in the queue, you need to start one or more workers.\n\n* **Via CLI:**\n ```bash\n flowerpower job-queue start-worker --base_dir . # Start the job queue worker\n ```\n\n* **Via Python:**\n ```python\n from flowerpower.job_queue import JobQueueManager\n with JobQueueManager(base_dir='.'):\n # Start the job queue worker\n jqm.start_worker()\n ```\n\n\n## Local Development Setup (Docker):\n\nTo easily set up required services like Redis, PostgreSQL, or MQTT locally for testing job queues, a basic `docker-compose.yml` file is provided in the `docker/` directory. This file includes configurations for various services useful during development.\n\n```bash\n# Navigate to the docker directory and start services\ncd docker\ndocker-compose up -d redis postgres # Example: Start Redis and PostgreSQL\n```\n*(Note: Review and adapt `docker/docker-compose.yml` for your specific needs. It's intended for development, not production.)*\n\n\n\n## \u2699\ufe0f Configuration Overview\n\nFlowerPower uses a layered configuration system:\n\n* **`conf/project.yml`:** Defines global settings for your project, primarily the `job_queue` backend (RQ or APScheduler) and configurations for integrated `adapter`s (like Hamilton Tracker, MLflow, etc.).\n* **`conf/pipelines/*.yml`:** Each file defines a specific pipeline. It contains:\n * `params`: Input parameters for your Hamilton functions.\n * `run`: Execution details like target outputs (`final_vars`), Hamilton runtime `config`, and `executor` settings.\n * `schedule`: Defines when the pipeline should run automatically (using `cron`, `interval`, or `date`).\n * `adapter`: Pipeline-specific overrides for adapter settings.\n\n## \ud83d\udee0\ufe0f Basic Usage\n\nThe primary way to interact with pipelines is often through the CLI:\n\n```bash\n# Run a pipeline manually\nflowerpower pipeline run hello_world --base_dir .\n\n# Add a job to the queue\nflowerpower pipeline add-job hello_world --base_dir .\n\n# Schedule a pipeline\nflowerpower pipeline schedule hello_world --base_dir . # Schedules like cron, interval, or date are configured in the pipeline config\n\n# And many more commands...\nflowerpower --help # List all available commands\n\n```\n\n## \ud83d\udda5\ufe0f UI\n\nThe FlowerPower web UI (Hamilton UI) provides a graphical interface for monitoring and managing your pipelines. It allows you to visualize pipeline runs, schedules, and potentially manage configurations.\n\n```bash\n# Start the web UI\nflowerpower ui\n```\n\n## \ud83d\udcd6 Documentation\n\nThere is not much documentation yet, but you can find some examples in the `examples/` directory. The examples cover various use cases, including:\n* Basic pipeline creation and execution.\n* Using different job queue backends (RQ and APScheduler).\n* Configuring and scheduling pipelines.\n\n\nThere is a first version of documentation in `docs/`. This documentation is generated using [Pocket Flow Tutorial Project](https://github.com/The-Pocket/PocketFlow-Tutorial-Codebase-Knowledge). Although it is not complete and might be wrong in some parts, it can be a good starting point for understanding how to use FlowerPower.\n\n\n## \ud83d\udcdc License\n\nThis project is licensed under the MIT License - see the `LICENSE` file for details. (Placeholder - update with actual license)\n",
"bugtrack_url": null,
"license": null,
"summary": "A simple workflow framework. Hamilton + APScheduler = FlowerPower",
"version": "0.11.6.19",
"project_urls": {
"Bug Tracker": "https://github.com/legout/flowerpower/issues",
"Homepage": "https://github.com/legout/flowerpower"
},
"split_keywords": [
"hamilton",
" workflow",
" pipeline",
" scheduler",
" apscheduler",
" dask",
" ray"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "302cf79670c575346956c8a8ede6ce0ff63db6818707d934da5fe63c25fdc52c",
"md5": "0d8e18c4ca9f0a43e6646784e865f8fa",
"sha256": "8239e4109263de9384939bd4beb57cdf50c8e3197deca7c45570032c1a270bf3"
},
"downloads": -1,
"filename": "flowerpower-0.11.6.19-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0d8e18c4ca9f0a43e6646784e865f8fa",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 213959,
"upload_time": "2025-07-16T19:14:25",
"upload_time_iso_8601": "2025-07-16T19:14:25.859137Z",
"url": "https://files.pythonhosted.org/packages/30/2c/f79670c575346956c8a8ede6ce0ff63db6818707d934da5fe63c25fdc52c/flowerpower-0.11.6.19-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "fbb147b3022931ebaf238462d2f58f22e6d54e7d2711d24cf951fa96de6880e2",
"md5": "ea73e34b827eba2cf734e3abcc754e38",
"sha256": "9878fc503ddd08bfc38490f618844d4b3e29d9ba65e858167754e94c8ace251d"
},
"downloads": -1,
"filename": "flowerpower-0.11.6.19.tar.gz",
"has_sig": false,
"md5_digest": "ea73e34b827eba2cf734e3abcc754e38",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 182825,
"upload_time": "2025-07-16T19:14:27",
"upload_time_iso_8601": "2025-07-16T19:14:27.521312Z",
"url": "https://files.pythonhosted.org/packages/fb/b1/47b3022931ebaf238462d2f58f22e6d54e7d2711d24cf951fa96de6880e2/flowerpower-0.11.6.19.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-16 19:14:27",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "legout",
"github_project": "flowerpower",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "flowerpower"
}
JSON
