Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def log_it(func, trial_number):
logger.log(log_level,
"Starting call to '%s', this is the %s time calling it.",
_utils.get_callback_name(func),
_utils.to_ordinal(trial_number))
def log_it(call_state):
if call_state.outcome.failed:
verb, value = 'raised', call_state.outcome.exception()
else:
verb, value = 'returned', call_state.outcome.result()
logger.log(log_level,
"Retrying %s in %s seconds as it %s %s.",
_utils.get_callback_name(call_state.fn),
getattr(call_state.next_action, 'sleep'),
verb, value)
return log_it
def log_it(func, trial_number, trial_time_taken):
logger.log(log_level, log_tpl,
_utils.get_callback_name(func), trial_time_taken,
_utils.to_ordinal(trial_number))
def warn_about_non_retry_state_deprecation(cbname, func, stacklevel):
msg = (
'"%s" function must accept single "retry_state" parameter,'
' please update %s' % (cbname, _utils.get_callback_name(func)))
warn(msg, DeprecationWarning, stacklevel=stacklevel + 1)
def before_log(retry_state):
if retry_state.outcome.failed:
verb, value = "raised", retry_state.outcome.exception()
else:
verb, value = "returned", retry_state.outcome.result()
logger.info(
"Retrying {callback} in {sleep} seconds as it {verb} {value}",
callback=_utils.get_callback_name(retry_state.fn),
sleep=getattr(retry_state.next_action, "sleep"),
verb=verb,
value=value,
)
def warn_about_dunder_non_retry_state_deprecation(fn, stacklevel):
msg = (
'"%s" method must be called with'
' single "retry_state" parameter' % (_utils.get_callback_name(fn)))
warn(msg, DeprecationWarning, stacklevel=stacklevel + 1)
def after_log(retry_state):
logger.info(
"Finished call to {callback!r} after {time:.2f}, this was the {attempt} time calling it.",
callback=_utils.get_callback_name(retry_state.fn),
time=retry_state.seconds_since_start,
attempt=_utils.to_ordinal(retry_state.attempt_number),
)