如何使用 Python 3.x 将回调及其参数从包装函数传递给装饰器？

Question

我正在围绕 REST API 编写通用包装器。我有几个类似下面的函数，负责从用户的电子邮件地址中检索用户。感兴趣的部分是如何处理响应，基于预期状态代码列表（除了 HTTP 200）和与每个预期状态代码关联的回调：

import requests

def get_user_from_email(email):
    response = requests.get('http://example.com/api/v1/users/email:%s' % email)

    # define callbacks
    def return_as_json(response):
        print('Found user with email [%s].' % email)
        return response.json()

    def user_with_email_does_not_exist(response):
        print('Could not find any user with email [%s]. Returning `None`.' % email),
        return None

    expected_status_codes_and_callbacks = {
        requests.codes.ok: return_as_json,  # HTTP 200 == success
        404: user_with_email_does_not_exist,
    }
    if response.status_code in expected_status_codes_and_callbacks:
        callback = expected_status_codes_and_callbacks[response.status_code]
        return callback(response)
    else:
        response.raise_for_status()


john_doe = get_user_from_email('john.doe@company.com')
print(john_doe is not None)  # True

unregistered_user = get_user_from_email('unregistered.user@company.com')
print(unregistered_user is None)  # True

上面的代码运行良好，所以我想重构和概括响应处理部分。我很想得到以下代码：

@process_response({requests.codes.ok: return_as_json, 404: user_with_email_does_not_exist})
def get_user_from_email(email):
    # define callbacks
    def return_as_json(response):
        print('Found user with email [%s].' % email)
        return response.json()

    def user_with_email_does_not_exist(response):
        print('Could not find any user with email [%s]. Returning `None`.' % email),
        return None

    return requests.get('https://example.com/api/v1/users/email:%s' % email)

process_response 装饰器定义为：

import functools

def process_response(extra_response_codes_and_callbacks=None):

    def actual_decorator(f):

        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            response = f(*args, **kwargs)

            if response.status_code in expected_status_codes_and_callbacks:
                action_to_perform = expected_status_codes_and_callbacks[response.status_code]
                return action_to_perform(response)
            else:
                response.raise_for_status()  # raise exception on unexpected status code

        return wrapper

    return actual_decorator

我的问题是装饰器抱怨无法访问return_as_json 和user_with_email_does_not_exist，因为这些回调定义在内部包装函数中。

如果我决定将回调移动到包装函数之外，例如与装饰器本身在同一级别，则回调无法访问包装函数内的 response 和 email 变量。

# does not work either, as response and email are not visible from the callbacks
def return_as_json(response):
    print('Found user with email [%s].' % email)
    return response.json()

def user_with_email_does_not_exist(response):
    print('Could not find any user with email [%s]. Returning `None`.' % email),
    return None

@process_response({requests.codes.ok: return_as_json, 404: user_with_email_does_not_exist})
def get_user_from_email(email):
    return requests.get('https://example.com/api/v1/users/email:%s' % email)

这里的正确方法是什么？我发现装饰器语法非常简洁，但我不知道如何将所需的部分传递给它（回调本身或它们的输入参数，如 response 和 email）。

Answer 1

这是一种在类方法上使用函数成员数据以便将响应函数映射到适当回调的方法。这对我来说似乎是最简洁的语法，但仍然有一个类变成了一个函数（如果需要，可以很容易地避免）。

# dummy response object
from collections import namedtuple
Response = namedtuple('Response', 'data status_code error')


def callback(status_code):
    def method(f):
        f.status_code = status_code
        return staticmethod(f)
    return method


def request(f):
    f.request = True
    return staticmethod(f)


def callback_redirect(cls):
    __callback_map = {}
    for attribute_name in dir(cls):
        attribute = getattr(cls, attribute_name)

        status_code = getattr(attribute, 'status_code', '')
        if status_code:
            __callback_map[status_code] = attribute

        if getattr(attribute, 'request', False):
            __request = attribute

    def call_wrapper(*args, **kwargs):
        response = __request(*args, **kwargs)
        callback = __callback_map.get(response.status_code)
        if callback is not None:
            return callback(response)
        else:
            return response.error

    return call_wrapper


@callback_redirect
class get_user_from_email:

    @callback('200')
    def json(response):
        return 'json response: {}'.format(response.data)

    @callback('404')
    def does_not_exist(response):
        return 'does not exist'

    @request
    def request(email):
        response = Response(email, '200', 'exception')
        return response


print get_user_from_email('generic@email.com')
# json response: generic@email.com

Answer 2

正如我在其他答案的 cmets 中提到的，这是一个使用类和装饰器的答案。这有点违反直觉，因为get_user_from_email 被声明为一个类，但在装饰后最终成为一个函数。但是，它确实具有所需的语法，所以这是一个优点。也许这可能是一个更清洁的解决方案的起点。

# dummy response object
from collections import namedtuple
Response = namedtuple('Response', 'data status_code error')

def callback_mapper(callback_map):

    def actual_function(cls):

        def wrapper(*args, **kwargs):
            request = getattr(cls, 'request')
            response = request(*args, **kwargs)

            callback_name = callback_map.get(response.status_code)
            if callback_name is not None:
                callback_function = getattr(cls, callback_name)
                return callback_function(response)

            else:
                return response.error

        return wrapper

    return actual_function


@callback_mapper({'200': 'json', '404': 'does_not_exist'})
class get_user_from_email:

    @staticmethod
    def json(response):
        return 'json response: {}'.format(response.data)

    @staticmethod
    def does_not_exist(response):
        return 'does not exist'

    @staticmethod
    def request(email):
        response = Response('response data', '200', 'exception')
        return response

print get_user_from_email('blah')
# json response: response data

Answer 3

您可以将装饰器键转换为字符串，然后从通过f.func_code.co_consts 传递给装饰器的外部函数中提取内部函数。不要这样做。

import functools, new
from types import CodeType

def decorator(callback_dict=None):

    def actual_decorator(f):
        code_dict = {c.co_name: c for c in f.func_code.co_consts if type(c) is CodeType}

        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            main_return = f(*args, **kwargs)

            if main_return['callback'] in callback_dict:
                callback_string = callback_dict[main_return['callback']]
                callback = new.function(code_dict[callback_string], {})
                return callback(main_return)

        return wrapper

    return actual_decorator


@decorator({'key_a': 'function_a'})
def main_function(callback):

    def function_a(callback_object):
        for k, v in callback_object.items():
            if k != 'callback':
                print '{}: {}'.format(k, v)

    return {'callback': callback, 'key_1': 'value_1', 'key_2': 'value_2'}


main_function('key_a')
# key_1: value_1
# key_2: value_2

你可以使用类吗？如果您可以使用类，解决方案是立竿见影的。

Answer 4

您可以将外部函数的函数参数传递给处理程序：

def return_as_json(response, email=None):  # email param
    print('Found user with email [%s].' % email)
    return response.json()

@process_response({requests.codes.ok: return_as_json, 404: ...})
def get_user_from_email(email):
    return requests.get('...: %s' % email)

# in decorator
     # email param will be passed to return_as_json
     return action_to_perform(response, *args, **kwargs)