Add support for async apply

Add support for async apply

Author: python-mario-bot

dev-requirements.in

@@ -1,3 +1,4 @@
+-r requirements.in
 bump2version
 coverage
 hypothesis

dev-requirements.txt

@@ -7,13 +7,17 @@
 aiohttp==3.5.4
 alabaster==0.7.12         # via sphinx
 ansicolors==1.0.2         # via cuvner
-appdirs==1.4.3            # via black
+anyio==1.0.0              # via asks
+appdirs==1.4.3
 argh==0.26.2              # via watchdog
+asks==2.3.3
 asn1crypto==0.24.0        # via cryptography
 astroid==2.2.5            # via pylint
+async-exit-stack==1.0.1
+async-generator==1.10
 async-timeout==3.0.1      # via aiohttp
 atomicwrites==1.3.0       # via pytest
-attrs==19.1.0             # via aiohttp, automat, black, hypothesis, pytest, service-identity, treq, twisted
+attrs==19.1.0
 automat==0.7.0            # via twisted
 babel==2.6.0              # via sphinx
 black==19.3b0
@@ -22,17 +26,18 @@ bump2version==0.5.10
 certifi==2019.3.9         # via requests
 cffi==1.12.2              # via cryptography
 chardet==3.0.4            # via aiohttp, requests
-click==7.0                # via black, cuvner
+click==7.0
 constantly==15.1.0        # via twisted
 coverage==4.5.3
 cryptography==2.6.1       # via pyopenssl, service-identity
 cuvner==18.0.1
 docshtest==0.0.2
 docutils==0.14            # via readme-renderer, sphinx
 filelock==3.0.10          # via tox
+h11==0.9.0                # via asks
 hyperlink==19.0.0         # via twisted
 hypothesis==4.17.1
-idna==2.8                 # via hyperlink, requests, twisted, yarl
+idna==2.8                 # via hyperlink, requests, trio, twisted, yarl
 imagesize==1.1.0          # via sphinx
 incremental==17.5.0       # via cuvner, treq, twisted
 isort==4.3.17             # via pylint
@@ -42,9 +47,13 @@ lazy-object-proxy==1.3.1  # via astroid
 lockfile==0.12.2
 lxml==4.3.4
 markupsafe==1.1.1         # via jinja2
+marshmallow==3.0.0rc7
 mccabe==0.6.1             # via pylint
 more-itertools==7.0.0     # via pytest
 multidict==4.5.2          # via aiohttp, yarl
+mypy-extensions==0.4.1    # via mypy, trio-typing
+mypy==0.711               # via trio-typing
+outcome==1.0.0            # via trio
 packaging==19.0           # via pytest-sugar, sphinx
 parso==0.4.0
 pathtools==0.1.2          # via watchdog
@@ -59,6 +68,7 @@ pyhamcrest==1.9.0         # via twisted
 pylint==2.3.1
 pyopenssl==19.0.0         # via twisted
 pyparsing==2.4.0          # via packaging
+pyrsistent==0.15.2
 pytest-clarity==0.1.0a1
 pytest-cov==2.6.1
 pytest-sugar==0.9.2
@@ -69,8 +79,10 @@ readme-renderer==24.0     # via twine
 requests-toolbelt==0.8.0  # via twine
 requests==2.21.0
 service-identity==18.1.0  # via twisted
-six==1.12.0               # via astroid, automat, bleach, cryptography, cuvner, packaging, pyhamcrest, pyopenssl, pytest, readme-renderer, tox, treq
+six==1.12.0               # via astroid, automat, bleach, cryptography, cuvner, packaging, pyhamcrest, pyopenssl, pyrsistent, pytest, readme-renderer, tox, treq
+sniffio==1.1.0            # via anyio, trio
 snowballstemmer==1.2.1    # via sphinx
+sortedcontainers==2.1.0   # via trio
 sphinx-rtd-theme==0.4.3
 sphinx==2.0.1
 sphinxcontrib-applehelp==1.0.1  # via sphinx
@@ -80,13 +92,17 @@ sphinxcontrib-jsmath==1.0.1  # via sphinx
 sphinxcontrib-qthelp==1.0.2  # via sphinx
 sphinxcontrib-serializinghtml==1.1.3  # via sphinx
 termcolor==1.1.0          # via pytest-clarity, pytest-sugar
-toml==0.10.0              # via black, tox
+toml==0.10.0
+toolz==0.9.0
 tox==3.9.0
 tqdm==4.31.1              # via twine
 treq==18.6.0
+trio-typing==0.2.0
+trio==0.11.0
 twine==1.13.0
 twisted[tls]==19.2.0
-typed-ast==1.3.1          # via astroid
+typed-ast==1.4.0          # via astroid, mypy
+typing-extensions==3.7.4  # via trio-typing
 unidiff==0.5.5            # via cuvner
 urllib3==1.24.2
 virtualenv==16.4.3        # via tox

requirements.in

@@ -11,3 +11,4 @@ toml
 asks
 marshmallow>=3.0rc1
 pyrsistent
+trio_typing

requirements.txt

@@ -15,6 +15,8 @@ click==7.0
 h11==0.9.0                # via asks
 idna==2.8                 # via trio
 marshmallow==3.0.0rc7
+mypy-extensions==0.4.1    # via mypy, trio-typing
+mypy==0.711               # via trio-typing
 outcome==1.0.0            # via trio
 parso==0.4.0
 pyrsistent==0.15.2
@@ -23,4 +25,7 @@ sniffio==1.1.0            # via anyio, trio
 sortedcontainers==2.1.0   # via trio
 toml==0.10.0
 toolz==0.9.0
+trio-typing==0.2.0
 trio==0.11.0
+typed-ast==1.4.0          # via mypy
+typing-extensions==3.7.4  # via trio-typing

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.0.109
+current_version = 0.0.110
 commit = True
 tag = True
 

setup.py

@@ -8,10 +8,12 @@
 with open(PROJECT_ROOT / "requirements.in") as f:
     INSTALL_REQUIRES = f.read().splitlines()
 
+with open(PROJECT_ROOT / "dev-requirements.in") as f:
+    DEV_INSTALL_REQUIRES = f.read().splitlines()
 
 setuptools.setup(
     name="mario",
-    version="0.0.109",
+    version="0.0.110",
     description="Shell pipes for Python.",
     long_description=open(PROJECT_ROOT / "README.rst").read(),
     long_description_content_type="text/x-rst",
@@ -44,4 +46,5 @@
         "console_scripts": ["mario = mario.cli:cli"],
         "mario_plugins": ["basic = mario.plugins"],
     },
+    extras={"dev": DEV_INSTALL_REQUIRES},
 )

src/mario/_version.py

@@ -1 +1 @@
-__version__ = "0.0.109"
+__version__ = "0.0.110"

src/mario/plugins/basic.py

@@ -86,11 +86,13 @@ async def async_filter(function, items, exit_stack, max_concurrent):
 
 @registry.add_traversal("apply", calculate_more_params=calculate_function)
 async def apply(function, items):
+    """When ``function`` takes an iterable."""
     return traversals.AsyncIterableWrapper([await function([x async for x in items])])
 
 
 @registry.add_traversal("async_apply", calculate_more_params=calculate_function)
 async def async_apply(function, items):
+    """When ``function`` takes an async iterable."""
     return await traversals.async_apply(function, items)
 
 

src/mario/testing/atools.py

@@ -0,0 +1,59 @@
+async def async_drop_falsy(items):
+    async for x in items:
+        if x:
+            yield x
+
+
+def drop_falsy(items):
+    for x in items:
+        if x:
+            yield x
+
+
+def my_max(items):
+    print("hello")
+
+    it = iter(items)
+
+    try:
+        current_max = next(it)
+    except StopIteration as e:
+        raise ValueError("Iterable was empty") from e
+    for x in it:
+
+        if x > current_max:
+            current_max = x
+    return x
+
+
+# def gen_max(items):
+#     it = iter(items)
+#     try:
+#         current_max = next(it)
+#     except StopIteration as e:
+#         raise ValueError("Iterable was empty") from e
+#     for x in it:
+
+#         if x > current_max:
+#             current_max = x
+#     yield x
+
+
+def gen_max(items):
+    yield from [my_max(items)]
+
+
+import functools
+
+
+def wrap_sync_fold(function):
+    @functools.wraps(function)
+    def wrap(items):
+        yield from [function(items)]
+
+    return wrap
+
+
+wrapped_max = wrap_sync_fold(my_max)
+
+wrapped_drop_falsy = wrap_sync_fold(drop_falsy)

src/mario/traversals.py

@@ -68,15 +68,101 @@ async def __anext__(self):
             raise StopAsyncIteration
 
 
-async def async_apply(
-    function: Callable[[Iterable[T]], Any], aiterable: AsyncIterable[T]
+import typing as t
+import functools
+import itertools
+import threading
+
+
+import trio
+import trio_typing
+
+T = t.TypeVar("T")
+
+
+def _pull_values_from_async_iterator(
+    in_trio: trio.BlockingTrioPortal,
+    ait: t.AsyncIterator[T],
+    send_to_trio: trio.abc.SendChannel[T],
+):
+    """Make a generator by asking Trio to async-iterate over the async iterable,
+    then yield the results.
+
+    This function is run in a thread.
+    """
+    while True:
+        try:
+            yield in_trio.run(ait.__anext__)
+        except StopAsyncIteration:
+            break
+
+
+def _threaded_sync_apply(
+    in_trio: trio.BlockingTrioPortal,
+    function: t.Callable[[t.Iterable[T]], t.Any],
+    send_to_trio: trio.abc.SendChannel,
+    ait: t.AsyncIterator[T],
+    receive_from_thread: trio.abc.ReceiveChannel,
+    iterator=None,
+):
+    """Extract values from async iterable into a sync iterator, apply function to
+    it, and send the result back into Trio.
+
+    This function is run in a thread.
+
+    """
+
+    try:
+        for x in iterator:
+            in_trio.run(send_to_trio.send, x)
+    finally:
+        in_trio.run(send_to_trio.aclose)
+
+
+async def sync_apply(
+    sync_function: t.Callable[[t.Iterable[T]], t.Any], data: t.AsyncIterable[T]
 ):
-    async for item in IterableToAsyncIterable(
-        [function((await x) for x in AsyncIterableToIterable(aiterable))]
-    ):
-        if isinstance(item, types.CoroutineType):
-            return await item
-        return item
+    """Apply a sync Callable[Iterable] to an async iterable by pulling values in a
+    thread.
+    """
+    in_trio = trio.BlockingTrioPortal()
+    send_to_trio, receive_from_thread = trio.open_memory_channel[t.Tuple[T, t.Any]](0)
+
+    async with trio.open_nursery() as n:
+
+        iterator = await sync_function(
+            _pull_values_from_async_iterator(in_trio, data, send_to_trio)
+        )
+        n.start_soon(
+            trio.run_sync_in_worker_thread,
+            functools.partial(
+                _threaded_sync_apply,
+                function=sync_function,
+                iterator=iterator,
+                ait=data,
+                in_trio=in_trio,
+                send_to_trio=send_to_trio,
+                receive_from_thread=receive_from_thread,
+            ),
+        )
+
+        async for x in receive_from_thread:
+            yield x
+
+
+import functools
+
+
+def wrap_sync_fold(function):
+    @functools.wraps(function)
+    def wrap(items):
+        yield from [function(items)]
+
+    return wrap
+
+
+async def async_apply(function, data):
+    return await function(data)
 
 
 @async_generator.asynccontextmanager