We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
I have a question: Is it possible to create async tenacity decorator which sleep time will be depended on exception value?
tenacity
class TelegramRetryAfter(TelegramAPIError): def __init__(self, retry_after: int) -> None: ... def retry_telegram_method(attempts: int = 3) -> Callable: return retry( stop=stop_after_attempt(attempts), retry=( retry_if_exception_cause_type(TelegramAPIError) | retry_if_not_exception_type(TelegramBadRequest) ), before_sleep=_my_before_sleep ) async def _my_before_sleep(retry_state: RetryCallState) -> None: if retry_state.outcome.failed: exc_info = retry_state.outcome.exception() else: exc_info = False default_logger = before_sleep_log(logging.getLogger(), log_level=logging.DEBUG, exc_info=exc_info) default_logger(retry_state) # Here i want to implement sleep if exc_info and isinstance(exc_info, TelegramRetryAfter): logging.debug(f'Retrying {retry_state.fn} | attempt {retry_state.attempt_number} | ' f'sleep for {exc_info.retry_after}') await asyncio.sleep(exc_info.retry_after)
Error: RuntimeWarning: coroutine '_my_before_sleep' was never awaited
RuntimeWarning: coroutine '_my_before_sleep' was never awaited
I can't make def _my_before_sleep because i need await asyncio.sleep() inside
def _my_before_sleep
await asyncio.sleep()
The text was updated successfully, but these errors were encountered:
def wait_if_retry_after(retry_state: RetryCallState) -> float: exc = retry_state.outcome.exception() if isinstance(exc, TelegramRetryAfter): logging.debug(f'{exc.__class__.__name__} {exc.retry_after=}') return exc.retry_after retry_telegram_method = retry( stop=stop_after_attempt(3), retry=( retry_if_exception_cause_type(TelegramAPIError) | retry_if_not_exception_type(TelegramBadRequest) ), before_sleep=before_sleep_log(logging.getLogger(), logging.DEBUG), after=after_log(logging.getLogger(), logging.DEBUG), wait=wait_if_retry_after )
Sorry, something went wrong.
No branches or pull requests
I have a question:
Is it possible to create async
tenacity
decorator which sleep time will be depended on exception value?Error:
RuntimeWarning: coroutine '_my_before_sleep' was never awaited
I can't make
def _my_before_sleep
because i needawait asyncio.sleep()
insideThe text was updated successfully, but these errors were encountered: