Skip to main content

A simple Pipeline creator

Project description



A simple pip install will do :

python -m pip install DemPipe


from DemPipe import PipeExecutor, Action

with PipeExecutor() as pipe:
    actions = [Action(lambda x: x**2, 2),  # returns 4
               Action(lambda x: x+3, ctx_in='last_value', ctx_out='my_result_name'),  # returns 4+3
               Action(lambda x: x*2, ctx_in='my_result_name')  # returns 2*7
    result = pipe.execute(actions)
    print('result:', result)  # 14
    print('result (from context):', pipe.context['last_value'])  # 14
    print('my_result_name:', pipe.context['my_result_name'])  # 7

You can also use a Trigger Action for conditional Actions in the pipeline !

from DemPipe import PipeExecutor, Action, Trigger

with PipeExecutor() as pipe:
    actions = [Action(lambda x: x**2, 2),  # returns 4
               Trigger(lambda x: x==3,   # returns False -> executes the second action
                       Action(lambda x: x+3, ctx_in='last_value'),  # ignored
                       Action(lambda x: x+7, ctx_in='last_value'),  # returns 11
    result = pipe.execute(actions)
    print('result:', result)  # 11

Error Handling

In case an error occurs, you can configure an automatic e-mail to be sent with the traceback and the error message :

  • Create a config file containing your credentials :
  "mail": {
    "mail_server": "",
	"mail_port": 587,
	"mail_user": "${os_environ[user]}",
	"mail_password": "${os_environ[password]}",
	"mail_use_tls": true,
	"mail_default_receiver": "${os_environ[receiver]}"
  "pipe_name": "My Pipe"
  • Specify config_file to the PipeExecutor
from DemPipe import PipeExecutor, Action

with PipeExecutor(config_file='ConfigsFolder.MyConfig') as pipe:
    pipe.execute(Action(lambda x: x/0, 2))  # raises ZeroDivisionException

Other Types of actions

In addition to Trigger and Action there exist other kinds of actions :

  • ContextSetter: Makes it easier to set context values :

    actions = [ContextSetter(var1=5, var2="test string"),
               Action(lambda x: x**2, ctx_in='var1')]
    print(pipe.execute(actions))  # 25
    actions = [ContextSetter(lambda c: {'var1': c['var2'] * 3}, var2=4), 
               Action(lambda x: x**2, ctx_in='variable1')]
    print(pipe.execute(actions))  # 144
  • Procedure: Same as an action but doesn't update the current context with its return value :

    actions = [ContextSetter(last_value=3),
               Procedure(lambda x: x**2, ctx_in='last_value'),
               Action(lambda x: x + 2, ctx_in='last_value')]
    print(pipe.execute(actions))  # 5

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for DemPipe, version 0.0.2
Filename, size File type Python version Upload date Hashes
Filename, size DemPipe-0.0.2-py3-none-any.whl (11.6 kB) File type Wheel Python version py3 Upload date Hashes View
Filename, size DemPipe-0.0.2.tar.gz (6.3 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page