diff --git a/.gitignore b/.gitignore
index b7522274b68c279154e6049d7fef8a887b10c1f2..c542b97aab1d6d40f84bdef9d0144c9ddf9493cb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,12 +4,14 @@
test*.json
test*.txt
.pytest_cache
+test-*
.eggs
satella.sublime*
hs_err_pid*.log
*.pydevproject
.coverage.*
.project
+tests/test_coding/__pycache__
venv
coverage.xml
.coverage.*
diff --git a/satella/coding/concurrent/callablegroup.py b/satella/coding/concurrent/callablegroup.py
index ac79846da369925556c36c7575a6a9bf531d7116..175aba0489e256ac72f6a971e49d7fbba236dfd1 100644
--- a/satella/coding/concurrent/callablegroup.py
+++ b/satella/coding/concurrent/callablegroup.py
@@ -6,6 +6,7 @@ import typing as tp
from satella.coding.deleters import DictDeleter, IterMode
from satella.coding.typing import T, NoArgCallable
+import inspect
class CancellableCallback:
@@ -24,6 +25,7 @@ class CancellableCallback:
Hashable and __eq__-able by identity. Equal only to itself.
:param callback_fun: function to call
+ :param hash: reserved for internal use, don't use
:ivar cancelled: whether this callback was cancelled (bool)
:ivar one_shotted: whether this callback was invoked (bool)
@@ -39,7 +41,7 @@ class CancellableCallback:
self.one_shotted = False
def __hash__(self):
- return hash(id(self))
+ return hash(id(self)) ^ self.__hash
def __eq__(self, other: CancellableCallback) -> bool:
return id(self) == id(other) and self.cancelled == other.cancelled
@@ -55,11 +57,12 @@ class CancellableCallback:
self.cancelled = True
-def _callable_to_cancellablecallback(callback: NoArgCallable[[T], None]) -> CancellableCallback:
- if isinstance(callback, NoArgCallable[[T], None]):
- return CancellableCallback(callback)
- elif isinstance(callback, CancellableCallback):
+def _callable_to_cancellablecallback(callback: NoArgCallable) -> CancellableCallback:
+ if isinstance(callback, CancellableCallback):
return callback
+ if callable(callback):
+ return CancellableCallback(callback)
+ raise ValueError('Invalid callable')
class CancellableCallbackGroup:
@@ -111,7 +114,7 @@ class CallableGroup(tp.Generic[T]):
__slots__ = 'callables', 'gather', 'swallow_exceptions',
def __init__(self, gather: bool = True, swallow_exceptions: bool = False):
- self.callables = collections.OrderedDict() # type: tp.Dict[tp.Callable, tuple[bool]]
+ self.callables = collections.OrderedDict() # type: tp.Dict[CancellableCallback, bool]
self.gather = gather # type: bool
self.swallow_exceptions = swallow_exceptions # type: bool
@@ -131,23 +134,22 @@ class CallableGroup(tp.Generic[T]):
"""
Remove it's entries that are CancelledCallbacks and that were cancelled and move one shots
"""
- with DictDeleter(self.callables, iter_mode=IterMode.ITER_VALUES) as dd:
+ with DictDeleter(self.callables, iter_mode=IterMode.ITER_ITEMS) as dd:
for callable_, oneshot in dd:
- if isinstance(callable_, CancellableCallback) and not callable_:
+ if isinstance(callable_, CancellableCallback) and not callable_.cancelled:
dd.delete()
if oneshot:
dd.delete()
- def add_many(self, callable_: tp.Sequence[tp.Union[NoArgCallable[T],
- tp.Tuple[NoArgCallable[T], bool]]]) -> CancellableCallbackGroup:
+ def add_many(self, callable_: tp.Sequence[tp.Union[NoArgCallable,
+ tp.Tuple[NoArgCallable, bool]]]) -> CancellableCallbackGroup:
"""
Add multiple callbacks
- .. note:: Same callable can't be added twice. It will silently fail.
- Note that already called one-shots can be added twice
-
Basically every callback is cancellable.
+ .. warning: Same callable cannot be added twice. A RuntimeError will be raised in that case.
+
:param callable_: sequence of either callables with will be registered as multiple-shots or a tuple of callback
(with an argument to register it as a one-shot)
:returns: CancellableCallbackGroup to cancel all of the callbacks
@@ -169,17 +171,17 @@ class CallableGroup(tp.Generic[T]):
"""
Add a callable.
- .. note:: Same callable can't be added twice. It will silently fail, and return an existing callbacks.
- Note that already called one-shots can be added twice
-
Can be a :class:`~satella.coding.concurrent.CancellableCallback`, in that case
method :meth:`~satella.coding.concurrent.CallableGroup.remove_cancelled` might
be useful.
+ .. warning: Same callable cannot be added twice. A RuntimeError will be raised in that case.
+
Basically every callback is cancellable.
:param callable_: callable
:param one_shot: if True, callable will be unregistered after single call
+ :param hash: internal, don't use
:returns: callable_ if it was a cancellable callback, else one constructed after it
.. deprecated:: v2.25.5
@@ -189,7 +191,7 @@ class CallableGroup(tp.Generic[T]):
callable_ = _callable_to_cancellablecallback(callable_)
self.callables[callable_] = one_shot
if callable_ in self.callables:
- return callable_
+ raise RuntimeError('Same callable added twice')
return callable_
def __call__(self, *args, **kwargs) -> tp.Optional[tp.List[T]]:
@@ -231,6 +233,10 @@ class CallableGroup(tp.Generic[T]):
if self.gather:
return results
+ def __len__(self):
+ self.remove_cancelled()
+ return len(self.callables)
+
class CallNoOftenThan:
"""
diff --git a/satella/coding/concurrent/functions.py b/satella/coding/concurrent/functions.py
index cb72b14ace5a80167190baac44773cc324425cf6..99014b038d24ccf8da822d06b4ae39f5f31db807 100644
--- a/satella/coding/concurrent/functions.py
+++ b/satella/coding/concurrent/functions.py
@@ -2,7 +2,7 @@ import typing as tp
from concurrent.futures import Future
from threading import Thread
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.sequences.sequences import infinite_iterator
from satella.coding.typing import T
diff --git a/satella/coding/concurrent/monitor.py b/satella/coding/concurrent/monitor.py
index d9d2bee1b424edf41085270992ea8bd85b605615..394c8a891b735274ea55b943b6d149ba62846749 100644
--- a/satella/coding/concurrent/monitor.py
+++ b/satella/coding/concurrent/monitor.py
@@ -3,7 +3,7 @@ import copy
import threading
import typing as tp
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import K, V, T
diff --git a/satella/coding/concurrent/queue.py b/satella/coding/concurrent/queue.py
index 264e1f4401e9b43db08eb3d6c118e82e95a2ace1..2db251ccf66134d80128b47fff08d1539c7d7803 100644
--- a/satella/coding/concurrent/queue.py
+++ b/satella/coding/concurrent/queue.py
@@ -55,6 +55,7 @@ class PeekableQueue(tp.Generic[T]):
try:
return item_getter(self.queue)
finally:
+ self.items_count -= 1
self.lock.release()
def __get_timeout(self, item_getter, timeout):
@@ -68,6 +69,7 @@ class PeekableQueue(tp.Generic[T]):
try:
return item_getter(self.queue)
finally:
+ self.items_count -= 1
self.lock.release()
else:
self.lock.release()
@@ -75,21 +77,19 @@ class PeekableQueue(tp.Generic[T]):
@rethrow_as(WouldWaitMore, Empty)
def __get(self, timeout, item_getter) -> T:
- try:
- self.lock.acquire()
- if len(self.queue):
- # Fast path
- try:
- return item_getter(self.queue)
- finally:
- self.lock.release()
+ self.lock.acquire()
+ if len(self.queue):
+ # Fast path
+ try:
+ return item_getter(self.queue)
+ finally:
+ self.items_count -= 1
+ self.lock.release()
+ else:
+ if timeout is None:
+ return self.__get_timeout_none(item_getter)
else:
- if timeout is None:
- return self.__get_timeout_none(item_getter)
- else:
- return self.__get_timeout(item_getter, timeout)
- finally:
- self.items_count -= 1
+ return self.__get_timeout(item_getter, timeout)
def get(self, timeout: tp.Optional[float] = None) -> T:
"""
@@ -100,10 +100,7 @@ class PeekableQueue(tp.Generic[T]):
:return: the item
:raise Empty: queue was empty
"""
- try:
- return self.__get(timeout, lambda queue: queue.popleft())
- finally:
- self.items_count -= 1
+ return self.__get(timeout, lambda queue: queue.popleft())
def peek(self, timeout: tp.Optional[float] = None) -> T:
"""
@@ -125,4 +122,4 @@ class PeekableQueue(tp.Generic[T]):
return self.items_count
def __len__(self):
- return self.items_count
\ No newline at end of file
+ return self.items_count
diff --git a/satella/coding/decorators/__init__.py b/satella/coding/decorators/__init__.py
index 4589d29cd20f6e0ce2e34e2e5ae19b2d888c32a2..4ad8d19f0ae74449cc762d2068ca0a175fb08324 100644
--- a/satella/coding/decorators/__init__.py
+++ b/satella/coding/decorators/__init__.py
@@ -2,11 +2,12 @@ from .arguments import auto_adapt_to_methods, attach_arguments, for_argument, \
execute_before, copy_arguments, replace_argument_if, transform_result, \
transform_arguments, execute_if_attribute_none, execute_if_attribute_not_none, \
cached_property
-from .decorators import wraps, chain_functions, has_keys, short_none, memoize, return_as_list, \
+from .decorators import chain_functions, has_keys, short_none, memoize, return_as_list, \
default_return, cache_memoize, call_method_on_exception
from .flow_control import loop_while, queue_get, repeat_forever
from .preconditions import postcondition, precondition
from .retry_dec import retry
+from .wraps import wraps
__all__ = ['retry', 'transform_result', 'transform_arguments', 'repeat_forever',
'execute_before', 'postcondition', 'precondition', 'wraps', 'queue_get',
diff --git a/satella/coding/decorators/arguments.py b/satella/coding/decorators/arguments.py
index 32fed98e48ebf2b0029347e2faa761d389b2da63..99e9058325ca6b3554e1c11bceaf8a09f2fb9ed0 100644
--- a/satella/coding/decorators/arguments.py
+++ b/satella/coding/decorators/arguments.py
@@ -4,7 +4,7 @@ import itertools
import typing as tp
from inspect import Parameter
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.misc import source_to_function, get_arguments, call_with_arguments, _get_arguments
from satella.coding.predicates import PredicateClass, build_structure
from satella.coding.typing import T, Predicate
diff --git a/satella/coding/decorators/decorators.py b/satella/coding/decorators/decorators.py
index a896e9ab296fdd3764125fd341eb47703edec262..523c8e3c463543fe9134549f8e4b09359570e531 100644
--- a/satella/coding/decorators/decorators.py
+++ b/satella/coding/decorators/decorators.py
@@ -1,8 +1,9 @@
-import inspect
import time
import typing as tp
import warnings
+from satella.coding.decorators.wraps import wraps
+
from satella.coding.typing import T, U
from satella.exceptions import PreconditionError
@@ -221,52 +222,6 @@ def memoize(fun):
return inner
-def wraps(cls_to_wrap: tp.Type) -> tp.Callable[[tp.Type], tp.Type]:
- """
- A functools.wraps() but for classes.
-
- As a matter of fact, this can replace functools.wraps() entirely.
- This replaces __doc__, __name__, __module__ and __annotations__.
- It also sets a correct __wrapped__.
-
- :param cls_to_wrap: class to wrap
- """
-
- def outer(cls: tp.Type) -> tp.Type:
- if hasattr(cls_to_wrap, '__doc__'):
- try:
- cls.__doc__ = cls_to_wrap.__doc__
- except AttributeError:
- pass
- if hasattr(cls_to_wrap, '__name__'):
- try:
- cls.__name__ = cls_to_wrap.__name__
- except (AttributeError, TypeError):
- pass
- if hasattr(cls_to_wrap, '__module__'):
- try:
- cls.__module__ = cls_to_wrap.__module__
- except AttributeError:
- pass
- if hasattr(cls_to_wrap, '__annotations__'):
- try:
- cls.__annotations__ = cls_to_wrap.__annotations__
- except (AttributeError, TypeError):
- pass
- try:
- sig = inspect.signature(cls_to_wrap)
- cls.__signature__ = sig
- except (TypeError, ValueError, RecursionError, AttributeError):
- pass
- try:
- cls.__wrapped__ = cls_to_wrap
- except AttributeError:
- pass
- return cls
-
- return outer
-
-
def has_keys(keys: tp.List[str]):
"""
A decorator for asserting that a dictionary has given keys. Will raise PreconditionError if
diff --git a/satella/coding/decorators/flow_control.py b/satella/coding/decorators/flow_control.py
index 5ad83c6e6ea9f8bb34b41ef9e028d645f51d6b99..9aff4f7179acede18b0335ee796e1b6520830926 100644
--- a/satella/coding/decorators/flow_control.py
+++ b/satella/coding/decorators/flow_control.py
@@ -1,7 +1,7 @@
import queue
import typing as tp
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import ExceptionClassType, NoArgCallable, Predicate
Queue = tp.TypeVar('Queue')
diff --git a/satella/coding/decorators/preconditions.py b/satella/coding/decorators/preconditions.py
index f27d4315d54bee6755186278c142a02659d6dd97..6f38c02914a8e08f49713a07f71d5c34a1399ca9 100644
--- a/satella/coding/decorators/preconditions.py
+++ b/satella/coding/decorators/preconditions.py
@@ -4,7 +4,7 @@ import typing as tp
from satella.coding.typing import T, Predicate
from satella.exceptions import PreconditionError
from .arguments import for_argument
-from .decorators import wraps
+from .wraps import wraps
from ..misc import source_to_function
Expression = tp.NewType('Expression', str)
diff --git a/satella/coding/decorators/retry_dec.py b/satella/coding/decorators/retry_dec.py
index 5a7de6874d3bfc574f44d8646da767ae916e4498..70a755f3bd6b525b85b2cbbcb53ac6e9288eaac1 100644
--- a/satella/coding/decorators/retry_dec.py
+++ b/satella/coding/decorators/retry_dec.py
@@ -2,7 +2,7 @@ import itertools
import typing as tp
import warnings
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import ExceptionClassType
diff --git a/satella/coding/decorators/wraps.py b/satella/coding/decorators/wraps.py
new file mode 100644
index 0000000000000000000000000000000000000000..aa2105515c63e969c8a0550ada04ef02a1d84960
--- /dev/null
+++ b/satella/coding/decorators/wraps.py
@@ -0,0 +1,47 @@
+import inspect
+import typing as tp
+
+def wraps(cls_to_wrap: tp.Type) -> tp.Callable[[tp.Type], tp.Type]:
+ """
+ A functools.wraps() but for classes.
+
+ As a matter of fact, this can replace functools.wraps() entirely.
+ This replaces __doc__, __name__, __module__ and __annotations__.
+ It also sets a correct __wrapped__.
+
+ :param cls_to_wrap: class to wrap
+ """
+
+ def outer(cls: tp.Type) -> tp.Type:
+ if hasattr(cls_to_wrap, '__doc__'):
+ try:
+ cls.__doc__ = cls_to_wrap.__doc__
+ except AttributeError:
+ pass
+ if hasattr(cls_to_wrap, '__name__'):
+ try:
+ cls.__name__ = cls_to_wrap.__name__
+ except (AttributeError, TypeError):
+ pass
+ if hasattr(cls_to_wrap, '__module__'):
+ try:
+ cls.__module__ = cls_to_wrap.__module__
+ except AttributeError:
+ pass
+ if hasattr(cls_to_wrap, '__annotations__'):
+ try:
+ cls.__annotations__ = cls_to_wrap.__annotations__
+ except (AttributeError, TypeError):
+ pass
+ try:
+ sig = inspect.signature(cls_to_wrap)
+ cls.__signature__ = sig
+ except (TypeError, ValueError, RecursionError, AttributeError):
+ pass
+ try:
+ cls.__wrapped__ = cls_to_wrap
+ except AttributeError:
+ pass
+ return cls
+
+ return outer
diff --git a/satella/coding/deleters.py b/satella/coding/deleters.py
index a5a27af3ffd25f88ebcf8d374bf7fd2bbac8d852..55363067024659f3952942ab8d055b8822c97b3c 100644
--- a/satella/coding/deleters.py
+++ b/satella/coding/deleters.py
@@ -39,7 +39,7 @@ class DictDeleter:
__slots__ = ('dict_to_process', 'current_iterator', 'keys_to_delete', 'iter_mode',
'current_key')
- def __init__(self, dict_to_process: collections.abc.MutableMapping, iter_mode: IterMode.ITER_KEYS):
+ def __init__(self, dict_to_process: collections.abc.MutableMapping, iter_mode: IterMode = IterMode.ITER_KEYS):
self.dict_to_process = dict_to_process
self.iter_mode = iter_mode
diff --git a/satella/coding/metaclasses.py b/satella/coding/metaclasses.py
index 9e35cf5b05fc1a9d1e90573152cdb7ee1b3eaf9e..86981b5ed4a523d646390d4ad2ebe34c4eb8b36b 100644
--- a/satella/coding/metaclasses.py
+++ b/satella/coding/metaclasses.py
@@ -1,6 +1,6 @@
import inspect
-from .decorators import wraps
+from .decorators.wraps import wraps
from .sequences.iterators import walk
from .typing import Predicate
diff --git a/satella/coding/predicates.py b/satella/coding/predicates.py
index 46b238769384b3e5a667442c78a145a286877de7..c0d440e8d634de42b7aa9f20bb1d9f673afe37f7 100644
--- a/satella/coding/predicates.py
+++ b/satella/coding/predicates.py
@@ -2,7 +2,7 @@ import operator
import typing as tp
from satella.coding.typing import Predicate
-from satella.configuration.schema import Descriptor
+from satella.configuration.schema.base import Descriptor
__all__ = ['x', 'build_structure', 'PredicateClass']
diff --git a/satella/coding/recast_exceptions.py b/satella/coding/recast_exceptions.py
index 738ced9fe63ea26df01470d9b3c73e9fb082278a..734201aa59163008dd79b27085571d038c5396ae 100644
--- a/satella/coding/recast_exceptions.py
+++ b/satella/coding/recast_exceptions.py
@@ -3,7 +3,7 @@ import logging
import threading
import typing as tp
-from .decorators.decorators import wraps
+from .decorators.wraps import wraps
from .typing import ExceptionClassType, T, NoArgCallable, ExceptionList
diff --git a/satella/coding/sequences/sequences.py b/satella/coding/sequences/sequences.py
index 99be8c89f9e94de804f381514ddb83c92cb32c86..37bde4cd4530f8bdb464dceff7d26f8ee6ea7800 100644
--- a/satella/coding/sequences/sequences.py
+++ b/satella/coding/sequences/sequences.py
@@ -1 +1 @@
-import copy
import typing as tp
from satella.coding.decorators.decorators import wraps
from satella.coding.recast_exceptions import rethrow_as
from .iterators import n_th
from ..typing import T, Iteratable, NoArgCallable, Predicate
def infinite_iterator(returns: tp.Optional[T] = None,
return_factory: tp.Optional[NoArgCallable[T]] = None) -> tp.Iterator[T]:
"""
Return an infinite number of objects.
:param returns: object to return. Note that this will be this very object, it will
not be copied.
:param return_factory: a callable that takes 0 args and returns an element to return.
:return: an infinite iterator of provided values
"""
while True:
if returns is None:
yield None if return_factory is None else return_factory()
else:
yield returns
def make_list(element: T, n: int, deep_copy: bool = False) -> tp.List[T]:
"""
Make a list consisting of n times element. Element will be copied via
copy.copy before adding to list.
:param element: element
:param n: times to repeat the element
:param deep_copy: whether to use copy.deepcopy instead of copy.copy
:return: list of length n
"""
output = []
if deep_copy:
copy_op = copy.deepcopy
else:
copy_op = copy.copy
for _ in range(n):
output.append(copy_op(element))
return output
# shamelessly copied from
# https://medium.com/better-programming/is-this-the-last-element-of-my-python-for-loop-784f5ff90bb5
def is_last(lst: Iteratable) -> tp.Iterator[tp.Tuple[bool, T]]:
"""
Return every element of the list, alongside a flag telling is this the last element.
Use like:
>>> for is_last, element in is_last(my_list):
>>> if is_last:
>>> ...
:param lst: list to iterate thru
:return: a p_gen returning (bool, T)
Note that this returns a nice, O(1) iterator.
"""
iterable = iter(lst)
ret_var = next(iterable)
for val in iterable:
yield False, ret_var
ret_var = val
yield True, ret_var
def add_next(lst: Iteratable,
wrap_over: bool = False,
skip_last: bool = False) -> tp.Iterator[tp.Tuple[T, tp.Optional[T]]]:
"""
Yields a 2-tuple of given iterable, presenting the next element as second element of the tuple.
The last element will be the last element alongside with a None, if wrap_over is False, or the
first element if wrap_over was True
Example:
>>> list(add_next([1, 2, 3, 4, 5])) == [(1, 2), (2, 3), (3, 4), (4, 5), (5, None)]
>>> list(add_next([1, 2, 3, 4, 5], True)) == [(1, 2), (2, 3), (3, 4), (4, 5), (5, 1)]
:param lst: iterable to iterate over
:param wrap_over: whether to attach the first element to the pair of the last element instead
of None
:param skip_last: if this is True, then last element, alongside with a None, won't be output
"""
iterator = iter(lst)
try:
first_val = prev_val = next(iterator)
except StopIteration:
return
for val in iterator:
yield prev_val, val
prev_val = val
if wrap_over:
yield prev_val, first_val
else:
if not skip_last:
yield prev_val, None
def half_cartesian(seq: tp.Iterable[T],
include_same_pairs: bool = True) -> tp.Iterator[tp.Tuple[T, T]]:
"""
Generate half of the Cartesian product of both sequences.
Useful when you have a commutative operation that you'd like to execute on both elements
(eg. checking for collisions).
Example:
>>> list(half_cartesian([1, 2, 3], [1, 2, 3])) == \
>>> [(1, 1), (1, 2), (1, 3), (2, 2), (2, 3), (3, 3)]
:param seq: The sequence
:param include_same_pairs: if True, then pairs returning two of the same objects will be
returned. For example, if False, the following will be true:
>>> list(half_cartesian([1, 2, 3], [1, 2, 3], include_same_pairs=False)) == \
>>> [(1, 2), (1, 3), (2, 3)]
"""
for i, elem1 in enumerate(seq):
for j, elem2 in enumerate(seq):
if include_same_pairs:
if j >= i:
yield elem1, elem2
else:
if j > i:
yield elem1, elem2
def group_quantity(length: int, seq: Iteratable) -> tp.Iterator[tp.List[T]]:
"""
Slice an iterable into lists containing at most len entries.
Eg.
>>> assert list(group_quantity(3, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])) == [[1, 2, 3], [4, 5, 6],
>>> [7, 8, 9], [10]]
This correctly detects sequences, and uses an optimized variant via slicing if
a sequence is passed.
You can safely pass ranges
:param length: p_len for the returning sequences
:param seq: sequence to split
"""
if isinstance(seq, tp.Sequence) and not isinstance(seq, range):
i = 0
while i < len(seq):
yield seq[i:i + length]
i += length
else:
entries = []
for elem in seq:
if len(entries) == length:
yield entries
entries = [elem]
else:
entries.append(elem)
if entries:
yield entries
def filter_out_nones(y: tp.Sequence[T]) -> tp.List[T]:
"""
Return all elements, as a list, that are not None
:param y: a sequence of items
:return: a list of all subelements, in order, that are not None
"""
output = []
for item in y:
if item is not None:
output.append(item)
return output
def filter_out_false(y: tp.Sequence[T]) -> tp.List[T]:
"""
Return all elements, as a list, that are True
:param y: a sequence of items
:return: a list of all subelements, in order, that are not None
"""
output = []
for item in y:
if item:
output.append(item)
return output
@rethrow_as(IndexError, ValueError)
def index_of_max(seq: tp.Sequence[T]) -> int:
"""
Return the index of the maximum element
:param seq: sequence to examine
:return: index of the maximum element
:raise ValueError: sequence was empty
"""
max_index = 0
max_elem = seq[0]
for i, elem in enumerate(seq):
if elem > max_elem:
max_index = i
max_elem = elem
return max_index
def index_of(predicate: Predicate, seq: tp.Sequence[T]) -> int:
"""
Return an index of first met element that calling predicate on it returns True
:param predicate: predicate to apply
:param seq: sequence to examine
:return: index of the element
:raises ValueError: if no element found
"""
i = 0
for elem in seq:
if predicate(elem):
return i
i += 1
raise ValueError('Element not found')
class Multirun:
"""
A class to launch the same operation on the entire sequence.
Consider:
>>> class Counter:
>>> def __init__(self, value=0):
>>> self.count = value
>>> def add(self, v):
>>> self.count += 1
>>> def __eq__(self, other):
>>> return self.count == other.count
>>> def __iadd__(self, other):
>>> self.add(other)
>>> a = [Counter(), Counter()]
The following:
>>> for b in a:
>>> b.add(2)
Can be replaced with
>>> Multirun(a).add(2)
And the following:
>>> for b in a:
>>> b += 3
With this
>>> b = Mulirun(a)
>>> b += 3
Furthermore note that:
>>> Multirun(a).add(2) == [Counter(2), Counter(2)]
:param sequence: sequence to execute these operations for
:param dont_return_list: the operation won't return a list if this is True
"""
__slots__ = 'sequence', 'dont_return_list'
def __bool__(self) -> bool:
return bool(self.sequence)
def __init__(self, sequence: tp.Iterable, dont_return_list: bool = False):
self.sequence = sequence
self.dont_return_list = dont_return_list
def __iter__(self):
return iter(self.sequence)
def __getattr__(self, item):
def inner(*args, **kwargs):
if not self.dont_return_list:
results = []
for element in self:
getattr(element, item)(*args, **kwargs)
results.append(element)
return results
else:
for element in self:
getattr(element, item)(*args, **kwargs)
# Take care: the array might just be empty...
try:
fun = getattr(n_th(self), item)
inner = wraps(fun)(inner)
except IndexError:
pass
return inner
def __iadd__(self, other):
for element in self:
element += other
return self
def __isub__(self, other):
for element in self:
element -= other
return self
def __imul__(self, other):
for element in self:
element *= other
return self
def __itruediv__(self, other):
for element in self:
element /= other
return self
def __ifloordiv__(self, other):
for element in self:
element //= other
return self
def __ilshift__(self, other):
for element in self:
element <<= other
return self
def __irshift__(self, other):
for element in self:
element >>= other
return self
def __ipow__(self, other):
for element in self:
element **= other
return self
\ No newline at end of file
+import copy
import typing as tp
from satella.coding.decorators.wraps import wraps
from satella.coding.recast_exceptions import rethrow_as
from .iterators import n_th
from ..typing import T, Iteratable, NoArgCallable, Predicate
def infinite_iterator(returns: tp.Optional[T] = None,
return_factory: tp.Optional[NoArgCallable[T]] = None) -> tp.Iterator[T]:
"""
Return an infinite number of objects.
:param returns: object to return. Note that this will be this very object, it will
not be copied.
:param return_factory: a callable that takes 0 args and returns an element to return.
:return: an infinite iterator of provided values
"""
while True:
if returns is None:
yield None if return_factory is None else return_factory()
else:
yield returns
def make_list(element: T, n: int, deep_copy: bool = False) -> tp.List[T]:
"""
Make a list consisting of n times element. Element will be copied via
copy.copy before adding to list.
:param element: element
:param n: times to repeat the element
:param deep_copy: whether to use copy.deepcopy instead of copy.copy
:return: list of length n
"""
output = []
if deep_copy:
copy_op = copy.deepcopy
else:
copy_op = copy.copy
for _ in range(n):
output.append(copy_op(element))
return output
# shamelessly copied from
# https://medium.com/better-programming/is-this-the-last-element-of-my-python-for-loop-784f5ff90bb5
def is_last(lst: Iteratable) -> tp.Iterator[tp.Tuple[bool, T]]:
"""
Return every element of the list, alongside a flag telling is this the last element.
Use like:
>>> for is_last, element in is_last(my_list):
>>> if is_last:
>>> ...
:param lst: list to iterate thru
:return: a p_gen returning (bool, T)
Note that this returns a nice, O(1) iterator.
"""
iterable = iter(lst)
ret_var = next(iterable)
for val in iterable:
yield False, ret_var
ret_var = val
yield True, ret_var
def add_next(lst: Iteratable,
wrap_over: bool = False,
skip_last: bool = False) -> tp.Iterator[tp.Tuple[T, tp.Optional[T]]]:
"""
Yields a 2-tuple of given iterable, presenting the next element as second element of the tuple.
The last element will be the last element alongside with a None, if wrap_over is False, or the
first element if wrap_over was True
Example:
>>> list(add_next([1, 2, 3, 4, 5])) == [(1, 2), (2, 3), (3, 4), (4, 5), (5, None)]
>>> list(add_next([1, 2, 3, 4, 5], True)) == [(1, 2), (2, 3), (3, 4), (4, 5), (5, 1)]
:param lst: iterable to iterate over
:param wrap_over: whether to attach the first element to the pair of the last element instead
of None
:param skip_last: if this is True, then last element, alongside with a None, won't be output
"""
iterator = iter(lst)
try:
first_val = prev_val = next(iterator)
except StopIteration:
return
for val in iterator:
yield prev_val, val
prev_val = val
if wrap_over:
yield prev_val, first_val
else:
if not skip_last:
yield prev_val, None
def half_cartesian(seq: tp.Iterable[T],
include_same_pairs: bool = True) -> tp.Iterator[tp.Tuple[T, T]]:
"""
Generate half of the Cartesian product of both sequences.
Useful when you have a commutative operation that you'd like to execute on both elements
(eg. checking for collisions).
Example:
>>> list(half_cartesian([1, 2, 3], [1, 2, 3])) == \
>>> [(1, 1), (1, 2), (1, 3), (2, 2), (2, 3), (3, 3)]
:param seq: The sequence
:param include_same_pairs: if True, then pairs returning two of the same objects will be
returned. For example, if False, the following will be true:
>>> list(half_cartesian([1, 2, 3], [1, 2, 3], include_same_pairs=False)) == \
>>> [(1, 2), (1, 3), (2, 3)]
"""
for i, elem1 in enumerate(seq):
for j, elem2 in enumerate(seq):
if include_same_pairs:
if j >= i:
yield elem1, elem2
else:
if j > i:
yield elem1, elem2
def group_quantity(length: int, seq: Iteratable) -> tp.Iterator[tp.List[T]]:
"""
Slice an iterable into lists containing at most len entries.
Eg.
>>> assert list(group_quantity(3, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])) == [[1, 2, 3], [4, 5, 6],
>>> [7, 8, 9], [10]]
This correctly detects sequences, and uses an optimized variant via slicing if
a sequence is passed.
You can safely pass ranges
:param length: p_len for the returning sequences
:param seq: sequence to split
"""
if isinstance(seq, tp.Sequence) and not isinstance(seq, range):
i = 0
while i < len(seq):
yield seq[i:i + length]
i += length
else:
entries = []
for elem in seq:
if len(entries) == length:
yield entries
entries = [elem]
else:
entries.append(elem)
if entries:
yield entries
def filter_out_nones(y: tp.Sequence[T]) -> tp.List[T]:
"""
Return all elements, as a list, that are not None
:param y: a sequence of items
:return: a list of all subelements, in order, that are not None
"""
output = []
for item in y:
if item is not None:
output.append(item)
return output
def filter_out_false(y: tp.Sequence[T]) -> tp.List[T]:
"""
Return all elements, as a list, that are True
:param y: a sequence of items
:return: a list of all subelements, in order, that are not None
"""
output = []
for item in y:
if item:
output.append(item)
return output
@rethrow_as(IndexError, ValueError)
def index_of_max(seq: tp.Sequence[T]) -> int:
"""
Return the index of the maximum element
:param seq: sequence to examine
:return: index of the maximum element
:raise ValueError: sequence was empty
"""
max_index = 0
max_elem = seq[0]
for i, elem in enumerate(seq):
if elem > max_elem:
max_index = i
max_elem = elem
return max_index
def index_of(predicate: Predicate, seq: tp.Sequence[T]) -> int:
"""
Return an index of first met element that calling predicate on it returns True
:param predicate: predicate to apply
:param seq: sequence to examine
:return: index of the element
:raises ValueError: if no element found
"""
i = 0
for elem in seq:
if predicate(elem):
return i
i += 1
raise ValueError('Element not found')
class Multirun:
"""
A class to launch the same operation on the entire sequence.
Consider:
>>> class Counter:
>>> def __init__(self, value=0):
>>> self.count = value
>>> def add(self, v):
>>> self.count += 1
>>> def __eq__(self, other):
>>> return self.count == other.count
>>> def __iadd__(self, other):
>>> self.add(other)
>>> a = [Counter(), Counter()]
The following:
>>> for b in a:
>>> b.add(2)
Can be replaced with
>>> Multirun(a).add(2)
And the following:
>>> for b in a:
>>> b += 3
With this
>>> b = Mulirun(a)
>>> b += 3
Furthermore note that:
>>> Multirun(a).add(2) == [Counter(2), Counter(2)]
:param sequence: sequence to execute these operations for
:param dont_return_list: the operation won't return a list if this is True
"""
__slots__ = 'sequence', 'dont_return_list'
def __bool__(self) -> bool:
return bool(self.sequence)
def __init__(self, sequence: tp.Iterable, dont_return_list: bool = False):
self.sequence = sequence
self.dont_return_list = dont_return_list
def __iter__(self):
return iter(self.sequence)
def __getattr__(self, item):
def inner(*args, **kwargs):
if not self.dont_return_list:
results = []
for element in self:
getattr(element, item)(*args, **kwargs)
results.append(element)
return results
else:
for element in self:
getattr(element, item)(*args, **kwargs)
# Take care: the array might just be empty...
try:
fun = getattr(n_th(self), item)
inner = wraps(fun)(inner)
except IndexError:
pass
return inner
def __iadd__(self, other):
for element in self:
element += other
return self
def __isub__(self, other):
for element in self:
element -= other
return self
def __imul__(self, other):
for element in self:
element *= other
return self
def __itruediv__(self, other):
for element in self:
element /= other
return self
def __ifloordiv__(self, other):
for element in self:
element //= other
return self
def __ilshift__(self, other):
for element in self:
element <<= other
return self
def __irshift__(self, other):
for element in self:
element >>= other
return self
def __ipow__(self, other):
for element in self:
element **= other
return self
\ No newline at end of file
diff --git a/satella/coding/structures/heaps/base.py b/satella/coding/structures/heaps/base.py
index 41c5da12cb9921b559f75797a1f55840e5a81397..4053619a7f716721718646ffa2123884630f231b 100644
--- a/satella/coding/structures/heaps/base.py
+++ b/satella/coding/structures/heaps/base.py
@@ -3,7 +3,7 @@ import copy
import heapq
import typing as tp
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import T, Predicate
diff --git a/satella/coding/structures/proxy.py b/satella/coding/structures/proxy.py
index 54ec9b5aa9fc3feb785a97f531f89288560dd1dd..20873dfa9b67ab536c0dbf7f3ca9d90ad55d159d 100644
--- a/satella/coding/structures/proxy.py
+++ b/satella/coding/structures/proxy.py
@@ -2,7 +2,7 @@ import logging
import math
import typing as tp
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.recast_exceptions import rethrow_as
from satella.coding.typing import T
diff --git a/satella/coding/structures/singleton.py b/satella/coding/structures/singleton.py
index 1744c96995394bc47706604733f7f9b2501a78d2..527ee91217a00f4c99c4029667a24c8a3f3dd17d 100644
--- a/satella/coding/structures/singleton.py
+++ b/satella/coding/structures/singleton.py
@@ -1,7 +1,7 @@
import typing as tp
import weakref
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
# noinspection PyPep8Naming
diff --git a/satella/dao.py b/satella/dao.py
index 329cea3be6aa2814744a2ac281731c4542deffa0..8393246c31778d452cd4cb1956ff2ff8a283d993 100644
--- a/satella/dao.py
+++ b/satella/dao.py
@@ -1,6 +1,6 @@
from abc import ABCMeta, abstractmethod
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
__all__ = ['Loadable', 'must_be_loaded']
diff --git a/satella/instrumentation/metrics/metric_types/measurable_mixin.py b/satella/instrumentation/metrics/metric_types/measurable_mixin.py
index 5b3c519e736076044867175eed35c2aeb5957c9b..2ab73900c332b0392092f946694e92ec1f6e12b0 100644
--- a/satella/instrumentation/metrics/metric_types/measurable_mixin.py
+++ b/satella/instrumentation/metrics/metric_types/measurable_mixin.py
@@ -2,7 +2,7 @@ import inspect
import time
from concurrent.futures import Future
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import NoArgCallable
from .base import MetricLevel
diff --git a/satella/time/backoff.py b/satella/time/backoff.py
index 74c3a419d2d72c4500d54233f000b3c10fbb2a42..6b1b4f9d82d9fa07129a9e240c87446bd37d7ada 100644
--- a/satella/time/backoff.py
+++ b/satella/time/backoff.py
@@ -1,7 +1,7 @@
import time
import typing as tp
-from satella.coding.decorators.decorators import wraps
+from satella.coding.decorators.wraps import wraps
from satella.coding.typing import ExceptionList
from satella.exceptions import WouldWaitMore
from satella.time.measure import measure
diff --git a/tests/test_cassandra.py b/tests/test_cassandra.py
index 608eef3488f917fd24b18d9d65f587197b4af9c0..da929599c320f92f4e71bcbdda0ab5ed0d55f74f 100644
--- a/tests/test_cassandra.py
+++ b/tests/test_cassandra.py
@@ -1,4 +1,4 @@
-from satella.coding.concurrent import CallableGroup
+from satella.coding.concurrent import CallableGroup, IDAllocator
from satella.cassandra import parallel_for, wrap_future
import unittest
@@ -6,6 +6,7 @@ import unittest
class TestCassandra(unittest.TestCase):
def test_wrap_future(self):
+
class MockCassandraFuture:
def __init__(self):
self.value = None
diff --git a/tests/test_coding/test_concurrent.py b/tests/test_coding/test_concurrent.py
index a0c58a36d6c081e6c715a6a8f0f00135b01520a0..2b47e13d64b5b8aa63331f0adfcb9e5124d51227 100644
--- a/tests/test_coding/test_concurrent.py
+++ b/tests/test_coding/test_concurrent.py
@@ -270,7 +270,7 @@ class TestConcurrent(unittest.TestCase):
nonlocal a
a[4] += 1
- cbgroup.add_many(y, z)
+ cb = cbgroup.add_many(y, z)
cbgroup.add(CancellableCallback(p, one_shotted=False))
cbgroup()
self.assertEqual(a[1],3)
@@ -278,6 +278,9 @@ class TestConcurrent(unittest.TestCase):
self.assertEqual(a[4], 7)
cbgroup()
self.assertEqual(a[4], 8)
+ cb.cancel()
+ cbgroup()
+ self.assertEqual(a[4], 7)
def test_peekable_queue_put_many(self):
pkb = PeekableQueue()
@@ -765,8 +768,11 @@ class TestConcurrent(unittest.TestCase):
def test_cg_proforma(self):
cg = CallableGroup()
a = {}
- cg.add(lambda: a.__setitem__('test', 'value'))
- cg()
+ def cg_proforma(val):
+ nonlocal a
+ a['test'] = val
+ cg.add(cg_proforma)
+ cg('value')
self.assertEqual(a['test'], 'value')
def test_terminable_thread(self):
@@ -877,18 +883,18 @@ class TestConcurrent(unittest.TestCase):
def test_callable_group(self):
a = {
- 'a': False,
- 'b': False
+ 'a': 1
}
def op_f(what):
- return lambda: a.__setitem__(what, True)
+ what['a'] += 1
cg = CallableGroup()
- cg.add(op_f('a'))
- cg.add(op_f('b'))
+ cg.add(op_f)
+ self.assertRaises(RuntimeError, cg.add, op_f)
+ self.assertEqual(len(cg), 2)
- cg()
+ cg(a)
- self.assertTrue(all(a.values()))
+ self.assertEqual(a, {'a': 3})
diff --git a/tests/test_configuration/test_schema.py b/tests/test_configuration/test_schema.py
index e1cdb81b23a7b0d2c8627f712cb74214eaaaf188..0268baae60401fd7f820e74f3109a28b1df305a7 100644
--- a/tests/test_configuration/test_schema.py
+++ b/tests/test_configuration/test_schema.py
@@ -15,6 +15,11 @@ class Environment(enum.IntEnum):
class TestSchema(unittest.TestCase):
+ def setUp(self):
+ for file in ['test']:
+ if os.path.exists(file):
+ os.unlink(file)
+
def test_file(self):
schema = {
"key": "file"
diff --git a/tests/test_files.py b/tests/test_files.py
index 6206d6f02b99fa2f53df3c1688915be45869fff6..517eb0fd1f39d803f3cc4ed3f56fe29e636a58a7 100644
--- a/tests/test_files.py
+++ b/tests/test_files.py
@@ -1,5 +1,6 @@
import io
import os
+import uuid
from os.path import join
import tempfile
import unittest
@@ -114,15 +115,17 @@ class TestFiles(unittest.TestCase):
self.assertFalse(try_unlink('test.txt'))
def test_write_out_file_if_different(self):
- try:
- os.unlink('test')
- except FileNotFoundError:
- pass
- try:
- self.assertTrue(write_out_file_if_different('test', 'test', 'UTF-8'))
- self.assertFalse(write_out_file_if_different('test', 'test', 'UTF-8'))
- finally:
- os.unlink('test')
+ # Wait until Windows releases the file
+ while True:
+ try:
+ os.unlink('test')
+ except (PermissionError, FileNotFoundError):
+ pass
+ try:
+ self.assertTrue(write_out_file_if_different('test-'+uuid.uuid4().hex, 'test', 'UTF-8'))
+ self.assertFalse(write_out_file_if_different('test-'+uuid.uuid4().hex, 'test', 'UTF-8'))
+ finally:
+ os.unlink('test')
def test_read_in_and_write_to(self):
data = 'żażółć gęślą jaźń'