python - 如何创建一个带有参数的装饰器来解析请求？

Question

我有以下方法:

@try_except_decorator
@json_response_decorator
async def find_videos_in_polygon(request):
    points = request.query['points']
    end_date = request.query['end_date']
    start_date = request.query['start_date']
    ding_kind = request.query.get('ding_kind', 'all')

    search_service = request.app['search_service']
    data = await search_service.search_in_polygon(points, start_date,
                                              end_date,
                                              ding_kind=ding_kind,
                                              doorbot_kind=doorbot_kind)

    return {'videos': data}

如何创建一个解析请求的装饰器？我想要这样的东西:

@try_except_decorator
@json_response_decorator
@parse_request('points', 'end_date', 'start_date', ('ding_kind', 'all'))
async def find_videos_in_polygon(request):
    search_service = request.app['search_service']
    data = await search_service.search_in_polygon(???)

    return {'videos': data}

此外，我不想更改 find_videos_in_polygon(request) 的签名

Answer 1

您似乎希望在每次调用函数范围时都将一些变量注入(inject)函数范围。一种方法是将数据临时插入函数的全局范围 (find_videos_in_polygon.__globals__)，然后再清理它。

def inject_variables(func, _new_values=None):
    for k, v in _new_values.items():
        func.__globals__[k] = v


def cleanup_variables(func, _default=None, _new_values=None, _old_values=None):
    """
    Reset function's global scope with data in `_old_value`.
    If a particular key's value is sentinel then it means the
    key didn't exist and we can remove it.
    """
    for k, v in _old_values.items():
        old_value = _old_values[k]
        if old_value is _default:
            del func.__globals__[k]
        else:
            func.__globals__[k] = old_value


def parse_request(*names):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            request = args[0]
            new_values = {}
            current_global_values = {}
            sentinel = object()

            for var in names:
                name = var
                if isinstance(var, tuple):
                    name, value = var
                    new_values[name] = request.query.get(name, value)
                else:
                    try:
                        new_values[name] = request.query[name]
                    except KeyError:
                        raise UnboundLocalError("local variable '{name}' referenced before assignment".format(
                            name=name
                        ))
                current_global_values[name] = func.__globals__.get(name, sentinel)

            inject_variables(func, _new_values=new_values)
            val = await func(*args, **kwargs)
            cleanup_variables(func, _default=sentinel, _new_values=new_values, _old_values=current_global_values)
            return val
        return wrapper
    return decorator

测试代码:

from asyncio import get_event_loop


missing = 10000

@parse_request('foo', 'bar', 'spam', 'eggs', ('missing', '10'))
async def func(request):
    print(foo, bar, spam, eggs, missing)


class Request:
    pass


loop = get_event_loop()

Request.query = {'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4}
loop.run_until_complete(func(Request))
print(missing)

Request.query = {'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4, 'missing': 15}
loop.run_until_complete(func(Request))
print(missing)

# Missing required key 'eggs', should raise an error
Request.query = {'foo': 1, 'bar': 2, 'spam': 3}
loop.run_until_complete(func(Request))
print(missing)

输出:

1 2 3 4 10
10000
1 2 3 4 15
10000
... 
UnboundLocalError: local variable 'eggs' referenced before assignment

---

但是，当然，这样做会降低代码的可读性和调试难度。因此，最好是明确地做一些更容易理解和测试的事情。我建议添加一个辅助函数，该函数可以解析并返回包含 search_in_polygon 所需的所有关键字参数的字典。

def parse_request(request, *names):
    data = {}
    for var in names:
        name = var
        if isinstance(var, tuple):
            name, value = var
            data[name] = request.query.get(name, value)
        else:
            data[name] = request.query[name]
    return data


async def func(request):
    request_data = parse_request(request, 'foo', 'bar', 'spam', 'eggs', ('missing', '10'))
    print(request_data)
    # data = await search_in_polygon(**request_data)

演示:

>>> Request.query = {'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4}
>>> loop.run_until_complete(func(Request))
{'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4, 'missing': '10'}
>>> Request.query = {'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4, 'missing': 15}
>>> loop.run_until_complete(func(Request))
{'foo': 1, 'bar': 2, 'spam': 3, 'eggs': 4, 'missing': 15}
>>> Request.query = {'foo': 1, 'bar': 2, 'spam': 3}
>>> loop.run_until_complete(func(Request))
---------------------------------------------------------------------------
KeyError: 'eggs'