def api_error_handle(func):
def func_wrapper(f, *args, **kwargs):
try:
return f(*args, **kwargs)
except RateLimitError as e:
raise APIQuotaError(str(e))
except TweepError as e:
if e.api_code == TWITTER_AUTOMATED_REQUEST_ERROR:
raise AutomatedRequestError
elif e.api_code == TWITTER_OVER_CAPACITY_ERROR:
raise OverCapacityError
elif e.api_code in [TWITTER_CHARACTER_LIMIT_ERROR_1, TWITTER_CHARACTER_LIMIT_ERROR_2]:
raise CharacterLimitError
elif e.api_code == TWITTER_DAILY_STATUS_UPDATE_LIMIT_ERROR:
raise DailyStatusUpdateError
elif e.api_code == TWITTER_STATUS_DUPLICATE_ERROR:
raise StatusDuplicateError
else:
raise
return decorate(func, func_wrapper)
python类decorate()的实例源码
def api_version(created_ver, last_changed_ver):
"""Version check decorator. Currently only checks Bigger Than."""
def api_min_version_decorator(function):
def wrapper(function, self, *args, **kwargs):
if not self.version_check_mode == "none":
if self.version_check_mode == "created":
version = created_ver
else:
version = last_changed_ver
major, minor, patch = parse_version_string(version)
if major > self.mastodon_major:
raise MastodonVersionError("Version check failed (Need version " + version + ")")
elif major == self.mastodon_major and minor > self.mastodon_minor:
raise MastodonVersionError("Version check failed (Need version " + version + ")")
elif major == self.mastodon_major and minor == self.mastodon_minor and patch > self.mastodon_patch:
raise MastodonVersionError("Version check failed (Need version " + version + ")")
return function(self, *args, **kwargs)
function.__doc__ = function.__doc__ + "\n\n *Added: Mastodon v" + created_ver + ", last changed: Mastodon v" + last_changed_ver + "*"
return decorate(function, wrapper)
return api_min_version_decorator
###
# Dict helper class.
# Defined at top level so it can be pickled.
###
def logging_scope(name, *wargs, **wkwargs):
'''
A decorator to add the decorated function as a new logging scope, with name `name`.
All additional arguments are passed to `StatusLogger._scope_enter`. Current
supported keyword arguments are "timed", in which case when the scope closes,
the duration of the call is shown.
'''
def logging_scope(func, *args, **kwargs):
logger._scope_enter(name, *wargs, **wkwargs)
success = True
try:
f = func(*args, **kwargs)
return f
except Exception as e:
success = False
raise_with_traceback(e)
finally:
logger._scope_exit(success)
return lambda func: decorate(func, logging_scope)
def leakcheck(fn):
# Compare the memory (after GC) before and after the test to determine
# if we leaked memory
def wrapper(fn, *args, **kwargs):
gc.collect()
mem_before = MemoryUsage()
rv = fn(*args, **kwargs)
gc.collect()
mem_after = MemoryUsage()
assert mem_before >= mem_after
return rv
return decorator.decorate(fn, wrapper)
def node(comp, name=None, *args, **kw):
def inner(f):
if name is None:
comp.add_node(f.__name__, f, *args, **kw)
else:
comp.add_node(name, f, *args, **kw)
return decorator.decorate(f, _node)
return inner
def _add_contract_to_setter(setter_fun, var_name, property_contract, property_name):
# 0. check that we can import contracts
try:
# noinspection PyUnresolvedReferences
from contracts import ContractNotRespected, contract
except ImportError as e:
raise Exception('Use of _add_contract_to_setter requires that PyContract library is installed. Check that you '
'can \'import contracts\'')
# -- check if a contract already exists on the function
if hasattr(setter_fun, '__contracts__'):
msg = 'overridden setter for attribute ' + property_name + ' implemented by function ' \
+ str(setter_fun.__qualname__) + ' has a contract while there is a contract already defined ' \
+ 'for this property in the __init__ constructor. This will lead to double-contract in the final ' \
+ 'setter, please remove the one on the overridden setter.'
warn(msg)
# -- add the generated contract
setter_fun_with_possible_contract = contract(setter_fun, **{var_name: property_contract})
# the only thing we can't do is to replace the function's parameter name dynamically in the error messages
# so we wrap the function again to catch the potential pycontracts error :(
# old:
# @functools.wraps(func) -> to make the wrapper function look like the wrapped function
# def wrapper(self, *args, **kwargs):
# new:
# we now use 'decorate' to have a wrapper that has the same signature, see below
def _contracts_parser_interceptor(func, self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except ContractNotRespected as er:
er.error = er.error.replace('\'val\'', '\'' + property_name + '\'')
raise er
# f = _contracts_parser_interceptor(f)
setter_fun_with_possible_contract = decorate(setter_fun_with_possible_contract, _contracts_parser_interceptor)
return setter_fun_with_possible_contract
def _add_validators_to_setter(setter_fun, var_name, validators, property_name):
# 0. check that we can import validate
# note: this is useless now but maybe one day validate will be another project ?
try:
# noinspection PyUnresolvedReferences
from valid8 import decorate_with_validators
except ImportError:
raise Exception('Use of _add_contract_to_setter requires that validate library is installed. Check that you can'
' \'import validate\'')
# -- check if a contract already exists on the function
if hasattr(setter_fun, '__validators__'):
msg = 'overridden setter for attribute ' + property_name + ' implemented by function ' \
+ str(setter_fun.__qualname__) + ' has validators while there are validators already defined ' \
'for this property in the __init__ constructor. This will lead to double-contract in the final ' \
'setter, please remove the one on the overridden setter.'
warn(msg)
# -- add the generated contract
setter_fun_with_validation = decorate_with_validators(setter_fun, **{var_name: validators})
# # the only thing we can't do is to replace the function's parameter name dynamically in the validation error
# # messages so we wrap the function again to catch the potential pycontracts error :(
# # old:
# # @functools.wraps(func) -> to make the wrapper function look like the wrapped function
# # def wrapper(self, *args, **kwargs):
# # new:
# # we now use 'decorate' to have a wrapper that has the same signature, see below
# def _contracts_parser_interceptor(func, self, *args, **kwargs):
# try:
# return func(self, *args, **kwargs)
# except ContractNotRespected as e:
# e.error = e.error.replace('\'val\'', '\'' + property_name + '\'')
# raise e
# f = _contracts_parser_interceptor(f)
# setter_fun_with_possible_contract = decorate(setter_fun_with_possible_contract, _contracts_parser_interceptor)
return setter_fun_with_validation
def autoprops_override_decorate(func: Callable, attribute:str = None, is_getter:bool = True) -> Callable:
"""
Used to decorate a function as an overridden getter or setter, without using the @getter_override or
@setter_override annotations. If the overridden setter has no @contract, the contract will still be
dynamically added. Note: this should be executed BEFORE @autoprops or autoprops_decorate().
:param func: the function on which to execute. Note that it won't be wrapped but simply annotated.
:param attribute: the attribute name. If None, the function name will be used
:param is_getter: True for a getter override, False for a setter override.
:return:
"""
# Simply annotate the fact that this is a function
attr_name = attribute or func.__name__
if is_getter:
if hasattr(func, __GETTER_OVERRIDE_ANNOTATION):
raise DuplicateOverrideError('Getter is overridden twice for attribute name : ' + attr_name)
else:
func.__getter_override__ = attr_name
else:
if hasattr(func, __SETTER_OVERRIDE_ANNOTATION):
raise DuplicateOverrideError('Setter is overridden twice for attribute name : ' + attr_name)
else:
func.__setter_override__ = attr_name
return func
def deprecated(*optional_message):
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used.
Parameters
----------
*optional_message : str
an optional user level hint which should indicate which feature to use otherwise.
"""
def _deprecated(func, *args, **kw):
caller_stack = stack()[1:]
while len(caller_stack) > 0:
frame = caller_stack.pop(0)
filename = frame[1]
# skip callee frames if they are other decorators or this file(func)
if 'decorator' in filename or __file__ in filename:
continue
else: break
lineno = frame[2]
# avoid cyclic references!
del caller_stack, frame
user_msg = 'Call to deprecated function "%s". Called from %s line %i. %s' \
% (func.__name__, filename, lineno, msg)
warnings.warn_explicit(
user_msg,
category=Chainsaw_DeprecationWarning,
filename=filename,
lineno=lineno
)
return func(*args, **kw)
# add deprecation notice to func docstring:
if len(optional_message) == 1 and callable(optional_message[0]):
# this is the function itself, decorate!
msg = ""
return decorate(optional_message[0], _deprecated)
else:
# actually got a message (or empty parenthesis)
msg = optional_message[0] if len(optional_message) > 0 else ""
return decorator(_deprecated)
