diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/tenacity/before_sleep.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/tenacity/before_sleep.py | 72 |
1 files changed, 72 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/tenacity/before_sleep.py b/.venv/lib/python3.12/site-packages/tenacity/before_sleep.py new file mode 100644 index 00000000..d04edcf9 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/tenacity/before_sleep.py @@ -0,0 +1,72 @@ +# Copyright 2016 Julien Danjou +# Copyright 2016 Joshua Harlow +# Copyright 2013-2014 Ray Holder +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import typing + +from tenacity import _utils + +if typing.TYPE_CHECKING: + import logging + + from tenacity import RetryCallState + + +def before_sleep_nothing(retry_state: "RetryCallState") -> None: + """Before call strategy that does nothing.""" + + +def before_sleep_log( + logger: "logging.Logger", + log_level: int, + exc_info: bool = False, +) -> typing.Callable[["RetryCallState"], None]: + """Before call strategy that logs to some logger the attempt.""" + + def log_it(retry_state: "RetryCallState") -> None: + local_exc_info: BaseException | bool | None + + if retry_state.outcome is None: + raise RuntimeError("log_it() called before outcome was set") + + if retry_state.next_action is None: + raise RuntimeError("log_it() called before next_action was set") + + if retry_state.outcome.failed: + ex = retry_state.outcome.exception() + verb, value = "raised", f"{ex.__class__.__name__}: {ex}" + + if exc_info: + local_exc_info = retry_state.outcome.exception() + else: + local_exc_info = False + else: + verb, value = "returned", retry_state.outcome.result() + local_exc_info = False # exc_info does not apply when no exception + + if retry_state.fn is None: + # NOTE(sileht): can't really happen, but we must please mypy + fn_name = "<unknown>" + else: + fn_name = _utils.get_callback_name(retry_state.fn) + + logger.log( + log_level, + f"Retrying {fn_name} " + f"in {retry_state.next_action.sleep} seconds as it {verb} {value}.", + exc_info=local_exc_info, + ) + + return log_it |