It appears you want to inject some variables to the function scope every time it is invoked. One way to do this would be to temporarily insert data to the function's globals scope(find_videos_in_polygon.__globals__) and later clean it up.
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
Test code:
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)
Output:
1 2 3 4 10
10000
1 2 3 4 15
10000
...
UnboundLocalError: local variable 'eggs' referenced before assignment
But of course, doing something like this would make the code less-readable and hard to debug. Hence it is better to be explicit and do things that are much easier to understand and test. I would recommend adding a helper function that can parse and return a dict with all keyword arguments expected by 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)
Demo:
>>> 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'
