small fixes to type annotations

added topic class

.gitea/workflows/build.yaml

@@ -1,25 +1,29 @@
 name: CI
 on:
   push:
-    branches: [ master ]
+    tags:
+      - '*'
 jobs:
   build:
     runs-on: woryzen
     steps:
       - name: Checkout sources
         uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          fetch-tags: true
       - uses: actions/setup-python@v5
         with:
           cache: 'pip'
       - name: Create virtualenv
         run: |
           python -m venv .venv
-          .venv/bin/pip install -r requirements.txt
+          .venv/bin/pip install -r requirements-dev.txt
       - name: Run unit tests
         run: .venv/bin/python -m unittest discover -s tests
       - name: Execute build
         run: |
-          .venv/bin/python -m build
+          .venv/bin/pyproject-build
       - name: Publish artifacts
         env:
           TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}

.gitignore

@@ -1,7 +1,8 @@
-dist
 .idea
 __pycache__
 .venv
 *.pyc
 *.egg-info
-
+/build
+/dist
+src/pwo/_version.py

pyproject.toml

@@ -1,10 +1,10 @@
 [build-system]
-requires = ["setuptools>=61.0"]
+requires = ["setuptools>=61.0", "setuptools-scm>=8"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "pwo"
-version = "0.0.2"
+dynamic = ["version"]
 authors = [
   { name="Walter Oggioni", email="oggioni.walter@gmail.com" },
 ]
@@ -17,7 +17,12 @@ classifiers = [
     "Operating System :: OS Independent",
 ]
 dependencies = [
-    'typing_extensions==4.7.1'
+    'typing_extensions'
+]
+
+[project.optional-dependencies]
+dev = [
+    "pip-tools", "build", "mypy", "ipdb", "twine"
 ]
 
 [project.urls]
@@ -35,4 +40,7 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unused_ignores = true
 exclude = ["scripts", "docs", "test"]
-strict = true
+strict = true
+
+[tool.setuptools_scm]
+version_file = "src/pwo/_version.py"

requirements-dev.txt

@@ -0,0 +1,140 @@
+#
+# This file is autogenerated by pip-compile with Python 3.12
+# by the following command:
+#
+#    pip-compile --extra=dev --output-file=requirements-dev.txt pyproject.toml
+#
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
+--extra-index-url https://pypi.org/simple
+
+asttokens==2.4.1
+    # via stack-data
+build==1.2.2.post1
+    # via
+    #   pip-tools
+    #   pwo (pyproject.toml)
+certifi==2024.8.30
+    # via requests
+cffi==1.17.1
+    # via cryptography
+charset-normalizer==3.4.0
+    # via requests
+click==8.1.7
+    # via pip-tools
+cryptography==43.0.3
+    # via secretstorage
+decorator==5.1.1
+    # via
+    #   ipdb
+    #   ipython
+docutils==0.21.2
+    # via readme-renderer
+executing==2.1.0
+    # via stack-data
+idna==3.10
+    # via requests
+importlib-metadata==8.5.0
+    # via twine
+ipdb==0.13.13
+    # via pwo (pyproject.toml)
+ipython==8.28.0
+    # via ipdb
+jaraco-classes==3.4.0
+    # via keyring
+jaraco-context==6.0.1
+    # via keyring
+jaraco-functools==4.1.0
+    # via keyring
+jedi==0.19.1
+    # via ipython
+jeepney==0.8.0
+    # via
+    #   keyring
+    #   secretstorage
+keyring==25.4.1
+    # via twine
+markdown-it-py==3.0.0
+    # via rich
+matplotlib-inline==0.1.7
+    # via ipython
+mdurl==0.1.2
+    # via markdown-it-py
+more-itertools==10.5.0
+    # via
+    #   jaraco-classes
+    #   jaraco-functools
+mypy==1.13.0
+    # via pwo (pyproject.toml)
+mypy-extensions==1.0.0
+    # via mypy
+nh3==0.2.18
+    # via readme-renderer
+packaging==24.1
+    # via build
+parso==0.8.4
+    # via jedi
+pexpect==4.9.0
+    # via ipython
+pip-tools==7.4.1
+    # via pwo (pyproject.toml)
+pkginfo==1.10.0
+    # via twine
+prompt-toolkit==3.0.48
+    # via ipython
+ptyprocess==0.7.0
+    # via pexpect
+pure-eval==0.2.3
+    # via stack-data
+pycparser==2.22
+    # via cffi
+pygments==2.18.0
+    # via
+    #   ipython
+    #   readme-renderer
+    #   rich
+pyproject-hooks==1.2.0
+    # via
+    #   build
+    #   pip-tools
+readme-renderer==44.0
+    # via twine
+requests==2.32.3
+    # via
+    #   requests-toolbelt
+    #   twine
+requests-toolbelt==1.0.0
+    # via twine
+rfc3986==2.0.0
+    # via twine
+rich==13.9.3
+    # via twine
+secretstorage==3.3.3
+    # via keyring
+six==1.16.0
+    # via asttokens
+stack-data==0.6.3
+    # via ipython
+traitlets==5.14.3
+    # via
+    #   ipython
+    #   matplotlib-inline
+twine==5.1.1
+    # via pwo (pyproject.toml)
+typing-extensions==4.12.2
+    # via
+    #   mypy
+    #   pwo (pyproject.toml)
+urllib3==2.2.3
+    # via
+    #   requests
+    #   twine
+wcwidth==0.2.13
+    # via prompt-toolkit
+wheel==0.44.0
+    # via pip-tools
+zipp==3.20.2
+    # via importlib-metadata
+
+# The following packages are considered to be unsafe in a requirements file:
+# pip
+# setuptools

requirements.txt

@@ -1,34 +1,11 @@
-build==1.2.1
+#
-certifi==2024.6.2
+# This file is autogenerated by pip-compile with Python 3.12
-cffi==1.16.0
+# by the following command:
-charset-normalizer==3.3.2
+#
-cryptography==42.0.8
+#    pip-compile --output-file=requirements.txt pyproject.toml
-docutils==0.21.2
+#
-idna==3.7
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
-importlib_metadata==7.2.0
+--extra-index-url https://pypi.org/simple
-jaraco.classes==3.4.0
+
-jaraco.context==5.3.0
+typing-extensions==4.12.2
-jaraco.functools==4.0.1
+    # via pwo (pyproject.toml)
-jeepney==0.8.0
-keyring==25.2.1
-markdown-it-py==3.0.0
-mdurl==0.1.2
-more-itertools==10.3.0
-mypy==1.10.0
-mypy-extensions==1.0.0
-nh3==0.2.17
-packaging==24.1
-pkginfo==1.11.1
-pycparser==2.22
-Pygments==2.18.0
-pyproject_hooks==1.1.0
-readme_renderer==43.0
-requests==2.32.3
-requests-toolbelt==1.0.0
-rfc3986==2.0.0
-rich==13.7.1
-SecretStorage==3.3.3
-twine==5.1.0
-typing_extensions==4.12.2
-urllib3==2.2.2
-zipp==3.19.2

src/pwo/__init__.py

@@ -1,5 +1,16 @@
-from .private import format_filesize, async_retry, retry, async_test, ExceptionHandlerOutcome
+from .private import (
+    format_filesize,
+    async_retry,
+    retry,
+    async_test,
+    ExceptionHandlerOutcome,
+    tmpdir,
+    decorator_with_kwargs,
+    classproperty,
+    AsyncQueueIterator
+)
 from .maybe import Maybe
+from .notification import TopicManager, Subscriber
 
 __all__ = [
     'format_filesize',
@@ -7,5 +18,11 @@ __all__ = [
     'retry',
     'async_test',
     'ExceptionHandlerOutcome',
-    'Maybe'
+    'Maybe',
+    'tmpdir',
+    'decorator_with_kwargs',
+    'classproperty',
+    'TopicManager',
+    'Subscriber',
+    'AsyncQueueIterator'
 ]

src/pwo/maybe.py

@@ -24,11 +24,10 @@ class Maybe(Generic[T]):
 
     @property
     def value(self) -> T:
-        value = self._value
+        if self.is_empty:
-        if not value:
             raise ValueError('Empty Maybe')
         else:
-            return value
+            return self._value
 
     @property
     def is_present(self) -> bool:
@@ -54,6 +53,7 @@ class Maybe(Generic[T]):
 
     def __or__(self, alt: Maybe[T]) -> Maybe[T]:
         return self if self.is_present else alt
+
     def or_else(self, alt: T) -> T:
         return self.value if self.is_present else alt
 
@@ -66,7 +66,7 @@ class Maybe(Generic[T]):
         else:
             raise supplier()
 
-    def or_else_get(self, supplier: Callable[[], T]) -> Maybe[T]:
+    def or_else_get(self, supplier: Callable[[], Optional[T]]) -> Maybe[T]:
         return self if self.is_present else Maybe.of_nullable(supplier())
 
     def if_present(self, callback: Callable[[T], U]) -> None:

src/pwo/notification.py

@@ -0,0 +1,91 @@
+from .private import AsyncQueueIterator
+from asyncio import Queue, AbstractEventLoop, Future, CancelledError
+from typing import Callable, Optional
+from logging import getLogger
+
+log = getLogger(__name__)
+
+
+class Subscriber:
+    _unsubscribe_callback: Callable[['Subscriber'], None]
+    _event: Optional[Future]
+    _loop: AbstractEventLoop
+
+    def __init__(self, unsubscribe: Callable[['Subscriber'], None], loop: AbstractEventLoop):
+        self._unsubscribe_callback = unsubscribe
+        self._event: Optional[Future] = None
+        self._loop = loop
+
+    def unsubscribe(self) -> None:
+        self._event.cancel()
+        self._unsubscribe_callback(self)
+        log.debug('Deleted subscriber %s', id(self))
+
+    async def wait(self, tout: float) -> bool:
+        self._event = self._loop.create_future()
+
+        def callback():
+            if not self._event.done():
+                self._event.set_result(False)
+
+        handle = self._loop.call_later(tout, callback)
+        try:
+            log.debug('Subscriber %s is waiting for an event', id(self))
+            return await self._event
+        except CancelledError:
+            return False
+        finally:
+            handle.cancel()
+
+    def notify(self) -> None:
+        log.debug('Subscriber %s notified', id(self))
+        if not self._event.done():
+            self._event.set_result(True)
+
+    def reset(self) -> None:
+        self._event = self._loop.create_future()
+
+
+class TopicManager:
+    _loop: AbstractEventLoop
+    _queue: Queue
+    _subscribers: dict[str, set[Subscriber]]
+
+    def __init__(self, loop: AbstractEventLoop):
+        self._subscribers: dict[str, set[Subscriber]] = dict()
+        self._loop = loop
+        self._queue = Queue()
+
+    def subscribe(self, topic: str) -> Subscriber:
+        subscriptions = self._subscribers
+        subscriptions_per_topic = subscriptions.setdefault(topic, set())
+
+        def unsubscribe_callback(subscription):
+            subscriptions_per_topic.remove(subscription)
+            log.debug('Unsubscribed %s from topic %s', id(result), topic)
+
+        result = Subscriber(unsubscribe_callback, self._loop)
+        log.debug('Created subscriber %s to topic %s', id(result), topic)
+        subscriptions_per_topic.add(result)
+        return result
+
+    def _notify_subscriptions(self, topic):
+        subscriptions = self._subscribers
+        subscriptions_per_topic = subscriptions.get(topic, None)
+        if subscriptions_per_topic:
+            log.debug(f"Subscribers on '{topic}': {len(subscriptions_per_topic)}")
+            for s in subscriptions_per_topic:
+                s.notify()
+
+    async def process_events(self):
+        async for evt in AsyncQueueIterator(self._queue):
+            log.debug(f"Processed event for topic '{evt}'")
+            self._notify_subscriptions(evt)
+        log.debug(f"Event processor has completed")
+
+    def post_event(self, topic):
+        def callback():
+            self._queue.put_nowait(topic)
+            log.debug(f"Posted event for topic '{topic}', queue size: {self._queue.qsize()}")
+
+        self._loop.call_soon_threadsafe(callback)

src/pwo/private.py

@@ -1,8 +1,78 @@
 import math
+from asyncio import sleep as async_sleep, Runner, Queue
 from enum import Enum, auto
-from typing import Callable
+from functools import wraps, partial
+from inspect import signature
+from pathlib import Path
+from tempfile import TemporaryDirectory
 from time import sleep
-from asyncio import sleep as async_sleep, Runner
+from typing import Callable, AsyncIterator
+
+
+def decorator_with_kwargs(decorator: Callable) -> Callable:
+    """Decorator factory to give decorated decorators the skill to receive
+    optional keyword arguments.
+
+    If a decorator "some_decorator" is decorated with this function:
+
+        @decorator_with_kwargs
+        def some_decorator(decorated_function, kwarg1=1, kwarg2=2):
+            def wrapper(*decorated_function_args, **decorated_function_kwargs):
+                '''Modifies the behavior of decorated_function according
+                to the value of kwarg1 and kwarg2'''
+                ...
+            return wrapper
+
+    It will be usable in the following ways:
+
+        @some_decorator
+        def func(x):
+            ...
+
+        @some_decorator()
+        def func(x):
+            ...
+
+        @some_decorator(kwarg1=3)  # or other combinations of kwargs
+        def func(x, y):
+            ...
+
+    :param decorator: decorator to be given optional kwargs-handling skills
+    :type decorator: Callable
+    :raises TypeError: if the decorator does not receive a single Callable or
+        keyword arguments
+    :raises TypeError: if the signature of the decorated decorator does not
+        conform to: Callable, **keyword_arguments
+    :return: modified decorator
+    :rtype: Callable
+    """
+
+    @wraps(decorator)
+    def decorator_wrapper(*args, **kwargs):
+        if (len(kwargs) == 0) and (len(args) == 1) and callable(args[0]):
+            return decorator(args[0])
+        if len(args) == 0:
+            return partial(decorator, **kwargs)
+        raise TypeError(
+            f'{decorator.__name__} expects either a single Callable '
+            'or keyword arguments'
+        )
+
+    signature_values = signature(decorator).parameters.values()
+    signature_args = [
+        param.name for param in signature_values
+        if param.default == param.empty
+    ]
+
+    if len(signature_args) != 1:
+        raise TypeError(
+            f'{decorator.__name__} signature should be of the form:\n'
+            f'{decorator.__name__}(function: typing.Callable, '
+            'kwarg_1=default_1, kwarg_2=default_2, ...) -> Callable'
+        )
+
+    return decorator_wrapper
+
 
 _size_uoms = ('B', 'KiB', 'MiB', 'GiB', 'KiB')
 
@@ -17,66 +87,138 @@ def format_filesize(size: int) -> str:
     return '%.2f ' % (size / math.pow(1024, counter)) + _size_uoms[counter]
 
 
-
 class ExceptionHandlerOutcome(Enum):
     THROW = auto()
     CONTINUE = auto()
 
 
+@decorator_with_kwargs
 def retry(
+        function,
         max_attempts: int = 3,
         multiplier: float = 2,
         initial_delay: float = 1.0,
         exception_handler: Callable[[Exception], ExceptionHandlerOutcome] =
         lambda _: ExceptionHandlerOutcome.CONTINUE
 ):
-    def wrapper(function):
+    @wraps(function)
-        def result(*args, **kwargs):
+    def result(*args, **kwargs):
-            attempts = 0
+        attempts = 0
-            delay = initial_delay
+        delay = initial_delay
-            while True:
+        while True:
-                attempts += 1
+            attempts += 1
-                try:
+            try:
-                    return function(*args, **kwargs)
+                return function(*args, **kwargs)
-                except Exception as ex:
+            except Exception as ex:
-                    if attempts < max_attempts and exception_handler(ex) == ExceptionHandlerOutcome.CONTINUE:
+                if attempts < max_attempts and exception_handler(ex) == ExceptionHandlerOutcome.CONTINUE:
-                        sleep(delay)
+                    sleep(delay)
-                        delay *= multiplier
+                    delay *= multiplier
-                    else:
+                else:
-                        raise ex
+                    raise ex
 
-        return result
+    return result
-
-    return wrapper
 
 
+@decorator_with_kwargs
 def async_retry(
+        function,
         max_attempts: int = 3,
         multiplier: float = 2,
         initial_delay: float = 1.0,
         exception_handler=lambda _: ExceptionHandlerOutcome.CONTINUE
 ):
-    def wrapper(function):
+    @wraps(function)
-        async def result(*args, **kwargs):
+    async def result(*args, **kwargs):
-            attempts = 0
+        attempts = 0
-            delay = initial_delay
+        delay = initial_delay
-            while True:
+        while True:
-                attempts += 1
+            attempts += 1
-                try:
+            try:
-                    return await function(*args, **kwargs)
+                return await function(*args, **kwargs)
-                except Exception as ex:
+            except Exception as ex:
-                    if attempts < max_attempts and exception_handler(ex) == ExceptionHandlerOutcome.CONTINUE:
+                if attempts < max_attempts and exception_handler(ex) == ExceptionHandlerOutcome.CONTINUE:
-                        await async_sleep(delay)
+                    await async_sleep(delay)
-                        delay *= multiplier
+                    delay *= multiplier
-                    else:
+                else:
-                        raise ex
+                    raise ex
 
-        return result
+    return result
 
-    return wrapper
 
 def async_test(coro):
+    @wraps(coro)
     def wrapper(*args, **kwargs):
         with Runner() as runner:
             runner.run(coro(*args, **kwargs))
-    return wrapper
+
+    return wrapper
+
+
+@decorator_with_kwargs
+def tmpdir(f,
+           argument_name='temp_dir',
+           suffix=None,
+           prefix=None,
+           dir=None,
+           ignore_cleanup_errors=False,
+           delete=True):
+    @wraps(f)
+    def result(*args, **kwargs):
+        with TemporaryDirectory(
+                suffix=suffix,
+                prefix=prefix,
+                dir=dir,
+                ignore_cleanup_errors=ignore_cleanup_errors,
+                delete=delete) as temp_dir:
+            f(*args, **kwargs, **{
+                argument_name: Path(temp_dir)
+            })
+
+    return result
+
+
+class ClassPropertyDescriptor:
+
+    def __init__(self, fget, fset=None):
+        self.fget = fget
+        self.fset = fset
+
+    def __get__(self, obj, klass=None):
+        if klass is None:
+            klass = type(obj)
+        return self.fget.__get__(obj, klass)()
+
+    def __set__(self, obj, value):
+        if not self.fset:
+            raise AttributeError("can't set attribute")
+        type_ = type(obj)
+        return self.fset.__get__(obj, type_)(value)
+
+    def setter(self, func):
+        if not isinstance(func, (classmethod, staticmethod)):
+            func = classmethod(func)
+        self.fset = func
+        return self
+
+
+def classproperty(func):
+    if not isinstance(func, (classmethod, staticmethod)):
+        func = classmethod(func)
+    return ClassPropertyDescriptor(func)
+
+
+class AsyncQueueIterator[T]:
+    _queue: Queue[T]
+
+    def __init__(self, queue: Queue[T]):
+        self._queue = queue
+
+    def __aiter__(self) -> AsyncIterator[T]:
+        return self
+
+    async def __anext__(self) -> [T]:
+        item = await self._queue.get()
+        if item is None:
+            raise StopAsyncIteration
+        return item
+