У меня есть два декоратора как тайм-аут и повтор, и у меня есть две функции, одна из них имеет тайм-аут, а другая имеет повтор, как это:
@timeout(seconds=1)
def func_inner(timeout):
time.sleep(timeout)
@retry(count=2, message="Failed command after {timeout} seconds")
def func(timeout):
func_inner(timeout)
func(timeout=3)
Дело в том, что когда func_inner
выдает ошибку тайм-аута из-за декоратора тайм-аута, я хочу, чтобы func()
знал, что его атрибут имеет значение timeout
, и мы повторим попытку для этой ошибки и покажем сообщение об ошибке, которое мы определили ' Сбой команды после ... ', это повторный декоратор, см. Жирную строку:
def retry(count=1, delay=None, expected_value=None, raise_exception=True,
ignore_exceptions=[], message=None):
def _get_param(kargs, name, default):
return kargs.pop('retry_' + name, default)
def decorator(fn):
def wrapper(*args, **kargs):
'''
Retry settings can be overridden on function call
@retry(count=2)
def fn(param):
return param
fn(123, retry_count=10, retry_delay=10)
'''
_count = _get_param(kargs, 'count', count)
_delay = _get_param(kargs, 'delay', delay)
_expected_value = _get_param(kargs, 'expected_value', expected_value)
_raise_exception = _get_param(kargs, 'raise_exception', raise_exception)
_ignore_exceptions = _get_param(kargs, 'ignore_exceptions', ignore_exceptions)
_message = _get_param(kargs, 'message', message)
# convert single item to list
_ignore_exceptions = \
_ignore_exceptions \
if isinstance(_ignore_exceptions, list) \
else [_ignore_exceptions]
logging.debug('bytecode:')
logging.debug(list(get_instructions(fn)))
# If function is wrapped into @timeout
# then ignore TimeoutError exception implicitly
if hasattr(fn, 'timeout') :
_ignore_exceptions.append(TimeoutError)
# Retry parameters can be used for recursive functions
# see autotest.sertver.utils.http_list
wrapper.retry_params = dict(
retry_count=_count,
retry_delay=_delay,
retry_expected_value=_expected_value,
retry_raise_exception=_raise_exception,
retry_message=_message,
)
log_params = dict(
wrapper.retry_params,
**convert_func_arguments_to_keyword(fn, *args, **kargs))
fn_expected = _expected_value \
if isinstance(_expected_value, types.FunctionType) \
else None
_message = _message + ' ' if _message else ''
message_unexpected_value = upperfirst(
_message +
("got unexpected value <{retry_actual_value}>" if fn_expected else
"expected value <{retry_expected_value}> but was <{retry_actual_value}>")
)
message_exception = upperfirst(_message + "got error: {exception_type} - {retry_error}")
actual_value = None
last_exception = None
index = 1
while (index <= _count):
try:
actual_value = fn(*args, **kargs)
if _expected_value is None or \
(fn_expected(actual_value) if fn_expected else actual_value == _expected_value):
return actual_value
# Don't log single try
if _message and _count > 1:
print "[%s/%s] %s" % (
index, _count,
message_unexpected_value.format(
retry_actual_value=actual_value,
** log_params)
)
except Exception as e:
# Reset previous actual_value in case of exception
actual_value = None
last_exception = e
for ex in _ignore_exceptions:
if (isinstance(ex, types.TypeType) and isinstance_or_cause(e, ex)) \
or (isinstance(ex, types.FunctionType) and ex(e)):
if _message:
print "[%s/%s] %s" % (
index, _count,
message_exception.format(
exception_type=type(e).__name__,
retry_error=e,
** log_params)
)
break
else:
raise e
index += 1
if _delay:
time.sleep(_delay)
if last_exception:
raise RetryError(
message_exception.format(
exception_type=type(last_exception).__name__,
retry_error=last_exception,
**log_params))
if _raise_exception:
raise RetryError(
message_unexpected_value.format(
retry_actual_value=actual_value,
**log_params))
return actual_value
return wrapper
return decorator
Но код выше, с этим:
if hasattr(fn, 'timeout') :
_ignore_exceptions.append(TimeoutError)
Я хочу добавить TimeoutError в ignore_exceptions, если функция имеет timeout attr, она будет иметься, если она будет заключена в тайм-аут, но теперь func_inner использует @timeout вместо func, _ignore_exceptions
никогда не будет включать TimeoutError
, как можно func
знаю, что она использует функцию func_inner
с использованием декоратора тайм-аута, и я хочу добавить TimeourError
в ignore_exceptions для повторной попытки.