# SPDX-FileCopyrightText: 2022 Hynek Schlawack <hs@ox.cx>## SPDX-License-Identifier: MITfrom__future__importannotationsfromthreadingimportLockfromtypingimportCallablefrom.instrumentationimportRetryHookFactoryfrom.instrumentation._hooksimportget_default_hooks,init_hooksfrom.typingimportRetryHookclass_Testing:""" Test mode specification. Strictly private. """__slots__=("attempts",)attempts:intdef__init__(self,attempts:int)->None:self.attempts=attemptsclass_Config:""" Global stamina configuration. Strictly private. """__slots__=("lock","_is_active","_on_retry","_get_on_retry","_testing",)lock:Lock_is_active:bool_testing:_Testing|None_on_retry:(tuple[RetryHook,...]|tuple[RetryHook|RetryHookFactory,...]|None)_get_on_retry:Callable[[],tuple[RetryHook,...]]def__init__(self,lock:Lock)->None:self.lock=lockself._is_active=Trueself._testing=None# Prepare delayed initialization.self._on_retry=Noneself._get_on_retry=self._init_on_first_retry@propertydefis_active(self)->bool:returnself._is_active@is_active.setterdefis_active(self,value:bool)->None:withself.lock:self._is_active=value@propertydeftesting(self)->_Testing|None:returnself._testing@testing.setterdeftesting(self,value:_Testing|None)->None:withself.lock:self._testing=value@propertydefon_retry(self)->tuple[RetryHook,...]:returnself._get_on_retry()@on_retry.setterdefon_retry(self,value:tuple[RetryHook|RetryHookFactory,...]|None)->None:withself.lock:self._get_on_retry=self._init_on_first_retryself._on_retry=valuedef_init_on_first_retry(self)->tuple[RetryHook,...]:""" Perform delayed initialization of on_retry hooks. """withself.lock:# Ensure hooks didn't init while waiting for the lock.ifself._get_on_retry==self._init_on_first_retry:ifself._on_retryisNone:self._on_retry=get_default_hooks()self._on_retry=init_hooks(self._on_retry)self._get_on_retry=lambda:self._on_retry# type: ignore[assignment, return-value]returnself._on_retry# type: ignore[return-value]CONFIG=_Config(Lock())
[docs]defis_active()->bool:""" Check whether retrying is active. Returns: Whether retrying is active. """returnCONFIG.is_active
[docs]defset_active(active:bool)->None:""" Activate or deactivate retrying. Is idempotent and can be called repeatedly with the same value. """CONFIG.is_active=bool(active)
[docs]defis_testing()->bool:""" Check whether test mode is enabled. .. versionadded:: 24.3.0 """returnCONFIG.testingisnotNone
[docs]defset_testing(testing:bool,*,attempts:int=1)->None:""" Activate or deactivate test mode. In testing mode, backoffs are disabled, and attempts are capped to *attempts*. Is idempotent and can be called repeatedly with the same values. .. versionadded:: 24.3.0 """CONFIG.testing=_Testing(attempts)iftestingelseNone