projects
/
python_utils.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Improve text wrapping; add indent option.
[python_utils.git]
/
decorator_utils.py
diff --git
a/decorator_utils.py
b/decorator_utils.py
index 68a9d69633f6babe78c80153753d5c4a41c150af..5d1e779deeeaa3df34547f853ce5e4fc0cdcb87b 100644
(file)
--- a/
decorator_utils.py
+++ b/
decorator_utils.py
@@
-133,7
+133,7
@@
def rate_limited(n_calls: int, *, per_period_in_seconds: float = 1.0) -> Callabl
wait_time = min_interval_seconds - elapsed_since_last
else:
wait_time = 0.0
wait_time = min_interval_seconds - elapsed_since_last
else:
wait_time = 0.0
- logger.debug(
f'@{time.time()}> wait_time = {wait_time}'
)
+ logger.debug(
'@%.4f> wait_time = %.4f', time.time(), wait_time
)
return wait_time
def wrapper_wrapper_rate_limited(*args, **kargs) -> Any:
return wait_time
def wrapper_wrapper_rate_limited(*args, **kargs) -> Any:
@@
-145,10
+145,12
@@
def rate_limited(n_calls: int, *, per_period_in_seconds: float = 1.0) -> Callabl
):
break
with cv:
):
break
with cv:
- logger.debug(
f'@{time.time()}> calling it...'
)
+ logger.debug(
'@%.4f> calling it...', time.time()
)
ret = func(*args, **kargs)
last_invocation_timestamp[0] = time.time()
ret = func(*args, **kargs)
last_invocation_timestamp[0] = time.time()
- logger.debug(f'@{time.time()}> Last invocation <- {last_invocation_timestamp[0]}')
+ logger.debug(
+ '@%.4f> Last invocation <- %.4f', time.time(), last_invocation_timestamp[0]
+ )
cv.notify()
return ret
cv.notify()
return ret
@@
-225,6
+227,11
@@
def debug_count_calls(func: Callable) -> Callable:
class DelayWhen(enum.IntEnum):
class DelayWhen(enum.IntEnum):
+ """When should we delay: before or after calling the function (or
+ both)?
+
+ """
+
BEFORE_CALL = 1
AFTER_CALL = 2
BEFORE_AND_AFTER = 3
BEFORE_CALL = 1
AFTER_CALL = 2
BEFORE_AND_AFTER = 3
@@
-259,11
+266,11
@@
def delay(
@functools.wraps(func)
def wrapper_delay(*args, **kwargs):
if when & DelayWhen.BEFORE_CALL:
@functools.wraps(func)
def wrapper_delay(*args, **kwargs):
if when & DelayWhen.BEFORE_CALL:
- logger.debug(
f"@delay for {seconds}s BEFORE_CALL to {func.__name__}"
)
+ logger.debug(
"@delay for %fs BEFORE_CALL to %s", seconds, func.__name__
)
time.sleep(seconds)
retval = func(*args, **kwargs)
if when & DelayWhen.AFTER_CALL:
time.sleep(seconds)
retval = func(*args, **kwargs)
if when & DelayWhen.AFTER_CALL:
- logger.debug(
f"@delay for {seconds}s AFTER_CALL to {func.__name__}"
)
+ logger.debug(
"@delay for %fs AFTER_CALL to %s", seconds, func.__name__
)
time.sleep(seconds)
return retval
time.sleep(seconds)
return retval
@@
-288,7
+295,7
@@
class _SingletonWrapper:
def __call__(self, *args, **kwargs):
"""Returns a single instance of decorated class"""
def __call__(self, *args, **kwargs):
"""Returns a single instance of decorated class"""
- logger.debug(
f"@singleton returning global instance of {self.__wrapped__.__name__}"
)
+ logger.debug(
'@singleton returning global instance of %s', self.__wrapped__.__name__
)
if self._instance is None:
self._instance = self.__wrapped__(*args, **kwargs)
return self._instance
if self._instance is None:
self._instance = self.__wrapped__(*args, **kwargs)
return self._instance
@@
-355,13
+362,13
@@
def memoized(func: Callable) -> Callable:
cache_key = args + tuple(kwargs.items())
if cache_key not in wrapper_memoized.cache:
value = func(*args, **kwargs)
cache_key = args + tuple(kwargs.items())
if cache_key not in wrapper_memoized.cache:
value = func(*args, **kwargs)
- logger.debug(
f"Memoizing {cache_key} => {value} for {func.__name__}"
)
+ logger.debug(
'Memoizing %s => %s for %s', cache_key, value, func.__name__
)
wrapper_memoized.cache[cache_key] = value
else:
wrapper_memoized.cache[cache_key] = value
else:
- logger.debug(
f"Returning memoized value for {func.__name__}"
)
+ logger.debug(
'Returning memoized value for %s', {func.__name__}
)
return wrapper_memoized.cache[cache_key]
return wrapper_memoized.cache[cache_key]
- wrapper_memoized.cache =
dict()
# type: ignore
+ wrapper_memoized.cache =
{}
# type: ignore
return wrapper_memoized
return wrapper_memoized
@@
-403,7
+410,7
@@
def retry_predicate(
@functools.wraps(f)
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay_sec # make mutable
@functools.wraps(f)
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay_sec # make mutable
- logger.debug(
f'deco_retry: will make up to {mtries} attempts...'
)
+ logger.debug(
'deco_retry: will make up to %d attempts...', mtries
)
retval = f(*args, **kwargs)
while mtries > 0:
if predicate(retval) is True:
retval = f(*args, **kwargs)
while mtries > 0:
if predicate(retval) is True:
@@
-537,7
+544,7
@@
def thunkify(func):
def _raise_exception(exception, error_message: Optional[str]):
if error_message is None:
def _raise_exception(exception, error_message: Optional[str]):
if error_message is None:
- raise Exception()
+ raise Exception(
exception
)
else:
raise Exception(error_message)
else:
raise Exception(error_message)
@@
-620,6
+627,7
@@
class _Timeout(object):
if flag:
return load
raise load
if flag:
return load
raise load
+ return None
def timeout(
def timeout(
@@
-722,7
+730,8
@@
def call_with_sample_rate(sample_rate: float) -> Callable:
if random.uniform(0, 1) < sample_rate:
return f(*args, **kwargs)
else:
if random.uniform(0, 1) < sample_rate:
return f(*args, **kwargs)
else:
- logger.debug(f"@call_with_sample_rate skipping a call to {f.__name__}")
+ logger.debug("@call_with_sample_rate skipping a call to %s", f.__name__)
+ return None
return _call_with_sample_rate
return _call_with_sample_rate