Merge branch 'release/3.3.0'

Author: wolph

_python_utils_tests/test_generators.py

@@ -0,0 +1,36 @@
+import pytest
+
+import python_utils
+
+
+@pytest.mark.asyncio
+async def test_abatcher():
+    async for batch in python_utils.abatcher(python_utils.acount(stop=9), 3):
+        assert len(batch) == 3
+
+    async for batch in python_utils.abatcher(python_utils.acount(stop=2), 3):
+        assert len(batch) == 2
+
+
+@pytest.mark.asyncio
+async def test_abatcher_timed():
+    batches = []
+    async for batch in python_utils.abatcher(
+        python_utils.acount(stop=10, delay=0.08),
+        interval=0.2
+    ):
+        batches.append(batch)
+
+    assert len(batches) == 3
+    assert sum(len(batch) for batch in batches) == 10
+
+
+def test_batcher():
+    batch = []
+    for batch in python_utils.batcher(range(9), 3):
+        assert len(batch) == 3
+
+    for batch in python_utils.batcher(range(4), 3):
+        pass
+
+    assert len(batch) == 1

_python_utils_tests/test_time.py

@@ -1,27 +1,28 @@
+import asyncio
 import itertools
 from datetime import timedelta
 
 import pytest
 
-from python_utils import aio
-from python_utils import time
+import python_utils
 
 
 @pytest.mark.parametrize(
     'timeout,interval,interval_multiplier,maximum_interval,iterable,result', [
-        (0.1, 0.06, 0.5, 0.1, aio.acount, 2),
-        (0.2, 0.06, 0.5, 0.1, aio.acount(), 4),
-        (0.3, 0.06, 1.0, None, aio.acount, 5),
+        (0.2, 0.1, 0.4, 0.2, python_utils.acount, 2),
+        (0.3, 0.1, 0.4, 0.2, python_utils.acount(), 3),
+        (0.3, 0.06, 1.0, None, python_utils.acount, 5),
         (timedelta(seconds=0.1), timedelta(seconds=0.06),
-         2.0, timedelta(seconds=0.1), aio.acount, 2),
+         2.0, timedelta(seconds=0.1), python_utils.acount, 2),
     ])
 @pytest.mark.asyncio
 async def test_aio_timeout_generator(timeout, interval, interval_multiplier,
                                      maximum_interval, iterable, result):
     i = None
-    async for i in time.aio_timeout_generator(
-            timeout, interval, iterable,
-            maximum_interval=maximum_interval):
+    async for i in python_utils.aio_timeout_generator(
+        timeout, interval, iterable,
+        maximum_interval=maximum_interval
+    ):
         pass
 
     assert i == result
@@ -41,13 +42,103 @@ async def test_aio_timeout_generator(timeout, interval, interval_multiplier,
 def test_timeout_generator(timeout, interval, interval_multiplier,
                            maximum_interval, iterable, result):
     i = None
-    for i in time.timeout_generator(
-            timeout=timeout,
-            interval=interval,
-            interval_multiplier=interval_multiplier,
-            iterable=iterable,
-            maximum_interval=maximum_interval,
+    for i in python_utils.timeout_generator(
+        timeout=timeout,
+        interval=interval,
+        interval_multiplier=interval_multiplier,
+        iterable=iterable,
+        maximum_interval=maximum_interval,
     ):
         pass
 
     assert i == result
+
+
+@pytest.mark.asyncio
+async def test_aio_generator_timeout_detector():
+    async def generator():
+        for i in range(10):
+            await asyncio.sleep(i / 100.0)
+            yield i
+
+    detector = python_utils.aio_generator_timeout_detector
+    # Test regular timeout with reraise
+    with pytest.raises(asyncio.TimeoutError):
+        async for i in detector(generator(), 0.05):
+            pass
+
+    # Test regular timeout with clean exit
+    async for i in detector(generator(), 0.05, on_timeout=None):
+        pass
+
+    assert i == 4
+
+    # Test total timeout with reraise
+    with pytest.raises(asyncio.TimeoutError):
+        async for i in detector(generator(), total_timeout=0.1):
+            pass
+
+    # Test total timeout with clean exit
+    async for i in detector(generator(), total_timeout=0.1, on_timeout=None):
+        pass
+
+    assert i == 4
+
+    # Test stop iteration
+    async for i in detector(generator(), on_timeout=None):
+        pass
+
+
+@pytest.mark.asyncio
+async def test_aio_generator_timeout_detector_decorator():
+    # Test regular timeout with reraise
+    @python_utils.aio_generator_timeout_detector_decorator(timeout=0.05)
+    async def generator():
+        for i in range(10):
+            await asyncio.sleep(i / 100.0)
+            yield i
+
+    with pytest.raises(asyncio.TimeoutError):
+        async for i in generator():
+            pass
+
+    # Test regular timeout with clean exit
+    @python_utils.aio_generator_timeout_detector_decorator(
+        timeout=0.05,
+        on_timeout=None
+    )
+    async def generator():
+        for i in range(10):
+            await asyncio.sleep(i / 100.0)
+            yield i
+
+    async for i in generator():
+        pass
+
+    assert i == 4
+
+    # Test total timeout with reraise
+    @python_utils.aio_generator_timeout_detector_decorator(total_timeout=0.1)
+    async def generator():
+        for i in range(10):
+            await asyncio.sleep(i / 100.0)
+            yield i
+
+    with pytest.raises(asyncio.TimeoutError):
+        async for i in generator():
+            pass
+
+    # Test total timeout with clean exit
+    @python_utils.aio_generator_timeout_detector_decorator(
+        total_timeout=0.1,
+        on_timeout=None
+    )
+    async def generator():
+        for i in range(10):
+            await asyncio.sleep(i / 100.0)
+            yield i
+
+    async for i in generator():
+        pass
+
+    assert i == 4

python_utils/__about__.py

@@ -6,4 +6,4 @@
     'with the standard Python install')
 __url__: str = 'https://github.com/WoLpH/python-utils'
 # Omit type info due to automatic versioning script
-__version__ = '3.2.3'
+__version__ = '3.3.0'

python_utils/__init__.py

@@ -1,35 +1,39 @@
-from . import aio
-from . import compat
-from . import converters
-from . import decorators
-from . import formatters
-from . import import_
-from . import logger
-from . import terminal
-from . import time
-from . import types
-
+from . import (
+    aio,
+    compat,
+    converters,
+    decorators,
+    formatters,
+    generators,
+    import_,
+    logger,
+    terminal,
+    time,
+    types,
+)
 from .aio import acount
-from .converters import remap
-from .converters import scale_1024
-from .converters import to_float
-from .converters import to_int
-from .converters import to_str
-from .converters import to_unicode
-from .decorators import listify
-from .decorators import set_attributes
-from .formatters import camel_to_underscore
-from .formatters import timesince
+from .converters import remap, scale_1024, to_float, to_int, to_str, to_unicode
+from .decorators import listify, set_attributes
+from .exceptions import raise_exception, reraise
+from .formatters import camel_to_underscore, timesince
+from .generators import abatcher, batcher
 from .import_ import import_global
-from .terminal import get_terminal_size
-from .time import format_time
-from .time import timedelta_to_seconds
-from .time import timeout_generator
-from .time import aio_timeout_generator
 from .logger import Logged, LoggerBase
+from .terminal import get_terminal_size
+from .time import (
+    aio_generator_timeout_detector,
+    aio_generator_timeout_detector_decorator,
+    aio_timeout_generator,
+    delta_to_seconds,
+    delta_to_seconds_or_none,
+    format_time,
+    timedelta_to_seconds,
+    timeout_generator,
+)
 
 __all__ = [
     'aio',
+    'generators',
     'compat',
     'converters',
     'decorators',
@@ -55,7 +59,15 @@
     'format_time',
     'timeout_generator',
     'acount',
+    'abatcher',
+    'batcher',
     'aio_timeout_generator',
+    'aio_generator_timeout_detector_decorator',
+    'aio_generator_timeout_detector',
+    'delta_to_seconds',
+    'delta_to_seconds_or_none',
+    'reraise',
+    'raise_exception',
     'Logged',
     'LoggerBase',
 ]

python_utils/aio.py

@@ -1,9 +1,16 @@
+'''
+Asyncio equivalents to regular Python functions.
+
+'''
 import asyncio
 import itertools
 
 
-async def acount(start=0, step=1, delay=0):
+async def acount(start=0, step=1, delay=0, stop=None):
     '''Asyncio version of itertools.count()'''
     for item in itertools.count(start, step):  # pragma: no branch
+        if stop is not None and item >= stop:
+            break
+
         yield item
         await asyncio.sleep(delay)

python_utils/converters.py

@@ -83,7 +83,7 @@ def to_int(
         raise TypeError('unknown argument for regexp parameter: %r' % regexp)
 
     try:
-        if regexp:
+        if regexp and input_:
             match = regexp.search(input_)
             if match:
                 input_ = match.groups()[-1]

python_utils/exceptions.py

@@ -0,0 +1,26 @@
+import typing
+
+
+def raise_exception(
+    exception_class: typing.Type[Exception],
+    *args: typing.Any,
+    **kwargs: typing.Any,
+) -> typing.Callable:
+    '''
+    Returns a function that raises an exception of the given type with the
+    given arguments.
+
+    >>> raise_exception(ValueError, 'spam')('eggs')
+    Traceback (most recent call last):
+        ...
+    ValueError: spam
+    '''
+
+    def raise_(*args_: typing.Any, **kwargs_: typing.Any) -> typing.Any:
+        raise exception_class(*args, **kwargs)
+
+    return raise_
+
+
+def reraise(*args: typing.Any, **kwargs: typing.Any) -> typing.Any:
+    raise

python_utils/generators.py

@@ -0,0 +1,62 @@
+import time
+
+import python_utils
+from python_utils import types
+
+
+async def abatcher(
+    generator: types.AsyncGenerator,
+    batch_size: types.Optional[int] = None,
+    interval: types.Optional[types.delta_type] = None,
+):
+    '''
+    Asyncio generator wrapper that returns items with a given batch size or
+    interval (whichever is reached first).
+    '''
+    batch: list = []
+
+    assert batch_size or interval, 'Must specify either batch_size or interval'
+
+    if interval:
+        interval_s = python_utils.delta_to_seconds(interval)
+        next_yield = time.perf_counter() + interval_s
+    else:
+        interval_s = 0
+        next_yield = 0
+
+    while True:
+        try:
+            item = await generator.__anext__()
+        except StopAsyncIteration:
+            if batch:
+                yield batch
+            break
+        else:
+            batch.append(item)
+
+        if batch_size is not None and len(batch) == batch_size:
+            yield batch
+            batch = []
+
+        if interval and batch and time.perf_counter() > next_yield:
+            yield batch
+            batch = []
+            # Always set the next yield time to the current time. If the
+            # loop is running slow due to blocking functions we do not
+            # want to burst too much
+            next_yield = time.perf_counter() + interval_s
+
+
+def batcher(iterable, batch_size):
+    '''
+    Generator wrapper that returns items with a given batch size
+    '''
+    batch = []
+    for item in iterable:
+        batch.append(item)
+        if len(batch) == batch_size:
+            yield batch
+            batch = []
+
+    if batch:
+        yield batch

python_utils/logger.py

@@ -88,7 +88,7 @@ class Logged(LoggerBase):
     'spam'
     '''
 
-    logger: logging.Logger
+    logger: logging.Logger  # pragma: no cover
 
     @classmethod
     def __get_name(cls, *name_parts: str) -> str: