蟒蛇管道

该库允许用户使用一系列装饰器轻松包装函数以形成执行管道。这在需要清理输入和需要以系统方式处理输出的情况下很有用。

安装

pip install execution-pipeline

用法

管道由四个可选段组成

预

pre执行段允许您修改传递给修饰函数的任何输入参数。传递给pre段的任何函数将始终首先执行。

from pipeline import execution_pipeline

def do_thing_before(params):
    params['arg1'] = 'okay'
    return params


@execution_pipeline(pre=[do_thing_before])
def do_thing(arg1=5):
    return arg1*10
 


do_thing()
# okayokayokayokayokayokayokayokayokayokay

邮政

执行段允许您在post装饰函数已经运行后修改或处理结果。

def do_thing_after(response):
    response['added'] = 'yup'
    return response
        
@execution_pipeline(post=[do_thing_after])
def do_thing(**kwargs):
    return {**kwargs}  # just make a new dictionary using the passed keyword arguments
    
    
do_thing(apples=2, oranges=3, bananas=0)
 # {'apples': 2, 'oranges': 3, 'bananas': 0, 'added': 'yup'}

就像其他部分一样，您可以根据需要使用任意数量的后期处理功能；它们将按照通过的顺序执行。

def do_another_thing_after(response):
    assert response['added'] == 'yup' # the one that is first in the pipeline happens first.
    response['also_added'] = 'also yup'
    return response
    
    
@execution_pipeline(post=[do_thing_after, do_another_thing_after])
def do_thing(**kwargs):
    return {**kwargs}
    
 
do_thing()
# {'apples': 2, 'oranges': 3, 'bananas': 0, 'added': 'yup', 'also_added': 'also yup'}

错误

执行段允许您将error错误处理传递给记录、修改、吸收等任何由包装函数引发的异常。

class MyException(BaseException):
    pass
    

def handle_this_error(e, response=None):
    print(f"oh no, Bob! {e}")
    return "Don't worry, we handled a TypeError."


def handle_that_error(e, response=None):
    print(f"oh no, Bob! {e}")
    return "Don't worry, we handled MyException."
    
def handle_other_errors(e, response=None):
    print(f"? {e}")
    return "Other errors?"
    
error_handlers = [
    {"exception_class": TypeError, "handler": handle_this_error},
    {"exception_class": MyException, "handler": handle_that_error},
    {"exception_classes": (Exception, BaseException), "handler": handle_other_errors},
]


@execution_pipeline(pre=[do_thing_before], post=[do_thing_after], error=error_handlers)
def fun_boys(arg1, arg4, arg2, arg3, thing=None):
    raise MyException('Something went wrong!')
    

result = fun_boys(1, 2, 3, 4, 5)
# oh no, Bob! Something went wrong!

print(result) 
# Don't worry, we handled MyException.

如果您愿意，还可以使用类实例而不是字典来定义错误处理程序。

class ErrorHandler:
    def __init__(self, exception_class, handler):
        self.exception_class = exception_class
        self.handler = handler
        

error_handlers = [
    ErrorHandler(TypeError, handle_this_error),
    ErrorHandler(MyException, handle_that_error),
]

缓存

执行段将cache记录所有参数（在pre段之前和之后）并存储结果（在post和error段之后）。

from pipeline.cache.mock import MockCache
# MockCache is basically just a dict() with some expiration convenience methods.
mock_cache = MockCache()

changing_value = 0

@execution_pipeline(cache=mock_cache)
def fun_boys(arg1, arg4, arg2, arg3, thing=None):
    return changing_value
    

    
fun_boys(1, 2, 3, 4, 5)
# 0

changing_value = 100


fun_boys(1, 2, 3, 4, 5)
# 0 # ignores the changes ( ¯\_(ツ)_/¯ that's caching! )

支持的缓存后端

注意：如果未安装适当的后端，它们将在运行时替换为MockCache实例。这是为了提高流水线代码的可移植性。

雷迪斯

pip install redis

然后与上面相同，除了

from pipeline.cache.redis import RedisCache
redis = RedisCache(host='localhost', port=6379) # defaults

内存缓存

pip install memcached

然后与上面相同，除了

from pipeline.cache.mem_cache import MemCache 
mem_cache = MemCache(host='localhost', port=11211) # defaults