python - 如何在 dag 模板中的任务函数上使用装饰器

Question

我有将近 20 个函数，我为它们编写了一个装饰器，而不是在每个函数中注入相同的代码。所以我的装饰器应该检查每个函数是否返回 500 响应状态代码然后引发异常。谁能给我更清楚？

这是我写的装饰器

def check_tasks_failed(a_func):
    """
     Wrapper function to check whether a task_failed
    :param task_id:
    :param kwargs:
    :param kwargs:
    :param a_func:
    :return:

    """
    def wrapper(**kwargs):
        task_id_resp_status_code = a_func(**kwargs).status_code
        if task_id_resp_status_code in [500]: # Check here what are the criterion for the task to fail
           raise ValueError('invalid return status 500')
    return wrapper

这是需要装饰器作为示例的功能。

@check_tasks_failed
def get_naming_(**kwargs) -> requests:
    """Get job from custom_etl."""
    try:
        naming_res = get_custom_etl_job(VALIDATED_CONTAINER,
                                                 VALIDATED_CONTAINER,
                                                 VALIDATED_CONTAINER,
                                                 **kwargs)
    except Exception as e:
        return {'success':False , 'reason': f'{e}'}

我的预期结果应该是当 get_naming_ 失败以下异常时引发的异常raise ValueError('invalid return status 500')

标签： pythondecorator