miquido-haiway-146

/repo/src/haiway/helpers/caching.py
/repo/src/haiway/helpers/concurrent.py
/repo/src/haiway/helpers/configuration.py
/repo/src/haiway/helpers/files.py
/repo/src/haiway/helpers/http_client.py
/repo/src/haiway/helpers/observability.py
/repo/src/haiway/helpers/retries.py
/repo/src/haiway/helpers/throttling.py
/repo/src/haiway/helpers/timeouting.py
/repo/src/haiway/helpers/__init__.py
/repo/src/haiway/httpx/client.py
/repo/src/haiway/httpx/__init__.py
/r
…[599 chars omitted]…
ormatting.py
/repo/src/haiway/utils/logs.py
/repo/src/haiway/utils/metadata.py
/repo/src/haiway/utils/mimic.py
/repo/src/haiway/utils/noop.py
/repo/src/haiway/utils/queue.py
/repo/src/haiway/utils/stream.py
/repo/src/haiway/utils/__init__.py
/repo/src/haiway/__init__.py
/repo/tests/test_async_queue.py
CLAUDE.md
LICENSE
Makefile
README.md
config
docs
llms.txt
mkdocs.yml
pyproject.toml
src
tests
uv.lock

from haiway.context import ctx

__all__ = (
    "concurrently",
    "execute_concurrently",
    "process_concurrently",
    "stream_concurrently",
)


async def process_concurrently[Element](  # noqa: C901
    source: AsyncIterable[Element] | Iterable[Element],
    /,
    handler: Callable[[Element], Coroutine[Any, Any, None]],
    *,
    concurrent_tasks: int = 2,
    ignore_exceptions: bool = False,
) -> None:
    """Process elements from an iterable concurrently.

    Consumes elements from an iterable and processes them using the provided
    handler function. Processing happens concurrently with a configurable maxi
…[8630 chars omitted]…
ream.py
test_attribute_path.py
test_attribute_requirement.py
test_auto_retry.py
test_cache.py
test_concurrently.py
test_context.py
test_context_presets.py
test_disposables.py
test_events.py
test_execute_concurrently.py
test_metadata.py
test_process_concurrently.py
test_state.py
test_state_typing.py
test_state_validation.py
test_stream_concurrently.py
test_streaming.py
test_timeout.py
test_variables.py

    Exception
        Any exception raised by handler tasks when return_exceptions is False.

    Examples
    --------
    >>> async def fetch_data(url: str) -> dict:
    ...     return await http_client.get(url)
    ...
    >>> urls = ["http://api.example.com/1", "http://api.example.com/2"]
    >>> results = await execute_concurrently(
    ...     fetch_data,
    ...     urls,
    ...     concurrent_tasks=10
    ... )
    >>> # results[0] corresponds to urls[0], results[1] to urls[
…[5900 chars omitted]…
utines,
    ...     concurrent_tasks=2,
    ...     return_exceptions=True
    ... )
    >>> for i, result in enumerate(results):
    ...     if isinstance(result, BaseException):
    ...         print(f"Coroutine {i} failed: {result}")
    ...     else:
    ...         print(f"Coroutine {i} succeeded")

    """
    assert concurrent_tasks > 0  # nosec: B101
    tasks: MutableSet[Task[Result]] = set()

        if len(tasks) < concurrent_tasks:
            return  # keep spawning tasks

        completed, tasks = await wait(
            tasks,
            return_when=FIRST_COMPLETED,
        )

        for task in completed:
            if exc := task.exception():
                if not return_exceptions:
                    raise exc

                ctx.log_error(
                    f"Concurrent execution error - {type(exc)}: {exc}",
                    exception=exc,
                )

    try:
        if isinstance(coroutines, AsyncIterable):
   
…[8647 chars omitted]…
   exc_type,
                exc_val,
                exc_tb,
            )

        except ExceptionGroup:
            pass  # skip task group exceptions

        finally:
            object.__setattr__(
                self,
                "_token",
                None,
            )
            object.__setattr__(
                self,
                "_group",
                None,
            )

async def async_range(
    start: int,
    stop: int,
    delay: float = 0,
) -> AsyncIterator[int]:
    for i in range(start, stop):
        if delay:
            await sleep(delay)
        yield i


async def async_letters(
    letters: str,
    delay: float = 0,
) -> AsyncIterator[str]:
    for letter in letters:
        if delay:
            await sleep(delay)
        yield letter


@mark.asyncio
async def test_merges_two_streams():
    items: list[int | str] = []

    async for item in stream_concurrently(async_range(0, 3), async_letters("abc")):
        items.append(item)

    # Should have all items from both sources
    assert len(items)
…[8629 chars omitted]…
   coroutines = [good_coro(), bad_coro(), good_coro()]
    with raises(FakeException):
        await concurrently(coroutines)


@mark.asyncio
async def test_returns_exceptions_when_configured():
    async def good_coro(value: int) -> int:
        return value * 2

    async def bad_coro() -> int:
        raise FakeException("Test exception")

    coroutines = [
        good_coro(1),  # Should return 2

  File "/tmp/repro.py", line 12, in main
    res = await concurrently(coros, return_exceptions=True)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/src/haiway/helpers/concurrent.py", line 465, in concurrently
    raise exc
  File "/repo/src/haiway/helpers/concurrent.py", line 458, in concurrently
    await process(element)
  
…[180 chars omitted]…
tasks.py", line 464, in wait
    return await _wait(fs, timeout, return_when, loop)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 550, in _wait
    await waiter
asyncio.exceptions.CancelledError
sys:1: RuntimeWarning: coroutine 'main.<locals>.bad' was never awaited
sys:1: RuntimeWarning: coroutine 'main.<locals>.good' was never awaited

358 |     """Execute multiple coroutines concurrently with controlled parallelism.
    | ^^^^
359 |
360 |     Executes a collection of coroutines concurrently, limiting the number of
    |

Found 2 errors.
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/repo/src/haiway/__init__.py", line 17, in <module>
    from haiway.helpers import (
  File "/repo/src/haiway/helpers/__init__.py", line 3, in <module>
    from haiway.helpers.concurrent import (
  File "/repo/src/haiway/helpers/concurrent.py", line 197
    """Execute handler for each element from a collection concurrently.
IndentationError: unexpected indent

    assert concurrent_tasks > 0  # nosec: B101
    tasks: MutableSet[Task[None]] = set()

    async def process(
        element: Element,
        /,
    ) -> None:
        nonlocal tasks
        if ignore_exceptions:
            tasks.add(_spawn_protected(handler(element)))
        else:
            tasks.add(ctx.spawn(handler, element))

        if len(tasks) < concurrent_tasks:
            return  # keep spawning tasks

        completed, tasks = await wait(
            tasks,
            return_when=FIRST_COMPLETED,
        )

        for task in completed:
            if ignore_exceptions:
                if isinstance(
…[197 chars omitted]…
            )

            elif exc := task.exception():
                raise exc

    try:
        if isinstance(source, AsyncIterable):
            async for element in source:
                await process(element)

        else:
            assert isinstance(source, Iterable)  # nosec: B101
            for element in source:
                await process(element)

    except BaseException as exc:

from haiway.context import ctx

__all__ = (
    "concurrently",
    "execute_concurrently",
    "process_concurrently",
    "stream_concurrently",
)


class _Exhausted:
    """Internal sentinel marking an exhausted source iterator."""

    __slots__ = ()

    def __repr__(self) -> str:
        return "EXHAUSTED"


_EXHAUSTED: _Exhausted = _Exhausted()


def _spawn_protected[Result](
    coro: Coroutine[None, None, Result],
    /,
) -> Task[Result | BaseException]:
    """Spawn a task which captures exceptions instead of propagating them.

        except Exception as exc:  # noqa: BLE001 - intentionally broad
            return exc

    return ctx.spawn(protected())

    """Process elements from an iterable concurrently.

    Consumes elements from an iterable and processes them using the provided
    handler function. Processing happens concurrently with a configurable maximum
    number of concurrent tasks. Elements are processed as they become available,
    maintaining the specified concurrency limit.

    The function continues until the source iterator is exhausted. If the function
    is cancelled, all running tasks are also cancelled. When ignore_exceptions is
    False, the first exception encountered will stop processing and propagate.

    Parameters
    ----------
    source : AsyncIterable[Element] | Iterable[Element]

        await wait(
            tasks,
            return_when=ALL_COMPLETED,
        )

    if return_exceptions:
        return [task.result() for task in results]

    return [task.result() for task in results]  # type: ignore[arg-type]


    """Merge streams from two async iterators processed concurrently.

    Concurrently consumes elements from two async iterators and yields them
    as they become available. Elements from both sources are interleaved based
    on which iterator produces them first. The function continues until both
    iterators are exhausted.

    This is useful for combining multiple async data sources into a single
    stream while maintaining concurrency. Each iterator is polled independently,
    and whichever has data available first will have its element yielded.

    Parameters
    ----------

    return await coro
           ^^^^^^^^^^
  File "/tmp/repro.py", line 10, in bad
    async def bad(): raise FakeException("boom")
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^
FakeExc
…[1850 chars omitted]…
op.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/tmp/repro.py", line 21, in main
    assert res2[1].args[0]=="odd 1" and res2[0]==2 and res2[5]==12
           ^^^^^^^^^^^^
AttributeError: 'int' object has no attribute 'args'

…[93 chars omitted]…
warnings.html#resource-warnings for more info.

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED tests/test_stream_concurrently.py::test_handles_empty_iterators - asse...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 257 passed, 2 warnings in 5.12s

  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/tmp/repro2.py", line 58, in main
    await asyncio.gather(t1,t2)
          ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 831, in gather
    fut = ensure_future(arg, loop=loop)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/tasks.py", line 689, in ensure_future
    raise TypeError('An asyncio.Future, a coroutine or an awaitable '
TypeError: An asyncio.Future, a coroutine or an awaitable is required

  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/tmp/repro2.py", line 58, in main
    await asyncio.gather(t1.result(), t2.result())
                         ^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'result'

TESTS_PATH := tests

# load environment config from .env if able
-include .env

ifndef UV_VERSION
	UV_VERSION := 0.7.20
endif

.PHONY: uv_check venv sync update format lint test docs docs-server release

# Check installed UV version and install if needed
uv_check:
	@echo 'Checking uv version...'

	# Install if not present
	@if ! command -v uv > /dev/null; then \
		echo '...installing uv...'; \
		curl -LsSf https://github.com/astra
…[6386 chars omitted]…
 --> src/haiway/helpers/concurrent.py:295:47
    |
294 |         task: Task[Result | BaseException] = (
    -             _spawn_protected(handler(element))
    -             if return_exceptions
    -             else ctx.spawn(handler, element)
295 +             _spawn_protected(handler(element)) if return_exceptions else ctx.spawn(handler, element)
296 |         )
    |

1 file would be reformatted

 
 
+class _Exhausted:
+    """Internal sentinel marking an exhausted source iterator."""
+
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return "EXHAUSTED"
+
+
+_EXHAUSTED: _Exhausted = _Exhausted()
+
+
+def _spawn_protected[Result](
+    coro: Coroutine[None, None, Result],
+    /,
+) -> Task[Result | BaseException]:
+    """Spawn a task which captures exceptions instead of propagating them.
+
+    Wrapping coroutines this way p
…[7220 chars omitted]…
        nonlocal results
-        task: Task[Result] = ctx.spawn(element)
+        task: Task[Result | BaseException] = (
+            _spawn_protected(element) if return_exceptions else ctx.spawn(element)
+        )
         results.append(task)
         tasks.add(task)
         if len(tasks) < concurrent_tasks:
@@ -438,14 +470,15 @@ async def concurrently[Result](  # noqa: C901, PLR0912
         )