Add to_a(sync)gen wrapper able to turn a sync generator function into its async variant and introduce an async version of os.walk.

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 25.1.0 (UNRELEASED)
 
+- Add `to_agen` generator asynchroniser tool with the `os.walk` asynchronous analogue introduced
+- Use `asyncio.to_thread` in the `to_coro` decorator that deprecates the `wrap` decorator
+  [#213](https://github.com/Tinche/aiofiles/pull/213)
 - Switch to [uv](https://docs.astral.sh/uv/) + add Python v3.14 support.
   ([#219](https://github.com/Tinche/aiofiles/pull/219))
 - Add `ruff` formatter and linter.

Justfile

@@ -1,19 +1,19 @@
+src_dir := "src"
 tests_dir := "tests"
-code_dirs := "src" + " " + tests_dir
-run_prefix := if env_var_or_default("VIRTUAL_ENV", "") == "" { "uv run " } else { "" }
+code_dirs := src_dir + " " + tests_dir
 
 check:
-	{{ run_prefix }}ruff format --check {{ code_dirs }}
-	{{ run_prefix }}ruff check {{ code_dirs }}
+	ruff format --check {{ code_dirs }}
+	ruff check {{ code_dirs }}
 
 coverage:
-	{{ run_prefix }}coverage run -m pytest {{ tests_dir }}
+	coverage run -m pytest {{ tests_dir }}
 
 format:
-	{{ run_prefix }}ruff format {{ code_dirs }}
+	ruff format {{ code_dirs }}
 
 lint: format
-	{{ run_prefix }}ruff check --fix {{ code_dirs }}
+	ruff check --fix {{ code_dirs }}
 
 test:
-	{{ run_prefix }}pytest -x --ff {{ tests_dir }}
+	pytest {{ tests_dir }}

pyproject.toml

@@ -68,6 +68,11 @@ source = [
 
 [tool.pytest.ini_options]
 minversion = "8.3"
+addopts = [
+    "-x",
+    "--ff",
+    "--maxfail=1"
+]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 

src/aiofiles/base.py

@@ -1,18 +1,77 @@
-from asyncio import get_running_loop
-from collections.abc import Awaitable
+from asyncio import get_running_loop, to_thread
+from collections.abc import AsyncIterator, Awaitable, Callable, Coroutine
 from contextlib import AbstractAsyncContextManager
 from functools import partial, wraps
+from queue import Empty, Queue
+from warnings import warn
 
 
-def wrap(func):
+def to_agen(func: Callable) -> AsyncIterator:
+    """Converts the routine `func` into an async generator function.
+
+    Args:
+        func: A generator function.
+
+    Returns:
+        An asynchronous generator function.
+    """
+
+    @wraps(func)
+    async def _wrapper(*args, **kwargs) -> AsyncIterator:
+        def _generate(q: Queue):
+            nonlocal is_over
+            try:
+                for row in func(*args, **kwargs):
+                    q.put_nowait(row)
+            finally:
+                is_over = True
+
+        loop = get_running_loop()
+        queue: Queue = Queue()
+
+        is_over = False
+
+        gen = partial(_generate, q=queue)
+        loop.run_in_executor(None, gen)
+        while True:
+            if is_over and queue.empty():
+                break
+            try:
+                item = queue.get_nowait()
+                queue.task_done()
+                yield item
+            except Empty:
+                pass
+        queue.join()
+
+    return _wrapper
+
+
+def to_coro(func: Callable) -> Callable:
+    """Converts the routine `func` into a coroutine.
+
+    The returned coroutine function runs the decorated function
+    in a separate thread.
+
+    Args:
+        func: A routine (regular function).
+
+    Returns:
+        A coroutine function.
+    """
+
     @wraps(func)
-    async def run(*args, loop=None, executor=None, **kwargs):
-        if loop is None:
-            loop = get_running_loop()
-        pfunc = partial(func, *args, **kwargs)
-        return await loop.run_in_executor(executor, pfunc)
+    async def _wrapper(*args, **kwargs) -> Coroutine:
+        return await to_thread(func, *args, **kwargs)
+
+    return _wrapper
+
 
-    return run
+def wrap(func: Callable) -> Callable:
+    warn(
+        "scheduled to removal, consider using to_coro", DeprecationWarning, stacklevel=1
+    )
+    return to_coro(func)
 
 
 class AsyncBase:
@@ -26,7 +85,6 @@ def _loop(self):
         return self._ref_loop or get_running_loop()
 
     def __aiter__(self):
-        """We are our own iterator."""
         return self
 
     def __repr__(self):

src/aiofiles/os.py

@@ -3,59 +3,61 @@
 import os
 
 from . import ospath as path
-from .base import wrap
+from .base import to_agen, to_coro
 
 __all__ = [
-    "path",
-    "stat",
+    "access",
+    "getcwd",
+    "listdir",
+    "makedirs",
+    "mkdir",
+    "readlink",
+    "remove",
+    "removedirs",
     "rename",
     "renames",
     "replace",
-    "remove",
-    "unlink",
-    "mkdir",
-    "makedirs",
     "rmdir",
-    "removedirs",
-    "symlink",
-    "readlink",
-    "listdir",
+    "path",
     "scandir",
-    "access",
-    "wrap",
-    "getcwd",
+    "stat",
+    "symlink",
+    "unlink",
+    "walk",
 ]
 
-access = wrap(os.access)
+access = to_coro(os.access)
+
+getcwd = to_coro(os.getcwd)
 
-getcwd = wrap(os.getcwd)
+listdir = to_coro(os.listdir)
 
-listdir = wrap(os.listdir)
+makedirs = to_coro(os.makedirs)
+mkdir = to_coro(os.mkdir)
 
-makedirs = wrap(os.makedirs)
-mkdir = wrap(os.mkdir)
+readlink = to_coro(os.readlink)
+remove = to_coro(os.remove)
+removedirs = to_coro(os.removedirs)
+rename = to_coro(os.rename)
+renames = to_coro(os.renames)
+replace = to_coro(os.replace)
+rmdir = to_coro(os.rmdir)
 
-readlink = wrap(os.readlink)
-remove = wrap(os.remove)
-removedirs = wrap(os.removedirs)
-rename = wrap(os.rename)
-renames = wrap(os.renames)
-replace = wrap(os.replace)
-rmdir = wrap(os.rmdir)
+scandir = to_coro(os.scandir)
+stat = to_coro(os.stat)
+symlink = to_coro(os.symlink)
 
-scandir = wrap(os.scandir)
-stat = wrap(os.stat)
-symlink = wrap(os.symlink)
+unlink = to_coro(os.unlink)
 
-unlink = wrap(os.unlink)
+walk = to_agen(os.walk)
 
 
 if hasattr(os, "link"):
     __all__ += ["link"]
-    link = wrap(os.link)
+    link = to_coro(os.link)
 if hasattr(os, "sendfile"):
     __all__ += ["sendfile"]
-    sendfile = wrap(os.sendfile)
+    sendfile = to_coro(os.sendfile)
 if hasattr(os, "statvfs"):
     __all__ += ["statvfs"]
-    statvfs = wrap(os.statvfs)
+    statvfs = to_coro(os.statvfs)

src/aiofiles/ospath.py

@@ -2,7 +2,7 @@
 
 from os import path
 
-from .base import wrap
+from .base import to_coro
 
 __all__ = [
     "abspath",
@@ -19,19 +19,19 @@
     "sameopenfile",
 ]
 
-abspath = wrap(path.abspath)
+abspath = to_coro(path.abspath)
 
-getatime = wrap(path.getatime)
-getctime = wrap(path.getctime)
-getmtime = wrap(path.getmtime)
-getsize = wrap(path.getsize)
+getatime = to_coro(path.getatime)
+getctime = to_coro(path.getctime)
+getmtime = to_coro(path.getmtime)
+getsize = to_coro(path.getsize)
 
-exists = wrap(path.exists)
+exists = to_coro(path.exists)
 
-isdir = wrap(path.isdir)
-isfile = wrap(path.isfile)
-islink = wrap(path.islink)
-ismount = wrap(path.ismount)
+isdir = to_coro(path.isdir)
+isfile = to_coro(path.isfile)
+islink = to_coro(path.islink)
+ismount = to_coro(path.ismount)
 
-samefile = wrap(path.samefile)
-sameopenfile = wrap(path.sameopenfile)
+samefile = to_coro(path.samefile)
+sameopenfile = to_coro(path.sameopenfile)

tests/test_os.py

@@ -3,9 +3,11 @@
 import asyncio
 import os
 import platform
+from collections import defaultdict
 from os import stat
 from os.path import dirname, exists, isdir, join
 from pathlib import Path
+from typing import Optional
 
 import pytest
 
@@ -498,3 +500,74 @@ async def test_abspath():
     abs_filename = join(dirname(__file__), "resources", "test_file1.txt")
     result = await aiofiles.os.path.abspath(relative_filename)
     assert result == abs_filename
+
+
+def _act_on_error(oserror):
+    print(f"aiofiles.os.walk onerror handler: {oserror}")
+
+
+@pytest.mark.parametrize(
+    ("top", "topdown", "onerror", "followlinks"),
+    [
+        ("non-existent", True, _act_on_error, False),
+        ("README.md", True, None, False),
+        ("README.md", False, _act_on_error, False),
+        ("./src", True, _act_on_error, False),
+        ("./src", False, _act_on_error, False),
+        ("./tests", True, None, True),
+        ("./tests", False, None, True),
+    ],
+)
+async def test_walk(top, topdown, onerror, followlinks):
+    result = [  # noqa: C416
+        row
+        async for row in aiofiles.os.walk(
+            top=top, topdown=topdown, onerror=onerror, followlinks=followlinks
+        )
+    ]
+    answer = list(
+        os.walk(top=top, topdown=topdown, onerror=onerror, followlinks=followlinks)
+    )
+
+    assert result == answer
+
+
+@pytest.mark.parametrize(
+    "top_path",
+    [
+        "c3Po-ar2D2",  # does not mean to exist
+        "./.github",
+        "./src",
+        "./tests",
+        # ".",  # it is long
+    ],
+)
+async def test_walk_non_blocking(top_path: str):
+    async def _test_walk(
+        top: str, key: str, storage: defaultdict[str, list]
+    ) -> Optional[bool]:
+        non_blocking = None
+        async for datum in aiofiles.os.walk(top=top):
+            storage[key].append(datum)
+            # checking the statistics (common shareable resource) for other tasks
+            other_keys = set(storage) - {key}
+            if not non_blocking:
+                non_blocking = any(storage[other_key] for other_key in other_keys)
+            # without the `asyncio.sleep(0)` the `async for` may block
+            await asyncio.sleep(0)
+
+        return non_blocking
+
+    keys = {f"w_{i}" for i in range(2)}
+    storage = defaultdict(list)
+
+    tasks: list[asyncio.Task] = [
+        asyncio.create_task(_test_walk(top=top_path, key=key, storage=storage))
+        for key in keys
+    ]
+    results = await asyncio.gather(*tasks)
+
+    # Asserting that each task does not block the event loop.
+    # Empty results are not assumed to be non-blocking by default.
+    if statuses := (r for r in results if r is not None):
+        assert all(statuses)