libs: Add aio.functional

.github/workflows/ci.yml

@@ -10,24 +10,52 @@ jobs:
 
     - uses: actions/setup-python@v2
 
-    - run: cd pytest-patches && pip install -e . && pip install -e .[lint] && pip install -e .[types] && pip install -e .[publish]
+    - run: cd pytest-patches && pip install -e . && pip install -e .[lint] && pip install -e .[types] && pip install -e .[publish]  && pip install -e .[test]
 
-    - name: Lint
+    - name: Lint (pytest-patches)
       run: flake8 pytest-patches
 
-    - name: Typing
+    - name: Typing (pytest-patches)
       run: mypy pytest-patches
 
-    - name: Test
+    - name: Test (pytest-patches)
       run: pytest --color=yes -vv pytest-patches
 
-    - name: Build
+    - name: Build (pytest-patches)
       run: cd pytest-patches && python setup.py bdist_wheel && ls dist
 
-    - name: Publish
+    - name: Publish (pytest-patches)
       if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
         user: __token__
         password: ${{ secrets.PYPI_TOKEN }}
         packages_dir: pytest-patches/dist
+
+  aio_functional:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v1
+    - uses: actions/setup-python@v2
+
+    - run: cd aio.functional && pip install -e . && pip install -e .[lint] && pip install -e .[types] && pip install -e .[publish] && pip install -e .[test]
+
+    - name: Lint (aio.functional)
+      run: flake8 aio.functional
+
+    - name: Typing (aio.functional)
+      run: mypy --namespace-packages aio.functional/aio
+
+    - name: Test (aio.functional)
+      run: pytest --color=yes -vv aio.functional
+
+    - name: Build (aio.functional)
+      run: cd aio.functional && python setup.py bdist_wheel && ls dist
+
+    - name: Publish (aio.functional)
+      if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
+      uses: pypa/gh-action-pypi-publish@release/v1
+      with:
+        user: __token__
+        password: ${{ secrets.PYPI_TOKEN }}
+        packages_dir: aio.functional/dist

aio.functional/README.rst

@@ -0,0 +1,3 @@
+
+aio.functional
+==============

aio.functional/aio/functional/__init__.py

@@ -0,0 +1,5 @@
+
+from .decorators import async_property
+
+
+__all__ = ("async_property", )

aio.functional/aio/functional/decorators.py

@@ -0,0 +1,87 @@
+#
+# Decorators
+#
+
+import inspect
+from typing import Any, Callable, Optional
+
+
+class NoCache(Exception):
+    pass
+
+
+class async_property:  # noqa: N801
+    name = None
+    cache_name = "__async_prop_cache__"
+    _instance = None
+
+    # If the decorator is called with `kwargs` then `fun` is `None`
+    # and instead `__call__` is triggered with `fun`
+    def __init__(self, fun: Optional[Callable] = None, cache: bool = False):
+        self.cache = cache
+        self._fun = fun
+        self.name = getattr(fun, "__name__", None)
+        self.__doc__ = getattr(fun, '__doc__')
+
+    def __call__(self, fun: Callable) -> 'async_property':
+        self._fun = fun
+        self.name = self.name or fun.__name__
+        self.__doc__ = getattr(fun, '__doc__')
+        return self
+
+    def __get__(self, instance: Any, cls=None) -> Any:
+        if instance is None:
+            return self
+        self._instance = instance
+        if inspect.isasyncgenfunction(self._fun):
+            return self.async_iter_result()
+        return self.async_result()
+
+    def fun(self, *args, **kwargs):
+        if self._fun:
+            return self._fun(*args, **kwargs)
+
+    @property
+    def prop_cache(self) -> dict:
+        return getattr(self._instance, self.cache_name, {})
+
+    # An async wrapper function to return the result
+    # This is returned when the prop is called if the wrapped
+    # method is an async generator
+    async def async_iter_result(self):
+        # retrieve the value from cache if available
+        try:
+            result = self.get_cached_prop()
+        except (NoCache, KeyError):
+            result = None
+
+        if result is None:
+            result = self.set_prop_cache(self.fun(self._instance))
+
+        async for item in result:
+            yield item
+
+    # An async wrapper function to return the result
+    # This is returned when the prop is called
+    async def async_result(self) -> Any:
+        # retrieve the value from cache if available
+        try:
+            return self.get_cached_prop()
+        except (NoCache, KeyError):
+            pass
+
+        # derive the result, set the cache if required, and return the result
+        return self.set_prop_cache(await self.fun(self._instance))
+
+    def get_cached_prop(self) -> Any:
+        if not self.cache:
+            raise NoCache
+        return self.prop_cache[self.name]
+
+    def set_prop_cache(self, result: Any) -> Any:
+        if not self.cache:
+            return result
+        cache = self.prop_cache
+        cache[self.name] = result
+        setattr(self._instance, self.cache_name, cache)
+        return result

aio.functional/setup.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+
+import os
+import codecs
+from setuptools import find_namespace_packages, setup  # type:ignore
+
+
+def read(fname):
+    file_path = os.path.join(os.path.dirname(__file__), fname)
+    return codecs.open(file_path, encoding='utf-8').read()
+
+
+setup(
+    name='aio.functional',
+    version='0.0.1',
+    author='Ryan Northey',
+    author_email='ryan@synca.io',
+    maintainer='Ryan Northey',
+    maintainer_email='ryan@synca.io',
+    license='Apache Software License 2.0',
+    url='https://github.com/phlax/pytooling/aio.functional',
+    description='A collection of functional utils for asyncio',
+    long_description=read('README.rst'),
+    py_modules=['aio.functional'],
+    packages=find_namespace_packages(),
+    python_requires='>=3.5',
+    extras_require={
+        "test": ["pytest", "pytest-asyncio"],
+        "lint": ['flake8'],
+        "types": ['mypy'],
+        "publish": ['wheel'],
+    },
+    classifiers=[
+        'Development Status :: 4 - Beta',
+        'Framework :: Pytest',
+        'Intended Audience :: Developers',
+        'Topic :: Software Development :: Testing',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.5',
+        'Programming Language :: Python :: 3.6',
+        'Programming Language :: Python :: 3.7',
+        'Programming Language :: Python :: 3.8',
+        'Programming Language :: Python :: 3 :: Only',
+        'Programming Language :: Python :: Implementation :: CPython',
+        'Programming Language :: Python :: Implementation :: PyPy',
+        'Operating System :: OS Independent',
+        'License :: OSI Approved :: Apache Software License',
+    ],
+)

aio.functional/tests/test_functional.py

@@ -0,0 +1,120 @@
+import types
+from unittest.mock import AsyncMock
+
+import pytest
+
+from aio import functional
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("cache", [None, True, False])
+@pytest.mark.parametrize("raises", [True, False])
+@pytest.mark.parametrize("result", [None, False, "X", 23])
+async def test_functional_async_property(cache, raises, result):
+    m_async = AsyncMock(return_value=result)
+
+    class SomeError(Exception):
+        pass
+
+    if cache is None:
+        decorator = functional.async_property
+        iter_decorator = functional.async_property
+    else:
+        decorator = functional.async_property(cache=cache)
+        iter_decorator = functional.async_property(cache=cache)
+
+    items = [f"ITEM{i}" for i in range(0, 5)]
+
+    class Klass:
+
+        @decorator
+        async def prop(self):
+            """This prop deserves some docs"""
+            if raises:
+                await m_async()
+                raise SomeError("AN ERROR OCCURRED")
+            else:
+                return await m_async()
+
+        @iter_decorator
+        async def iter_prop(self):
+            """This prop also deserves some docs"""
+            if raises:
+                await m_async()
+                raise SomeError("AN ITERATING ERROR OCCURRED")
+            result = await m_async()
+            for item in items:
+                yield item, result
+
+    klass = Klass()
+
+    # The class.prop should be an instance of async_prop
+    # and should have the name and docs of the wrapped method.
+    assert isinstance(
+        type(klass).prop,
+        functional.async_property)
+    assert (
+        type(klass).prop.__doc__
+        == "This prop deserves some docs")
+    assert (
+        type(klass).prop.name
+        == "prop")
+
+    if raises:
+        with pytest.raises(SomeError) as e:
+            await klass.prop
+
+        with pytest.raises(SomeError) as e2:
+            async for result in klass.iter_prop:
+                pass
+
+        assert (
+            e.value.args[0]
+            == 'AN ERROR OCCURRED')
+        assert (
+            e2.value.args[0]
+            == 'AN ITERATING ERROR OCCURRED')
+        assert (
+            list(m_async.call_args_list)
+            == [[(), {}]] * 2)
+        return
+
+    # results can be repeatedly awaited
+    assert await klass.prop == result
+    assert await klass.prop == result
+
+    # results can also be repeatedly iterated
+    results1 = []
+    async for returned_result in klass.iter_prop:
+        results1.append(returned_result)
+    assert results1 == [(item, result) for item in items]
+
+    results2 = []
+    async for returned_result in klass.iter_prop:
+        results2.append(returned_result)
+
+    if not cache:
+        assert results2 == results1
+        assert (
+            list(list(c) for c in m_async.call_args_list)
+            == [[(), {}]] * 4)
+        assert not hasattr(klass, functional.async_property.cache_name)
+        return
+
+    # with cache we can keep awaiting the result but the fun
+    # is still only called once
+    assert await klass.prop == result
+    assert await klass.prop == result
+    assert (
+        list(list(c) for c in m_async.call_args_list)
+        == [[(), {}]] * 2)
+
+    iter_prop = getattr(
+        klass, functional.async_property.cache_name)["iter_prop"]
+    assert isinstance(iter_prop, types.AsyncGeneratorType)
+    assert (
+        getattr(klass, functional.async_property.cache_name)
+        == dict(prop=m_async.return_value, iter_prop=iter_prop))
+
+    # cached iterators dont give any more results once they are done
+    assert results2 == []