pipeflow-lite 1.0.0

A flexible and powerful data processing pipeline library

Overview

Pipeflow-lite is a flexible and powerful data processing pipeline framework designed to simplify the flow of data from sources to destinations such as databases, file systems, APIs, and more. It leverages a plugin-based architecture that allows users to easily extend functionality as needed while keeping the core logic simple and efficient.

Installation

Pipeflow-lite can be installed via Python's package management tool pip (assuming you have packaged the project as an installable Python package):

pip install pipeflow-lite

Alternatively, if you are developing locally, you can use the project's source code directly:

python setup.py develop  

Quick Start

1. Configure the Environment

Ensure your Python environment has Pipeflow-lite and its dependencies installed.

2. Define Actions

Create your action files

from pipeflow.core.pipeflow_action import PipeflowAction
from types import MappingProxyType
from typing import Any


class AAction(PipeflowAction):
    async def execute(self, params: MappingProxyType) -> Any:
        # Implement your data processing logic  
        # Access and modify data in the context  
        pass


class BAction(PipeflowAction):
    async def execute(self, params: MappingProxyType) -> Any:
        # ...
        pass


class CAction(PipeflowAction):
    def upstream(self):
        return [AAction, BAction]

    async def execute(self, params: MappingProxyType) -> Any:
        # ... 
        pass
# class D and E...

3. Define Context

Create your context file, e.g., one_context.py:

from pipeflow.core.pipeflow_context import PipeflowContext, load_from_directory


class OneContext(PipeflowContext):
    def __init__(self):
        # Load from path
        # Relative to the project root path, or just use the absolute path
        action_classes = load_from_directory("./example")

        # # Or load by type
        # from example.actions.a_action import AAction
        # from example.actions.b_action import BAction
        # from example.actions.c_action import CAction
        # from example.actions.d_action import DAction
        # from example.actions.e_action import EAction
        # action_classes = [
        #     AAction,
        #     BAction,
        #     CAction,
        #     DAction,
        #     EAction
        # ]

        super().__init__(action_classes)

4. Run the Pipeline

Write code in example/main.py to create and run your pipeline:

from example.one_context import OneContext

# Create a context instance  
pipeline = OneContext()
result = pipeline.execute(initial_params={"key1": "value1"})

5. Visualize the Flow

Use view.py to generate and view the flowchart of your action node sequences:

python view.py