How can I pass callbacks and their arguments from a wrapped function to a decorator using Python 3.x?
I am writing a general wrapper around the REST API. I have several functions like the ones below, responsible for fetching a user from their email address. Part of the interest lies in how the response is handled based on the list of expected status codes (other than HTTP 200
) and the callbacks associated with each expected status code:
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
The code above works well, so I want to refactor and generalize some of the response handling. I would like to receive the following code:
@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)
with a decorator process_response
defined as:
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
My problem is that the decorator is complaining about not having access to return_as_json
and user_with_email_does_not_exist
because these callbacks are defined inside the wrapped function.
If I decide to move the callbacks outside of the wrapped function, for example at the same level as the decorator itself, then the callbacks don't have access to the response and email variables inside the wrapped function.
# 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)
What's the correct approach? I find the decorator syntax very clean, but I can't figure out how to pass the necessary parts to it (either the callbacks themselves, or their input arguments like response
and email
).
source to share
Here's an approach that uses element member data for class methods to map a response function to an appropriate callback. This seems to me the cleanest syntax, but I still have a class that turns into a function (which could be easily avoided if desired).
# 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
source to share
You can convert decorator keys to strings and then infer the inner functions from the outer function passed to the decorator via f.func_code.co_consts
. Don't do it this way.
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
Can you use classes? This decision is immediate if you can use the class.
source to share
As mentioned in the comments to my other answer, here is an answer that uses classes and decorators. This is a bit counterintuitive because it is get_user_from_email
declared as a class but ends up as a function after decoration. It does have the desired syntax, however, so a plus. Perhaps this could be the starting point for a cleaner solution.
# 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
source to share
You can pass functional parameters of an external function to handlers:
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)
source to share