improved format_size function

added topic class

.gitea/workflows/build.yaml

@@ -1,25 +1,31 @@
 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.txt .
+      - name: Run mypy
+        run: .venv/bin/python -m mypy -p src
       - 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]
@@ -36,3 +41,6 @@ warn_return_any = true
 warn_unused_ignores = true
 exclude = ["scripts", "docs", "test"]
 strict = true
+
+[tool.setuptools_scm]
+version_file = "src/pwo/_version.py"

requirements.txt

@@ -1,34 +1,147 @@
-build==1.2.1
-certifi==2024.6.2
-cffi==1.16.0
-charset-normalizer==3.3.2
-cryptography==42.0.8
-docutils==0.21.2
-idna==3.7
-importlib_metadata==7.2.0
-jaraco.classes==3.4.0
-jaraco.context==5.3.0
-jaraco.functools==4.0.1
-jeepney==0.8.0
-keyring==25.2.1
-markdown-it-py==3.0.0
+#
+# This file is autogenerated by pip-compile with Python 3.14
+# by the following command:
+#
+#    pip-compile --allow-unsafe --extra=dev --output-file=requirements.txt pyproject.toml
+#
+--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
+--extra-index-url https://pypi.org/simple
+
+asttokens==3.0.1
+    # via stack-data
+build==1.4.3
+    # via
+    #   pip-tools
+    #   pwo (pyproject.toml)
+certifi==2026.2.25
+    # via requests
+cffi==2.0.0
+    # via cryptography
+charset-normalizer==3.4.7
+    # via requests
+click==8.3.2
+    # via pip-tools
+cryptography==46.0.7
+    # via secretstorage
+decorator==5.2.1
+    # via
+    #   ipdb
+    #   ipython
+docutils==0.22.4
+    # via readme-renderer
+executing==2.2.1
+    # via stack-data
+id==1.6.1
+    # via twine
+idna==3.11
+    # via requests
+ipdb==0.13.13
+    # via pwo (pyproject.toml)
+ipython==9.12.0
+    # via ipdb
+ipython-pygments-lexers==1.1.1
+    # via ipython
+jaraco-classes==3.4.0
+    # via keyring
+jaraco-context==6.1.2
+    # via keyring
+jaraco-functools==4.4.0
+    # via keyring
+jedi==0.19.2
+    # via ipython
+jeepney==0.9.0
+    # via
+    #   keyring
+    #   secretstorage
+keyring==25.7.0
+    # via twine
+librt==0.9.0
+    # via mypy
+markdown-it-py==4.0.0
+    # via rich
+matplotlib-inline==0.2.1
+    # via ipython
 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
+    # via markdown-it-py
+more-itertools==11.0.2
+    # via
+    #   jaraco-classes
+    #   jaraco-functools
+mypy==1.20.1
+    # via pwo (pyproject.toml)
+mypy-extensions==1.1.0
+    # via mypy
+nh3==0.3.4
+    # via readme-renderer
+packaging==26.1
+    # via
+    #   build
+    #   twine
+    #   wheel
+parso==0.8.6
+    # via jedi
+pathspec==1.0.4
+    # via mypy
+pexpect==4.9.0
+    # via ipython
+pip-tools==7.5.3
+    # via pwo (pyproject.toml)
+prompt-toolkit==3.0.52
+    # via ipython
+ptyprocess==0.7.0
+    # via pexpect
+pure-eval==0.2.3
+    # via stack-data
+pycparser==3.0
+    # via cffi
+pygments==2.20.0
+    # via
+    #   ipython
+    #   ipython-pygments-lexers
+    #   readme-renderer
+    #   rich
+pyproject-hooks==1.2.0
+    # via
+    #   build
+    #   pip-tools
+readme-renderer==44.0
+    # via twine
+requests==2.33.1
+    # via
+    #   requests-toolbelt
+    #   twine
 requests-toolbelt==1.0.0
+    # via twine
 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
+    # via twine
+rich==15.0.0
+    # via twine
+secretstorage==3.5.0
+    # via keyring
+stack-data==0.6.3
+    # via ipython
+traitlets==5.14.3
+    # via
+    #   ipython
+    #   matplotlib-inline
+twine==6.2.0
+    # via pwo (pyproject.toml)
+typing-extensions==4.15.0
+    # via
+    #   mypy
+    #   pwo (pyproject.toml)
+urllib3==2.6.3
+    # via
+    #   id
+    #   requests
+    #   twine
+wcwidth==0.6.0
+    # via prompt-toolkit
+wheel==0.46.3
+    # via pip-tools
+
+# The following packages are considered to be unsafe in a requirements file:
+pip==26.0.1
+    # via pip-tools
+setuptools==82.0.1
+    # via pip-tools

src/pwo/__init__.py

@@ -1,9 +1,34 @@
-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,
+    aenumerate,
+    index_of_with_escape
+)
+from .maybe import Maybe
+from .notification import TopicManager, Subscriber
+from ._try import Try
 
 __all__ = [
     'format_filesize',
     'async_retry',
     'retry',
     'async_test',
-    'ExceptionHandlerOutcome'
+    'ExceptionHandlerOutcome',
+    'Maybe',
+    'tmpdir',
+    'decorator_with_kwargs',
+    'classproperty',
+    'TopicManager',
+    'Subscriber',
+    'AsyncQueueIterator',
+    'aenumerate',
+    'Try',
+    'index_of_with_escape'
 ]

src/pwo/_try.py

@@ -0,0 +1,52 @@
+from typing import (
+    Callable,
+    TypeVar,
+    Optional
+)
+
+ERR = TypeVar("ERR", bound=Exception)
+
+
+class Try[T]:
+    value: T | Exception
+
+    def __init__(self, value: T | Exception):
+        self.value = value
+
+    def handle[U](self, cb: Callable[[Optional[T], Optional[Exception]], U]) -> 'Try[U]':
+        value = self.value
+        if isinstance(value, Exception):
+            return Try.of(lambda: cb(None, value))
+        else:
+            return Try.of(lambda: cb(value, None))
+
+    def get(self, alternative: Optional[T] = None) -> T:
+        if isinstance(self.value, Exception):
+            if alternative is None:
+                raise self.value
+            else:
+                return alternative
+        else:
+            return self.value
+
+    def then_try[U](self, cb: Callable[[T], U]) -> 'Try[U]':
+        value = self.value
+        if isinstance(value, Exception):
+            return Try.failure(value)
+        else:
+            return Try.of(lambda: cb(value))
+
+    @staticmethod
+    def success[U](value: U) -> 'Try[U]':
+        return Try(value)
+
+    @staticmethod
+    def failure[U](ex: Exception) -> 'Try[U]':
+        return Try(ex)
+
+    @staticmethod
+    def of[U](cb: Callable[[], U]) -> 'Try[U]':
+        try:
+            return Try(cb())
+        except Exception as ex:
+            return Try(ex)

src/pwo/maybe.py

@@ -24,11 +24,11 @@ class Maybe(Generic[T]):
 
     @property
     def value(self) -> T:
-        value = self._value
-        if not value:
+        result = self._value
+        if result is None:
             raise ValueError('Empty Maybe')
         else:
-            return value
+            return result
 
     @property
     def is_present(self) -> bool:
@@ -52,6 +52,9 @@ class Maybe(Generic[T]):
     def flat_map(self, transformer: Callable[[T], Maybe[U]]) -> Maybe[U]:
         return transformer(self.value) if self.is_present else Maybe.empty()
 
+    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
 
@@ -64,7 +67,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, timeout
+from typing import Callable, Optional
+from logging import getLogger
+
+log = getLogger(__name__)
+
+
+class Subscriber:
+    _unsubscribe_callback: Callable[['Subscriber'], None]
+    _event: Optional[Future[bool]]
+    _loop: AbstractEventLoop
+
+    def __init__(self, unsubscribe: Callable[['Subscriber'], None], loop: AbstractEventLoop):
+        self._unsubscribe_callback = unsubscribe
+        self._event: Optional[Future[bool]] = None
+        self._loop = loop
+
+    def unsubscribe(self) -> None:
+        evt = self._event
+        if evt is not None:
+            evt.cancel()
+        self._unsubscribe_callback(self)
+        log.debug('Deleted subscriber %s', id(self))
+
+    async def wait(self, tout: Optional[float]) -> bool:
+
+        future: Future[bool] = self._loop.create_future()
+        self._event = future
+        try:
+            async with timeout(tout):
+                log.debug('Subscriber %s is waiting for an event', id(self))
+                return await future
+        except TimeoutError:
+            return False
+
+    def notify(self) -> None:
+        log.debug('Subscriber %s notified', id(self))
+        evt = self._event
+        if evt is None:
+            raise ValueError('Event is None')
+        if not evt.done():
+            evt.set_result(True)
+
+    def reset(self) -> None:
+        self._event = self._loop.create_future()
+
+
+class TopicManager:
+    _loop: AbstractEventLoop
+    _queue: Queue[Optional[str]]
+    _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: Subscriber) -> None:
+            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: str) -> None:
+        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) -> None:
+        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: str) -> None:
+        def callback() -> None:
+            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,21 +1,102 @@
 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,
+    Optional,
+    Self,
+    AsyncIterable,
+    Awaitable,
+    Any,
+    Coroutine,
+    Never,
+    Tuple,
+    no_type_check
+)
+
+
+def decorator_with_kwargs(decorator: Callable[..., Any]) -> Callable[..., Any]:
+    """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) -> Any:  # type: ignore
+        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')
 
 
-def format_filesize(size: int) -> str:
+def format_filesize(size: int, width: int = 5, decimals : int = 2) -> str:
     counter = 0
     tmp_size = size
     while tmp_size > 0:
         tmp_size //= 1024
         counter += 1
     counter -= 1
-    return '%.2f ' % (size / math.pow(1024, counter)) + _size_uoms[counter]
-
+    return f'%{width}.{decimals}f ' % (size / math.pow(1024, counter)) + _size_uoms[counter]
 
 
 class ExceptionHandlerOutcome(Enum):
@@ -23,15 +104,17 @@ class ExceptionHandlerOutcome(Enum):
     CONTINUE = auto()
 
 
+@decorator_with_kwargs
 def retry(
+        function: Callable[..., Any],
         max_attempts: int = 3,
         multiplier: float = 2,
         initial_delay: float = 1.0,
         exception_handler: Callable[[Exception], ExceptionHandlerOutcome] =
         lambda _: ExceptionHandlerOutcome.CONTINUE
-):
-    def wrapper(function):
-        def result(*args, **kwargs):
+) -> Callable[..., Any]:
+    @wraps(function)
+    def result(*args: Any, **kwargs: Any) -> Any:
         attempts = 0
         delay = initial_delay
         while True:
@@ -47,17 +130,17 @@ def retry(
 
     return result
 
-    return wrapper
-
 
+@decorator_with_kwargs
 def async_retry(
+        function: Callable[..., Any],
         max_attempts: int = 3,
         multiplier: float = 2,
         initial_delay: float = 1.0,
-        exception_handler=lambda _: ExceptionHandlerOutcome.CONTINUE
-):
-    def wrapper(function):
-        async def result(*args, **kwargs):
+        exception_handler: Callable[[Exception], ExceptionHandlerOutcome] = lambda _: ExceptionHandlerOutcome.CONTINUE
+) -> Callable[..., Any]:
+    @wraps(function)
+    async def result(*args: Any, **kwargs: Any) -> Any:
         attempts = 0
         delay = initial_delay
         while True:
@@ -73,10 +156,133 @@ def async_retry(
 
     return result
 
-    return wrapper
 
-def async_test(coro):
-    def wrapper(*args, **kwargs):
+def async_test(coro: Callable[..., Coroutine[Never, Never, None]]) -> Callable[..., None]:
+    @wraps(coro)
+    def wrapper(*args: Any, **kwargs: Any) -> None:
         with Runner() as runner:
             runner.run(coro(*args, **kwargs))
+
     return wrapper
+
+
+@no_type_check
+@decorator_with_kwargs
+def tmpdir(f,
+           argument_name='temp_dir',
+           suffix=None,
+           prefix=None,
+           dir=None,
+           ignore_cleanup_errors=False,
+           delete=True):
+    @no_type_check
+    @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[T]:
+
+    def __init__(self, fget: Callable[[], T], fset: Optional[Callable[[T], None]]=None):
+        self.fget = fget
+        self.fset = fset
+
+    def __get__(self, obj, klass=None):  # type: ignore
+        if klass is None:
+            klass = type(obj)
+        return self.fget.__get__(obj, klass)()
+
+    def __set__(self, obj, value):  # type: ignore
+        if not self.fset:
+            raise AttributeError("can't set attribute")
+        type_ = type(obj)
+        return self.fset.__get__(obj, type_)(value)
+
+    def setter(self, func):  # type: ignore
+        if not isinstance(func, (classmethod, staticmethod)):
+            func = classmethod(func)
+        self.fset = func
+        return self
+
+
+def classproperty(func):  # type: ignore
+    if not isinstance(func, (classmethod, staticmethod)):
+        func = classmethod(func)
+    return ClassPropertyDescriptor(func)
+
+
+class AsyncQueueIterator[T](AsyncIterator[T]):
+    _queue: Queue[Optional[T]]
+
+    def __init__(self, queue: Queue[Optional[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
+
+
+class aenumerate[T](AsyncIterator[Tuple[int, T]]):
+    """enumerate for async for"""
+
+    _aiterable: AsyncIterable[T]
+    _i: int
+
+    def __init__(self, aiterable: AsyncIterable[T], start: int = 0):
+        self._aiterable = aiterable
+        self._i = start - 1
+
+    def __aiter__(self) -> Self:
+        self._ait = self._aiterable.__aiter__()
+        return self
+
+    async def __anext__(self) -> Tuple[int, T]:
+        val = await self._ait.__anext__()
+        self._i += 1
+        return self._i, val
+
+
+def index_of_with_escape(haystack: str, needle: str, escape: str, begin: int, end: int = 0) -> int:
+    result = -1
+    cursor = begin
+    if end == 0:
+        end = len(haystack)
+    escape_count = 0
+
+    while cursor < end:
+        c = haystack[cursor]
+
+        if escape_count > 0:
+            escape_count -= 1
+            if c[0] == escape:
+                result = -1
+        elif escape_count == 0:
+            if c[0] == escape:
+                escape_count += 1
+            if cursor + len(needle) <= len(haystack):
+                test = haystack[cursor:cursor + len(needle)]
+                if test == needle:
+                    result = cursor
+
+        if result >= 0 and escape_count == 0:
+            break
+
+        cursor += 1
+
+    return result
+

tests/test_private.py

@@ -1,6 +1,7 @@
 import unittest
 
-from src.pwo import retry, async_retry, async_test
+from pwo import retry, async_retry, async_test, AsyncQueueIterator, aenumerate, index_of_with_escape
+from asyncio import Queue
 
 
 class PrivateTest(unittest.TestCase):
@@ -70,5 +71,59 @@ class PrivateTest(unittest.TestCase):
             await bar()
         self.assertEqual(max_attempts, attempt)
 
-if __name__ == '__main__':
-    unittest.main()
+    @async_test
+    async def test_async_queue_iterator(self):
+        queue = Queue()
+        queue_size = 10
+        objects = [object() for _ in range(queue_size)]
+
+        async def poll() -> int:
+            completed = 0
+            async for i, obj in aenumerate(AsyncQueueIterator(queue)):
+                self.assertIs(objects[i], obj)
+                completed += 1
+            return completed
+
+        handle = poll()
+
+        for o in objects:
+            queue.put_nowait(o)
+        queue.put_nowait(None)
+        processed = await handle
+        self.assertEqual(queue_size, processed)
+
+
+class TestIndexOfWithEscape(unittest.TestCase):
+
+    def run_test_case(self, haystack, needle, escape, expected_solution):
+        solution = []
+        i = 0
+        while True:
+            i = index_of_with_escape(haystack, needle, escape, i, len(haystack))
+            if i < 0:
+                break
+            solution.append(i)
+            i += 1
+        self.assertEqual(expected_solution, solution)
+
+    def test_simple(self):
+        self.run_test_case(" dsds $sdsa \\$dfivbdsf \\\\$sdgsga", '$', '\\', [6, 25])
+
+    def test_simple2(self):
+        self.run_test_case("asdasd$$vdfv$", '$', '$', [12])
+
+    def test_no_needle(self):
+        self.run_test_case("asdasd$$vdfv$", '#', '\\', [])
+
+    def test_escaped_needle(self):
+        self.run_test_case("asdasd$$vdfv$#sdfs", '#', '$', [])
+
+    def test_not_escaped_needle(self):
+        self.run_test_case("asdasd$$#vdfv$#sdfs", '#', '$', [8])
+
+    def test_special_case(self):
+        self.run_test_case("\n${sys:user.home}${env:HOME}", ':', '\\', [6, 22])
+
+    def test_wide_needle(self):
+        self.run_test_case("asdasd\\${#vdfv|${#sdfs}", '${', '\\', [15])
+

tests/test_try.py

@@ -0,0 +1,42 @@
+import unittest
+from pwo import Try
+
+
+class TestException(Exception):
+
+    def __init__(self, msg: str):
+        super().__init__(msg)
+
+
+class TryTest(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def test_try(self):
+
+        with self.subTest("Test failure"):
+            def throw_test_exception():
+                raise TestException("error")
+
+            t = Try.of(throw_test_exception)
+
+            with self.assertRaises(TestException):
+                t.get()
+
+            t = Try.failure(TestException("error"))
+
+            with self.assertRaises(TestException):
+                t.get()
+
+        with self.subTest("Test success"):
+            def complete_successfully():
+                return 42
+
+            t = Try.of(complete_successfully)
+
+            self.assertEqual(42, t.get())
+
+            t2 = t.handle(lambda value, err: value * 2)
+            self.assertEqual(84, t2.get())
+