### Install Pandas

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pandas_example.md

Install the pandas library to version 1.5.3 or higher. This is a prerequisite for running the example.

```console
pip install pandas>=1.5.3
```

--------------------------------

### Run an Example Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Instantiate the ExampleFlow class and call its run method to execute the flow. This is the primary way to start a flow in FlowRunner.

```python
ExampleFlow().run()
```

--------------------------------

### Install Development Packages

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Install FlowRunner along with its development dependencies for a full development setup.

```bash
pip install -e .[dev]
```

--------------------------------

### Install Documentation Packages

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Install FlowRunner along with its documentation generation dependencies.

```bash
pip install -e .[doc]
```

--------------------------------

### Install Pre-commit Hooks

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_docs.md

Install the pre-commit hooks to ensure code quality checks are performed before committing.

```bash
pre-commit install
```

--------------------------------

### Install PySpark

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pyspark_example.md

Install PySpark using pip. Ensure you have a compatible version.

```console
pip install pyspark>=3.3.2
```

--------------------------------

### Install Project with Pip

Source: https://github.com/prithvijitguha/flowrunner/blob/main/tests/test_examples/test_example_notebook.ipynb

Installs the current project using pip. This is often a prerequisite for running tests or using the library in other projects.

```bash
subprocess.check_call([sys.executable, "-m", "pip", "install", "."])
```

--------------------------------

### Install Flowrunner

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pyspark_example.html

Use this command to install or upgrade Flowrunner. The Python interpreter will restart after installation.

```python
%pip install --upgrade flowrunner
```

--------------------------------

### Install Testing Packages

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Install FlowRunner along with its testing dependencies.

```bash
pip install -e .[test]
```

--------------------------------

### Install Flowrunner

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pandas_notebook_example.html

Use this command to install the Flowrunner library with pip. It is recommended to install it in a virtual environment.

```bash
!pip install flowrunner
```

--------------------------------

### Install FlowRunner using pip

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Install the latest stable version of FlowRunner from PyPI.

```powershell
pip install flowrunner
```

--------------------------------

### Import Pandas

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pandas_notebook_example.html

This code imports the Pandas library, which is essential for data manipulation in this example. Ensure Pandas is installed in your environment.

```python
# -*- coding: utf-8 -*-
import pandas as pd
```

--------------------------------

### Install FlowRunner Development Version

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Install the latest development version directly from the GitHub repository.

```powershell
pip install git+https://github.com/prithvijitguha/flowrunner@main
```

--------------------------------

### Install {{ cookiecutter.project_slug }} Package

Source: https://github.com/prithvijitguha/flowrunner/blob/main/flowrunner/core/templates/{{ cookiecutter.project_name }}/readme.md

Use this command to install the {{ cookiecutter.project_slug }} package using pip. Ensure you have Python 3.8 or higher installed.

```sh
pip install {{ cookiecutter.project_slug }}
```

--------------------------------

### Install FlowRunner (Editable)

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Install FlowRunner in editable mode, allowing for quick testing of code changes.

```bash
pip install -e .
```

--------------------------------

### Run Pandas Flow Example

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pandas_example.md

This code snippet shows how to instantiate and run a Pandas flow example using the flowrunner library. It's used to execute the defined data processing steps.

```python
# we create an instance of the class and run its corresponding method
ExamplePandas().show()
```

--------------------------------

### Install Project Dependencies

Source: https://github.com/prithvijitguha/flowrunner/blob/main/tests/test_examples/test_example_notebook.ipynb

Installs pytest, ipytest, and the local Flowrunner package using pip. Ensure you are in the correct directory when installing the local package.

```python
import sys
import subprocess
import os


subprocess.check_call([sys.executable, "-m", "pip", "install", "pytest"])
subprocess.check_call([sys.executable, "-m", "pip", "install", "ipytest"])
subprocess.check_call([sys.executable, "-m", "pip", "install", "../../."])
```

--------------------------------

### Define a Basic FlowRunner Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Create a Python class inheriting from BaseFlow to define a workflow with start, step, and end decorators. This example demonstrates a simple flow with multiple paths.

```python
# -*- coding: utf-8 -*-
from flowrunner import BaseFlow, end, start, step


class ExampleFlow(BaseFlow):
    @start
    @step(next=["method2", "method3"])
    def method1(self):
        """Example of a method with a docstring which
        will become description"""
        self.a = 1

    @step(next=["method4"])
    def method2(self):
        self.a += 1

    @step(next=["method4"])
    def method3(self):
        self.a += 2

    @end
    @step
    def method4(self):
        self.a += 3
        print(self.a)


class ExampleFlow2(BaseFlow):
    @start
    @step(next=["method2", "method3"])
    def method1(self):
        self.a = 1

    @step(next=["method4"])
    def method2(self):
        self.a += 1

    @step(next=["method4"])
    def method3(self):
        self.a += 2

    @end
    @step
    def method4(self):
        self.a += 3
        print(self.a)
```

--------------------------------

### Install venv Package

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Install the `venv` package, which is used for creating isolated Python virtual environments.

```powershell
pip install venv
```

--------------------------------

### Example Flow Execution Output

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

This is the console output you can expect when running the ExampleFlow. It shows the logging messages during flow validation and execution.

```console
2023-03-08 22:29:48 LAPTOP flowrunner.system.logger[13528] INFO Found flow ExampleFlow
2023-03-08 22:29:48 LAPTOP flowrunner.system.logger[13528] DEBUG Validating flow for ExampleFlow
2023-03-08 22:29:48 LAPTOP flowrunner.system.logger[13528] WARNING Validation will raise InvalidFlowException if invalid Flow found
✅ Validated number of start nodes
✅ Validated start nodes 'next' values
✅ Validate number of middle_nodes
✅ Validated middle_nodes 'next' values
✅ Validated end nodes
✅ Validated start nodes 'next' values
2023-03-08 22:29:48 LAPTOP flowrunner.system.logger[13528] DEBUG Running flow for ExampleFlow
7
```

--------------------------------

### Define a basic Flow with @start, @step, and @end

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

This snippet demonstrates the basic structure of a FlowRunner DAG. Use @start for the entry point, @step for intermediate logic, and @end for the terminal node. Ensure methods are decorated correctly and specify the 'next' argument for @step to define the execution order.

```python
from flowrunner import BaseFlow, start, step, end

class MyFlow(BaseFlow):
    @start
    @step(next=["process"])
    def ingest(self):
        self.raw = [1, 2, 3, 4, 5]

    @step(next=["output"])
    def process(self):
        self.processed = [x * 2 for x in self.raw]

    @end
    @step
    def output(self):
        print("Result:", self.processed)
        return self.processed

# Run the flow
MyFlow().run()
# Result: [2, 4, 6, 8, 10]
```

--------------------------------

### Run FlowRunner Example

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pandas_notebook_example.html

Executes the defined FlowRunner pipeline. This is typically the final step after defining the flow.

```python
example.run()
```

--------------------------------

### flowrunner.core.decorators.start()

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.md

Decorator to mark a function as a start node in the graph.

```APIDOC
## Function: start()

### Description
Decorator to mark a function as a start node in the graph.

### Usage
```python
from flowrunner.core.decorators import start

@start()
def initial_step():
    pass
```
```

--------------------------------

### flowrunner.core.decorators.start

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Decorator to mark a function as the starting point of a flow.

```APIDOC
## flowrunner.core.decorators.start(func: Callable) -> Callable

### Description
This decorator indicates the start of a flow.
```

--------------------------------

### Define a basic flow with steps

Source: https://github.com/prithvijitguha/flowrunner/blob/main/readme.md

Define a simple flow using BaseFlow and step decorators. This example demonstrates method chaining and basic data sharing via attributes.

```python
# example.py
from flowrunner import BaseFlow, step, start, end

class ExampleFlow(BaseFlow):
    @start
    @step(next=['method2', 'method3'])
    def method1(self):
        self.a = 1

    @step(next=['method4'])
    def method2(self):
        self.a += 1

    @step(next=['method4'])
    def method3(self):
        self.a += 2

    @end
    @step
    def method4(self):
        self.a += 3
        print("output of flow is:", self.a)
```

--------------------------------

### Check Python Version

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Verify that your Python installation meets the minimum version requirement (>= 3.9).

```powershell
python --version
```

--------------------------------

### Define a Flow with Steps

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.md

Example of creating a custom Flow by inheriting from BaseFlow and using decorators to define steps and their execution order. Use this to structure your data processing pipelines.

```python
from flowrunner import BaseFlow, step, start, end

class ExampleFlow(BaseFlow):
  @start
  @step(next=['method2', 'method3'])
  def method1(self):
    self.a = 1
  
  @step(next=['method4'])
  def method2(self):
    self.a += 1
  
  @step(next=['method4'])
  def method3(self):
    self.a += 2
  
  @end
  @step
  def method4(self):
    self.a += 3
    print(self.a)
```

--------------------------------

### Install python-dateutil Package

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pyspark_example.html

Installs the 'python-dateutil' package, which may be required for date and time manipulation within the PySpark flow. This is typically run in environments where the package is not pre-installed.

```python
%pip install python-dateutil
```

--------------------------------

### Create branching logic in a Flow using @step's 'next' argument

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

This example shows how to implement fan-out and fan-in patterns in a Flow. The 'next' argument of the @step decorator can accept a list of method names to create branches, allowing for parallel execution paths before merging.

```python
from flowrunner import BaseFlow, start, step, end

class BranchingFlow(BaseFlow):
    @start
    @step(next=["branch_a", "branch_b"])  # fan-out to two branches
    def source(self):
        self.value = 10

    @step(next=["merge"])
    def branch_a(self):
        self.result_a = self.value + 5   # 15

    @step(next=["merge"])
    def branch_b(self):
        self.result_b = self.value * 2   # 20

    @end
    @step
    def merge(self):
        print("A:", self.result_a, "B:", self.result_b)

BranchingFlow().run()
# A: 15 B: 20
```

--------------------------------

### flowrunner.core.base.GraphOptions

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Configuration class for a Graph, specifying the base flow and collecting start, middle, and end nodes based on decorators.

```APIDOC
## class flowrunner.core.base.GraphOptions(base_flow: Type)

### Description
A class for all the options to be given to Graph, a collection of start, middle and end nodes. It iterates over all methods in the base_flow class and arranges them according to ‘start’, ‘step’, ‘end’ decorators.

### Attributes
* **base_flow**: A subclass of BaseFlow.
  * **Type:** Type
* **functions**: A dictionary mapping function names to their references.
* **middle_nodes**: A list of methods decorated with ‘step’ and no ‘start’ or ‘end’ decorators.
* **start**: A list of methods decorated with the ‘start’ decorator.
* **end**: A list of methods decorated with the ‘end’ decorator.
* **base_flow_instance**: An instance of base_flow used for running flows.
* **base_flow**: Type hint for the base_flow attribute.
```

--------------------------------

### Test Flowrunner Example

Source: https://github.com/prithvijitguha/flowrunner/blob/main/tests/test_examples/test_example_notebook.ipynb

Defines test functions for the ExamplePandas flow using pytest. These tests verify the execution and display methods of the flow.

```python
example_pandas = ExamplePandas()

def test_run():
    example_pandas.run()

def test_show():
    example_pandas.show()
    
def test_validate():
    example_pandas.validate()
    
def test_display():
    example_pandas.display()

```

--------------------------------

### Run Pytest

Source: https://github.com/prithvijitguha/flowrunner/blob/main/tests/test_examples/test_example_notebook.ipynb

Executes all tests in the current directory using the pytest command-line interface. This is used to verify the correctness of the flowrunner example.

```bash
!python -m pytest .
```

--------------------------------

### Example Pandas Flow Definition

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pandas_example.md

Define a FlowRunner flow that utilizes Pandas DataFrames. This includes creating initial data, transforming it by adding a snapshot date, and appending the transformed dataframes.

```python
# -*- coding: utf-8 -*-
import pandas as pd

from flowrunner import BaseFlow, end, start, step


class ExamplePandas(BaseFlow):
    @start
    @step(next=["transformation_function_1", "transformation_function_2"])
    def create_data(self):
        """
        This method we create the dataset we are going use. In real use cases,
        you'll have to read from a source (csv, parquet, etc)

        For this example we create two dataframes for students ranked by marked scored
        for when they attempted the example on 1st January 2023 and 12th March 2023

        After creating the dataset we pass it to the next methods

        - transformation_function_1
        - transformation_function_2
        """
        data1 = {"Name": ["Hermione", "Harry", "Ron"], "marks": [100, 85, 75]}

        data2 = {"Name": ["Hermione", "Ron", "Harry"], "marks": [100, 90, 80]}

        df1 = pd.DataFrame(data1, index=["rank1", "rank2", "rank3"])

        df2 = pd.DataFrame(data2, index=["rank1", "rank2", "rank3"])

        self.input_data_1 = df1
        self.input_data_2 = df2

    @step(next=["append_data"])
    def transformation_function_1(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-03-12
        """
        transformed_df = self.input_data_1
        transformed_df.insert(1, "snapshot_date", "2023-03-12")
        self.transformed_df_1 = transformed_df

    @step(next=["append_data"])
    def transformation_function_2(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-01-01
        """
        transformed_df = self.input_data_2
        transformed_df.insert(1, "snapshot_date", "2023-01-01")
        self.transformed_df_2 = transformed_df

    @step(next=["show_data"])
    def append_data(self):
        """
        Here we append the two dataframe together
        """
        self.final_df = pd.concat([self.transformed_df_1, self.transformed_df_2])

    @end
    @step
    def show_data(self):
        """
        Here we show the new final dataframe of aggregated data. However in real use cases. It would
        be more likely to write the data to some final layer/format
        """
        print(self.final_df)
        return self.final_df

```

--------------------------------

### PySpark Flow Definition

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pyspark_example.md

Define a PySpark data processing flow using FlowRunner. This example creates two datasets, adds a snapshot date to each, unions them, and displays the result.

```python
# -*- coding: utf-8 -*-
from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

from flowrunner import BaseFlow, end, start, step

spark = SparkSession.builder.getOrCreate()


class ExamplePySpark(BaseFlow):
    @start
    @step(next=["transformation_function_1", "transformation_function_2"])
    def create_data(self):
        """
        This method we create the dataset we are going use. In real use cases,
        you'll have to read from a source (csv, parquet, etc)

        For this example we create two dataframes for students ranked by marked scored
        for when they attempted the example on 1st January 2023 and 12th March 2023

        After creating the dataset we pass it to the next methods

        - transformation_function_1
        - transformation_function_2
        """

        data1 = [
            ("Hermione", 100),
            ("Harry", 85),
            ("Ron", 75),
        ]

        data2 = [
            ("Hermione", 100),
            ("Harry", 90),
            ("Ron", 80),
        ]

        columns = ["Name", "marks"]

        rdd1 = spark.sparkContext.parallelize(data1)
        rdd2 = spark.sparkContext.parallelize(data2)
        self.df1 = spark.createDataFrame(rdd1).toDF(*columns)
        self.df2 = spark.createDataFrame(rdd2).toDF(*columns)

    @step(next=["append_data"])
    def transformation_function_1(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-03-12
        """

        self.transformed_df_1 = self.df1.withColumn("snapshot_date", lit("2023-03-12"))

    @step(next=["append_data"])
    def transformation_function_2(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-01-01
        """
        self.transformed_df_2 = self.df2.withColumn("snapshot_date", lit("2023-01-01"))

    @step(next=["show_data"])
    def append_data(self):
        """
        Here we append the two dataframe together
        """
        self.final_df = self.transformed_df_1.union(self.transformed_df_2)

    @end
    @step
    def show_data(self):
        """
        Here we show the new final dataframe of aggregated data. However in real use cases.
        It would be more likely to write the data to some final layer/format
        """
        self.final_df.show()
        return self.final_df

```

--------------------------------

### Instantiate and Run Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pandas_notebook_example.html

Creates an instance of the ExamplePandas flow and then calls the 'show()' method to execute the defined data processing steps. This is how you initiate the flow's execution.

```python
example = ExamplePandas() # create an instance of your flow
```

```python
example.show()
```

--------------------------------

### Build Documentation with Sphinx

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_docs.md

Build the HTML documentation from the source files using Sphinx. Ensure you are in the 'flowrunner' directory.

```bash
sphinx-build -b html docs/source/ docs/build/html
```

--------------------------------

### Display Flow Execution Plan with BaseFlow.show()

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Use `show()` to print the ordered execution plan, including method names, docstrings, and next targets, without executing method logic. Useful for inspecting pipeline structure.

```python
from flowrunner import BaseFlow, start, step, end

class DocFlow(BaseFlow):
    @start
    @step(next=["clean"])
    def load_data(self):
        """Load raw data from source."""
        self.raw = [3, 1, 4, 1, 5]

    @step(next=["report"])
    def clean(self):
        """Remove duplicates and sort."""
        self.clean_data = sorted(set(self.raw))

    @end
    @step
    def report(self):
        """Print the final cleaned dataset."""
        print(self.clean_data)

DocFlow().show()
# load_data
# Load raw data from source.
#    Next=clean
#
# clean
# Remove duplicates and sort.
#    Next=report
#
# report
# Print the final cleaned dataset.
```

--------------------------------

### Create and Show PySpark Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pyspark_example.md

This snippet demonstrates the initial creation of a PySpark flow instance and the execution of its show method. It's a basic entry point for PySpark flows.

```python
# we create an instance of the class and run its corresponding method
ExamplePySpark().show()
```

--------------------------------

### Initialize a New Flowrunner Project

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_flowrunner_projects.md

Use this command to create a new Flowrunner project based on the default cookiecutter template. This sets up the basic directory structure and configuration files.

```bash
flowrunner init
```

--------------------------------

### Create Virtual Environment (Windows)

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Create a virtual environment named `flow_dev` and activate it on Windows.

```powershell
python -m venv flow_dev
flow_dev/scripts/activate
```

--------------------------------

### Show Flow Structure using CLI

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Display the iteration order and step descriptions (from docstrings) of your Flow using the `show` command in the CLI.

```powershell
python -m flowrunner show example.py
```

--------------------------------

### Instantiate and Run a FlowRunner PySpark Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pyspark_example.html

This code instantiates the previously defined ExampleSparkFlow and then calls the 'show()' method to execute the flow and display the resulting dataframe.

```python
# create an instance of the flow 

example_spark_flow = ExampleSparkFlow()

example_spark_flow.show()
```

--------------------------------

### Run PySpark Flow using Python Method

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pyspark_example.md

Instantiate the ExamplePySpark class and call its run method to execute the PySpark flow. This is an alternative to using the CLI.

```python
# we create an instance of the class and run its corresponding method
ExamplePySpark().run()
```

--------------------------------

### GraphValidator.validate_start_next_nodes

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Checks that the start nodes specified are valid by ensuring each start node has a defined 'next' node.

```APIDOC
## GraphValidator.validate_start_next_nodes()

### Description
Method to check that the start nodes specified are valid. We make sure that each of the nodes specified has a next.

### Returns
A tuple of (test_result, output_message) where test_result will be a True/False bool and output_message is a string value of the output message.
```

--------------------------------

### Create Virtual Environment (Unix/macOS)

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/creating_your_development_environment.md

Create a virtual environment named `flow_dev` and activate it on Unix-based systems (Linux/macOS).

```bash
python -m venv flow_dev
source flow_dev/bin/activate
```

--------------------------------

### PySpark DataFrame Integration in FlowRunner

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Integrate PySpark DataFrames within FlowRunner flows. Ensure PySpark support is installed via `pip install flowrunner[pyspark]`. Spark DataFrames are shared via `self` attributes.

```python
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit
from flowrunner import BaseFlow, start, step, end

spark = SparkSession.builder.master("local").appName("FlowRunnerSpark").getOrCreate()

class SparkETLFlow(BaseFlow):
    @start
    @step(next=["enrich"])
    def extract(self):
        """Load raw records into a Spark DataFrame."""
        data = [("Alice", 200), ("Bob", 150), ("Carol", 300)]
        self.df = spark.createDataFrame(data, ["customer", "revenue"])

    @step(next=["write"])
    def enrich(self):
        """Add a processing timestamp column."""
        self.df = self.df.withColumn("snapshot", lit("2024-01-01"))
        self.df = self.df.filter(col("revenue") > 100)

    @end
    @step
    def write(self):
        """Show the final enriched dataset."""
        self.df.show()
        return self.df

SparkETLFlow().run()
# +---------+-------+----------+
# | customer|revenue|  snapshot|
# +---------+-------+----------+
# |    Alice|    200|2024-01-01|
# |      Bob|    150|2024-01-01|
# |    Carol|    300|2024-01-01|
# +---------+-------+----------+

```

--------------------------------

### Display PySpark Flow DAG using Python Method

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/pyspark_example.md

Instantiate the ExamplePySpark class and call its display method to visualize the PySpark flow's DAG. This is useful for understanding the flow's structure.

```python
# we create an instance of the class and run its corresponding method
ExamplePySpark().display()
```

--------------------------------

### GraphValidator.validate_length_start_nodes

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Validates the number of start nodes in the graph, ensuring there is at least one.

```APIDOC
## GraphValidator.validate_length_start_nodes()

### Description
Method to validate the length of start nodes. We make sure that there is atleast one start node.

### Returns
A tuple of (test_result, output_message) where test_result will be a True/False bool and output_message is a string value of the output message.
```

--------------------------------

### Show Flow Structure using Python Method

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Display the iteration order and step descriptions of your Flow by instantiating the Flow class and calling its `show` method.

```python
# we create an instance of the class and run its corresponding method
ExampleFlow().show()
```

--------------------------------

### GraphValidator.validate_length_middle_nodes

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Validates the number of middle nodes in the graph, ensuring there is at least one start node.

```APIDOC
## GraphValidator.validate_length_middle_nodes()

### Description
Method to validate the length of middle nodes. We make sure that there is atleast one start node.

### Returns
A tuple of (test_result, output_message) where test_result will be a True/False bool and output_message is a string value of the output message.
```

--------------------------------

### Run a Flow from the Command Line

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.md

Demonstrates how to execute a defined Flow using the FlowRunner CLI. Ensure your Flow class is saved in a Python file and then run it using the 'python -m flowrunner <your_file.py>' command.

```sh
$ python -m flowrunner example.py
7
```

--------------------------------

### flowrunner.core.base.Graph

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Represents a directed acyclic graph (DAG) of nodes, organizing them into start, middle, and end stages for execution.

```APIDOC
## class flowrunner.core.base.Graph(graph_options: [GraphOptions](#flowrunner.core.base.GraphOptions))

### Description
A class containing an arranged collection of Nodes from start, middle, end in Graph.levels.

### Attributes
* **graph_options**: An instance of GraphOptions class.
  * **Type:** [flowrunner.core.base.GraphOptions](#flowrunner.core.base.GraphOptions)
* **start**: A list of start nodes, taken from GraphOptions.start. Nodes decorated with @start.
* **middle_nodes**: A list of middle nodes, taken from GraphOptions.middle_nodes. Nodes decorated with @step only.
* **end**: A list of end nodes, taken from GraphOptions.end. Nodes decorated with @end.
* **nodes**: A list of all nodes: start + middle_nodes + end.
* **node_map**: A dictionary mapping node names to node objects for later reference.
* **levels**: A list defining the iteration order for methods from start -> middle_nodes -> end.
* **graph_options**: Type hint for the graph_options attribute.
```

--------------------------------

### Utilize param_store for configurable Flow parameters

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

This snippet illustrates how to inject and access parameters within a Flow using the `param_store`. Parameters can be passed during instantiation, allowing for flexible configuration of flow behavior without modifying the code.

```python
from flowrunner import BaseFlow, start, step, end

class ParamFlow(BaseFlow):
    @start
    @step(next=["transform"])
    def extract(self):
        # Access param_store for configurable values
        multiplier = self.param_store.get("multiplier", 1)
        self.values = [i * multiplier for i in range(5)]

    @step(next=["load"])
    def transform(self):
        self.values = [v + 100 for v in self.values]

    @end
    @step
    def load(self):
        print("Final values:", self.values)
        return self.values

# Inject parameters at instantiation
flow = ParamFlow(param_store={"multiplier": 3})
flow.run()
# Final values: [100, 103, 106, 109, 112]

# Inspect stored outputs
print(flow.data_store["load"])  # [100, 103, 106, 109, 112]
```

--------------------------------

### flowrunner.core.decorators.Step

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.md

A decorator used to define a step within a flow. It can be used to mark functions as start, middle, or end nodes.

```APIDOC
## Decorator: Step

### Description
A decorator used to define a step within a flow. It can be used to mark functions as start, middle, or end nodes.

### Usage
```python
from flowrunner.core.decorators import Step

@Step()
def my_step():
    pass
```
```

--------------------------------

### Run Flow and Handle Validation Errors

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Demonstrates how to run a flow and catch potential validation errors before execution.

```python
try:
    BadFlow().run()
except InvalidFlowException as e:
    print(f"Flow refused to run: {e}")
```

--------------------------------

### GraphValidator.check_step_not_included_next

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.core.md

Validates that a step mentioned is used in the next sequence. It identifies any step that is not mentioned in a 'next' sequence and is not a start or end node.

```APIDOC
## GraphValidator.check_step_not_included_next()

### Description
Method to validate that a step mentioned is used in the next. We find any step that is not mentioned in a next that is not start or end, meaning a mid starting node.

### Returns
A tuple of (test_result, output_message) where test_result will be a True/False bool and output_message is a string value of the output message.
```

--------------------------------

### BaseFlow.show()

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Prints the ordered execution plan of the Flow to the terminal, including method names, docstrings, and their `next` targets, without executing any method logic. Useful for inspecting pipeline structure.

```APIDOC
## `BaseFlow.show()`

Prints the ordered execution plan of the Flow to the terminal — method names, docstrings, and their `next` targets — without actually running any method logic. Useful for quickly inspecting pipeline structure.

### Method

- `show()`

### Request Example

```python
from flowrunner import BaseFlow, start, step, end

class DocFlow(BaseFlow):
    @start
    @step(next=["clean"])
    def load_data(self):
        """Load raw data from source."""
        self.raw = [3, 1, 4, 1, 5]

    @step(next=["report"])
    def clean(self):
        """Remove duplicates and sort."""
        self.clean_data = sorted(set(self.raw))

    @end
    @step
    def report(self):
        """Print the final cleaned dataset."""
        print(self.clean_data)

DocFlow().show()
```

### Response

- Returns None. Prints the execution plan to the terminal.

### Response Example

```
# load_data
# Load raw data from source.
#    Next=clean
#
# clean
# Remove duplicates and sort.
#    Next=report
#
# report
# Print the final cleaned dataset.
```
```

--------------------------------

### Handling FlowRunner Exceptions

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Catch `InvalidFlowException` for structural validation errors and `CyclicFlowException` for detected cycles during DAG construction. The example demonstrates catching a missing-end-node error.

```python
from flowrunner import BaseFlow, start, step, end
from flowrunner.system.exceptions import InvalidFlowException, CyclicFlowException

# Example: catching a missing-end-node error
class BadFlow(BaseFlow):
    @start
    @step(next=["middle"])
    def begin(self):
        pass

    @step(next=["finish"])
    def middle(self):
        pass

    # 'finish' is referenced in next but never defined — invalid flow
    @step(next=["finish"])
    def finish(self):
        pass

try:
    BadFlow().validate_with_error(terminal_output=True)
except InvalidFlowException as e:
    print(f"Caught: {e}")
# ❌ ...
# Caught: Invalid Flow detected

```

--------------------------------

### Run Pre-commit Checks on a File

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_docs.md

Run pre-commit checks on a specific file to verify it meets the project's standards before committing.

```bash
pre-commit run --files modified_file
```

--------------------------------

### Define a Pandas DataFrame Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/tests/test_examples/test_example_notebook.ipynb

Defines a data processing flow using flowrunner's BaseFlow class and pandas DataFrames. This example demonstrates creating, transforming, and appending dataframes.

```python
import pandas as pd

from flowrunner import BaseFlow, end, start, step


class ExamplePandas(BaseFlow):
    @start
    @step(next=["transformation_function_1", "transformation_function_2"])
    def create_data(self):
        """
        This method we create the dataset we are going use. In real use cases,
        you'll have to read from a source (csv, parquet, etc)

        For this example we create two dataframes for students ranked by marked scored
        for when they attempted the example on 1st January 2023 and 12th March 2023

        After creating the dataset we pass it to the next methods

        - transformation_function_1
        - transformation_function_2
        """
        data1 = {"Name": ["Hermione", "Harry", "Ron"], "marks": [100, 85, 75]}

        data2 = {"Name": ["Hermione", "Ron", "Harry"], "marks": [100, 90, 80]}

        df1 = pd.DataFrame(data1, index=["rank1", "rank2", "rank3"])

        df2 = pd.DataFrame(data2, index=["rank1", "rank2", "rank3"])

        self.input_data_1 = df1
        self.input_data_2 = df2

    @step(next=["append_data"])
    def transformation_function_1(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-03-12
        """
        transformed_df = self.input_data_1
        transformed_df.insert(1, "snapshot_date", "2023-03-12")
        self.transformed_df_1 = transformed_df

    @step(next=["append_data"])
    def transformation_function_2(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-01-01
        """
        transformed_df = self.input_data_2
        transformed_df.insert(1, "snapshot_date", "2023-01-01")
        self.transformed_df_2 = transformed_df

    @step(next=["show_data"])
    def append_data(self):
        """
        Here we append the two dataframe together
        """
        self.final_df = pd.concat([self.transformed_df_1, self.transformed_df_2])

    @end
    @step
    def show_data(self):
        """
        Here we show the new final dataframe of aggregated data. However in real use cases. It would
        be more likely to write the data to some final layer/format
        """
        print(self.final_df)
        return self.final_df

```

--------------------------------

### Mermaid Initialization with Custom Theme

Source: https://github.com/prithvijitguha/flowrunner/blob/main/flowrunner/core/templates/base.html

Initializes the Mermaid JavaScript library with a custom theme and theme variables. This is used to style Mermaid diagrams globally.

```javascript
import mermaid from 'https://cdn.jsdelivr.net/npm/mermaid@10/dist/mermaid.esm.min.mjs'; // We use the mermaid API to change the theme page wide mermaid.initialize({
  theme: 'base',
  themeVariables: {
    'primaryColor': '#5A5A5A',
    'primaryTextColor': 'white',
    'primaryBorderColor': '#5A5A5A',
    'lineColor': '#F8B229',
    'secondaryColor': '#006100',
    'tertiaryColor': '#D3D3D3',
    'tertiaryTextColor': 'black',
  }
})
```

--------------------------------

### Define ExamplePandas Flow

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/_static/flowrunner_pandas_notebook_example.html

Defines a Flowrunner flow class named ExamplePandas. It includes methods for creating initial dataframes, transforming them, appending them, and displaying the final result. Requires pandas to be imported.

```python
from flowrunner import BaseFlow, end, start, step
import pandas as pd

class ExamplePandas(BaseFlow):
    @start
    @step(next=["transformation_function_1", "transformation_function_2"])
    def create_data(self):
        """
        This method we create the dataset we are going use. In real use cases,
        you'll have to read from a source (csv, parquet, etc)

        For this example we create two dataframes for students ranked by marked scored
        for when they attempted the example on 1st January 2023 and 12th March 2023

        After creating the dataset we pass it to the next methods

        - transformation_function_1
        - transformation_function_2
        """
        data1 = {"Name": ["Hermione", "Harry", "Ron"], "marks": [100, 85, 75]}

        data2 = {"Name": ["Hermione", "Ron", "Harry"], "marks": [100, 90, 80]}

        df1 = pd.DataFrame(data1, index=["rank1", "rank2", "rank3"])

        df2 = pd.DataFrame(data2, index=["rank1", "rank2", "rank3"])

        self.input_data_1 = df1
        self.input_data_2 = df2

    @step(next=["append_data"])
    def transformation_function_1(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-03-12
        """
        transformed_df = self.input_data_1
        transformed_df.insert(1, "snapshot_date", "2023-03-12")
        self.transformed_df_1 = transformed_df

    @step(next=["append_data"])
    def transformation_function_2(self):
        """
        Here we add a snapshot_date to the input dataframe of 2023-01-01
        """
        transformed_df = self.input_data_2
        transformed_df.insert(1, "snapshot_date", "2023-01-01")
        self.transformed_df_2 = transformed_df

    @step(next=["show_data"])
    def append_data(self):
        """
        Here we append the two dataframe together
        """
        self.final_df = pd.concat([self.transformed_df_1, self.transformed_df_2])

    @end
    @step
    def show_data(self):
        """
        Here we show the new final dataframe of aggregated data. However in real use cases. It would
        be more likely to write the data to some final layer/format
        """
        print(self.final_df)
        return self.final_df
```

--------------------------------

### Create and Checkout a New Branch

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_docs.md

Create a new branch for your changes to keep the master branch clean. This can be done in one step.

```bash
git checkout -b new_doc_change
```

--------------------------------

### Display Flow DAG using Python Method

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Visualize the Flow's DAG by instantiating the Flow class and calling its `display` method.

```python
# we create an instance of the class and run its corresponding method
ExampleFlow().display()
```

--------------------------------

### Run a flow from the command line

Source: https://github.com/prithvijitguha/flowrunner/blob/main/readme.md

Execute a defined flow using the flowrunner CLI. Ensure the flow definition is in a Python file.

```console
$ python -m flowrunner run example.py
```

--------------------------------

### Generate HTML DAG with BaseFlow.dag()

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Use `dag()` to render the Flow as a Mermaid.js-powered HTML flowchart. Pass `save_file=True` to write to disk or `path='output/'` to specify a directory. Optionally set `description=False` to omit docstring annotations.

```python
from flowrunner import BaseFlow, start, step, end

class PipelineFlow(BaseFlow):
    @start
    @step(next=["validate_data"])
    def ingest(self):
        """Read data from CSV source."""
        self.df = [{"id": 1, "val": 10}, {"id": 2, "val": 20}]

    @step(next=["write"])
    def validate_data(self):
        """Check for nulls and type errors."""
        assert all("id" in row for row in self.df)

    @end
    @step
    def write(self):
        """Write validated data to output."""
        print("Written:", len(self.df), "rows")

flow = PipelineFlow()

# Get HTML content as string
html_content = flow.dag(description=True)

# Save to a specific directory
flow.dag(save_file=True, path="output/", description=True)
# Creates: output/pipelineflow.html

# Save to current directory (default filename: pipelineflow.html)
flow.dag(save_file=True)
```

--------------------------------

### Display Flow DAG using CLI

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Visualize the Directed Acyclic Graph (DAG) of your Flow using the `display` command in the CLI. This can be used in notebooks or saved to a file.

```powershell
python -m flowrunner display example.py
```

--------------------------------

### Inline Notebook DAG Visualization with BaseFlow.display()

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Use `display()` in Jupyter notebooks to render the Flow DAG as an inline image using the Mermaid.ink API. This method does not save a file. Docstring annotations can be disabled.

```python
# In a Jupyter notebook cell:
from flowrunner import BaseFlow, start, step, end

class NotebookFlow(BaseFlow):
    @start
    @step(next=["transform"])
    def extract(self):
        """Extract records from database."""
        self.records = list(range(100))

    @step(next=["load"])
    def transform(self):
        """Normalize and enrich records."""
        self.records = [r * 1.5 for r in self.records]

    @end
    @step
    def load(self):
        """Load records into data warehouse."""
        return self.records

# Displays an interactive flowchart inline in the notebook
NotebookFlow().display()

# Disable docstring annotations in the diagram
NotebookFlow().display(description=False)
```

--------------------------------

### Run Flow using CLI

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/getting_started.md

Execute your defined Flow using the `run` command in the FlowRunner CLI.

```powershell
python -m flowrunner run example.py
```

--------------------------------

### Stage and Commit Changes

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_code.md

Stage modified files and commit them with a descriptive message. Ensure pre-commit hooks pass before committing.

```bash
git add modified_file
```

```bash
pre-commit install
```

```bash
pre-commit run --files modified_file
```

```bash
git commit -m "modified file modified_file"
```

--------------------------------

### Clone Fork and Add Upstream Remote

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_code.md

Clone your forked repository and add the upstream remote to fetch changes from the original project.

```bash
git clone https://github.com/prithvijitguha/flowrunner.git flowrunner-username
cd flowrunner-username
git remote add upstream https://github.com/prithvijitguha/flowrunner.git
```

--------------------------------

### CLI - `python -m flowrunner run`

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

The command-line interface for FlowRunner allows running, validating, showing, and displaying flows defined in Python files.

```APIDOC
## CLI — `python -m flowrunner run`

Runs all `BaseFlow` subclasses found in a Python file. The CLI automatically discovers every subclass of `BaseFlow` in the target file and executes them in order.

### Commands

- `python -m flowrunner run <file.py>`: Executes all `BaseFlow` subclasses in the specified file.
- `python -m flowrunner validate <file.py>`: Validates flows in the file, printing check results (✅/❌).
- `python -m flowrunner show <file.py>`: Displays the execution order of flows without running them.
- `python -m flowrunner display <file.py>`: Generates and saves an HTML DAG for all flows in the file.

### Request Example

```bash
# example_flow.py contains one or more BaseFlow subclasses

# Run all flows in a file
python -m flowrunner run example_flow.py

# Validate a flow file (prints ✅/❌ per check)
python -m flowrunner validate example_flow.py

# Show execution order without running
python -m flowrunner show example_flow.py

# Generate and save HTML DAG for all flows in a file
python -m flowrunner display example_flow.py
```
```

--------------------------------

### Create and Checkout a New Branch

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/contributing_guide_code.md

Create a new branch for your feature or bug fix. This isolates your changes from the main development line.

```bash
git branch shiny-new-feature
git checkout shiny-new-feature
```

```bash
git checkout -b shiny-new-feature
```

--------------------------------

### flowrunner.core.base.GraphOptions

Source: https://github.com/prithvijitguha/flowrunner/blob/main/docs/source/flowrunner.md

Configuration options for a Graph, defining its structure and behavior.

```APIDOC
## Class: GraphOptions

### Description
Configuration options for a Graph, defining its structure and behavior.

### Attributes
- `base_flow` (BaseFlow): The base flow class used for execution.
- `functions` (dict): A dictionary mapping function names to callable functions.
- `middle_nodes` (list[str]): Names of the middle nodes.
- `start` (str): Name of the start node.
- `end` (str): Name of the end node.
- `base_flow_instance` (BaseFlow): An instance of the base flow class.
```

--------------------------------

### Generate DAGs with FlowRunner CLI

Source: https://context7.com/prithvijitguha/flowrunner/llms.txt

Use the FlowRunner CLI to display a DAG for a single flow file or all flow files in a directory. Scaffold new projects using the cookiecutter template.

```bash
python -m flowrunner display example_flow.py --path output/dags/
```

```bash
python -m flowrunner display_dir ./my_flows/
```

```bash
python -m flowrunner init
```

```bash
python -m flowrunner init --output-dir ./projects/
```

--------------------------------

### Show flow structure

Source: https://github.com/prithvijitguha/flowrunner/blob/main/readme.md

Display the structure of the flow, including nodes and their connections, using the show() method. This provides a textual representation of the DAG.

```python
ExampleFlow().show()
```

--------------------------------

### Display flow as a DAG

Source: https://github.com/prithvijitguha/flowrunner/blob/main/readme.md

Visualize the defined flow as a Directed Acyclic Graph (DAG) using the display() method. This helps in understanding the flow's structure.

```python
ExampleFlow().display()
```