Skip to content

Async iterator utilities

Async iterator utilities for Result.

collect(iterable, *, concurrency=None) async

Await an iterable of coroutines or tasks concurrently, collecting results into Ok[list].

Results are returned in input order. Returns the first Err encountered, cancelling remaining tasks.

concurrency limits how many run at the same time. None means unlimited — all are scheduled at once.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

>>> import asyncio
>>> async def fetch(i: int) -> Result[int, str]:
...     return Ok(i) if i > 0 else Err("bad")
>>> asyncio.run(collect([fetch(1), fetch(2), fetch(3)]))
Ok([1, 2, 3])
>>> asyncio.run(collect([fetch(1), fetch(0)], concurrency=4))
Err('bad')

Raised exceptions always arrive as an ExceptionGroup — one failure is a group of one, several failures are collected together:

>>> async def broken(source: str) -> Result[int, str]:
...     raise ConnectionError(source)
>>> async def demo(sources: list[str]) -> list[str]:
...     errors: list[str] = []
...     try:
...         await collect([broken(s) for s in sources])
...     except* ConnectionError as group:
...         errors = sorted(str(e) for e in group.exceptions)
...     return errors
>>> asyncio.run(demo(["eu"]))
['eu']
>>> asyncio.run(demo(["eu", "us"]))
['eu', 'us']
Source code in src/corrode/async_iterator.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def collect(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int | None = None,
) -> Result[list[T], E]:
    """
    Await an iterable of coroutines or tasks concurrently, collecting results into ``Ok[list]``.

    Results are returned in input order.
    Returns the first ``Err`` encountered, cancelling remaining tasks.

    *concurrency* limits how many run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        >>> import asyncio
        >>> async def fetch(i: int) -> Result[int, str]:
        ...     return Ok(i) if i > 0 else Err("bad")
        >>> asyncio.run(collect([fetch(1), fetch(2), fetch(3)]))
        Ok([1, 2, 3])
        >>> asyncio.run(collect([fetch(1), fetch(0)], concurrency=4))
        Err('bad')

        Raised exceptions always arrive as an ``ExceptionGroup`` — one failure
        is a group of one, several failures are collected together:

        >>> async def broken(source: str) -> Result[int, str]:
        ...     raise ConnectionError(source)
        >>> async def demo(sources: list[str]) -> list[str]:
        ...     errors: list[str] = []
        ...     try:
        ...         await collect([broken(s) for s in sources])
        ...     except* ConnectionError as group:
        ...         errors = sorted(str(e) for e in group.exceptions)
        ...     return errors
        >>> asyncio.run(demo(["eu"]))
        ['eu']
        >>> asyncio.run(demo(["eu", "us"]))
        ['eu', 'us']

    """
    it = iter(iterable)
    pending, next_idx = _make_pending_indexed(it, concurrency)
    indexed: dict[int, T] = {}

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            batch, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            batch.sort(key=operator.itemgetter(0))
            for idx, result in batch:
                match result:
                    case Ok(value):
                        indexed[idx] = value
                        next_item = next(it, None)
                        if next_item is not None:
                            pending.add(_wrap_indexed(next_idx, next_item))
                            next_idx += 1
                    case Err() as err:
                        return err
    finally:
        await _cancel_all(pending, it)

    return Ok([indexed[i] for i in range(len(indexed))])

map_collect(iterable, f, *, concurrency=None) async

Apply f to each element concurrently and collect into Ok[list].

Results are returned in input order. Returns the first Err produced by f, cancelling remaining tasks.

concurrency limits how many calls to f run at the same time. None means unlimited — all are scheduled at once.

Exceptions: if f raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

>>> import asyncio
>>> async def double(x: int) -> Result[int, str]:
...     return Ok(x * 2)
>>> asyncio.run(map_collect([1, 2, 3], double, concurrency=2))
Ok([2, 4, 6])
Source code in src/corrode/async_iterator.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
async def map_collect(
    iterable: Iterable[T],
    f: Callable[[T], _CoroOrTask[Result[U, E]]],
    *,
    concurrency: int | None = None,
) -> Result[list[U], E]:
    """
    Apply *f* to each element concurrently and collect into ``Ok[list]``.

    Results are returned in input order.
    Returns the first ``Err`` produced by *f*, cancelling remaining tasks.

    *concurrency* limits how many calls to *f* run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    **Exceptions**: if *f* raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        >>> import asyncio
        >>> async def double(x: int) -> Result[int, str]:
        ...     return Ok(x * 2)
        >>> asyncio.run(map_collect([1, 2, 3], double, concurrency=2))
        Ok([2, 4, 6])

    """
    return await collect(
        (f(element) for element in iterable),
        concurrency=concurrency,
    )

partition(iterable, *, concurrency=None) async

Await an iterable of coroutines or tasks concurrently, splitting results into (oks, errs).

Results are collected in input order within each list. Unlike collect, never short-circuits — all awaitables run to completion.

concurrency limits how many run at the same time. None means unlimited — all are scheduled at once.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

>>> import asyncio
>>> async def fetch(i: int) -> Result[int, str]:
...     return Ok(i) if i > 0 else Err(f"bad: {i}")
>>> asyncio.run(partition([fetch(1), fetch(-1), fetch(2)]))
([1, 2], ['bad: -1'])
Source code in src/corrode/async_iterator.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def partition(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int | None = None,
) -> tuple[list[T], list[E]]:
    """
    Await an iterable of coroutines or tasks concurrently, splitting results into ``(oks, errs)``.

    Results are collected in input order within each list.
    Unlike ``collect``, never short-circuits — all awaitables run to completion.

    *concurrency* limits how many run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        >>> import asyncio
        >>> async def fetch(i: int) -> Result[int, str]:
        ...     return Ok(i) if i > 0 else Err(f"bad: {i}")
        >>> asyncio.run(partition([fetch(1), fetch(-1), fetch(2)]))
        ([1, 2], ['bad: -1'])

    """
    it = iter(iterable)
    pending, next_idx = _make_pending_indexed(it, concurrency)
    indexed: dict[int, Result[T, E]] = {}

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            batch, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            for idx, result in batch:
                indexed[idx] = result
                next_item = next(it, None)
                if next_item is not None:
                    pending.add(_wrap_indexed(next_idx, next_item))
                    next_idx += 1
    finally:
        await _cancel_all(pending, it)

    oks: list[T] = []
    errs: list[E] = []
    for result in (indexed[i] for i in range(len(indexed))):
        match result:
            case Ok(value):
                oks.append(value)
            case Err(e):
                errs.append(e)
    return oks, errs

map_partition(iterable, f, *, concurrency=None) async

Apply f to each element concurrently and split the results into (oks, errs).

Results are collected in input order within each list. Never short-circuits — all calls run to completion.

concurrency limits how many calls to f run at the same time. None means unlimited — all are scheduled at once.

Exceptions: if f raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

>>> import asyncio
>>> async def check(i: int) -> Result[int, str]:
...     return Ok(i) if i > 0 else Err(f"bad: {i}")
>>> asyncio.run(map_partition([1, -1, 2], check, concurrency=2))
([1, 2], ['bad: -1'])
Source code in src/corrode/async_iterator.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def map_partition(
    iterable: Iterable[T],
    f: Callable[[T], _CoroOrTask[Result[U, E]]],
    *,
    concurrency: int | None = None,
) -> tuple[list[U], list[E]]:
    """
    Apply *f* to each element concurrently and split the results into ``(oks, errs)``.

    Results are collected in input order within each list.
    Never short-circuits — all calls run to completion.

    *concurrency* limits how many calls to *f* run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    **Exceptions**: if *f* raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        >>> import asyncio
        >>> async def check(i: int) -> Result[int, str]:
        ...     return Ok(i) if i > 0 else Err(f"bad: {i}")
        >>> asyncio.run(map_partition([1, -1, 2], check, concurrency=2))
        ([1, 2], ['bad: -1'])

    """
    return await partition(
        (f(element) for element in iterable),
        concurrency=concurrency,
    )

collect_all(iterable, *, concurrency=None) async

Await coroutines or tasks concurrently, accumulating all errors.

Returns Ok of all success values (in input order) only if every result is Ok; otherwise returns Err of every error (in input order). Unlike collect, never short-circuits — all awaitables run to completion, so the caller gets a complete error report.

concurrency limits how many run at the same time. None means unlimited — all are scheduled at once.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

>>> import asyncio
>>> async def fetch(i: int) -> Result[int, str]:
...     return Ok(i) if i > 0 else Err(f"bad: {i}")
>>> asyncio.run(collect_all([fetch(1), fetch(2)]))
Ok([1, 2])
>>> asyncio.run(collect_all([fetch(1), fetch(-1), fetch(-2)]))
Err(['bad: -1', 'bad: -2'])
Source code in src/corrode/async_iterator.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
async def collect_all(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int | None = None,
) -> Result[list[T], list[E]]:
    """
    Await coroutines or tasks concurrently, accumulating **all** errors.

    Returns ``Ok`` of all success values (in input order) only if every result
    is ``Ok``; otherwise returns ``Err`` of every error (in input order).
    Unlike ``collect``, never short-circuits — all awaitables run to
    completion, so the caller gets a complete error report.

    *concurrency* limits how many run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        >>> import asyncio
        >>> async def fetch(i: int) -> Result[int, str]:
        ...     return Ok(i) if i > 0 else Err(f"bad: {i}")
        >>> asyncio.run(collect_all([fetch(1), fetch(2)]))
        Ok([1, 2])
        >>> asyncio.run(collect_all([fetch(1), fetch(-1), fetch(-2)]))
        Err(['bad: -1', 'bad: -2'])

    """
    oks, errs = await partition(iterable, concurrency=concurrency)
    if errs:
        return Err(errs)
    return Ok(oks)

filter_ok_unordered(iterable, *, concurrency=None) async

Await coroutines or tasks concurrently, yielding Ok values as they complete.

Err values are silently skipped. Values are yielded in completion order, not input order.

concurrency limits how many run at the same time. None means unlimited — all are scheduled at once.

If the consumer stops iterating early (break, exception, aclose()), all in-flight tasks are cancelled and unconsumed coroutines are closed.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

async for user in filter_ok_unordered([fetch(1), fetch(2), fetch(3)]):
    print(user)
Source code in src/corrode/async_iterator.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
async def filter_ok_unordered(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[T]:
    """
    Await coroutines or tasks concurrently, yielding ``Ok`` values as they complete.

    ``Err`` values are silently skipped.
    Values are yielded in completion order, not input order.

    *concurrency* limits how many run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    If the consumer stops iterating early (``break``, exception, ``aclose()``),
    all in-flight tasks are cancelled and unconsumed coroutines are closed.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        ```python
        async for user in filter_ok_unordered([fetch(1), fetch(2), fetch(3)]):
            print(user)
        ```

    """
    it = iter(iterable)
    pending: set[asyncio.Task[Result[T, E]]] = {
        asyncio.ensure_future(item)
        for item in (it if concurrency is None else itertools.islice(it, concurrency))
    }

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            results, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            for result in results:
                match result:
                    case Ok(value):
                        yield value
                    case Err():
                        pass
                next_item = next(it, None)
                if next_item is not None:
                    pending.add(asyncio.ensure_future(next_item))
    finally:
        await _cancel_all(pending, it)

filter_err_unordered(iterable, *, concurrency=None) async

Await coroutines or tasks concurrently, yielding Err values as they complete.

Ok values are silently skipped. Values are yielded in completion order, not input order.

concurrency limits how many run at the same time. None means unlimited — all are scheduled at once.

If the consumer stops iterating early (break, exception, aclose()), all in-flight tasks are cancelled and unconsumed coroutines are closed.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

async for err in filter_err_unordered([fetch(1), fetch(2), fetch(3)]):
    print(err)
Source code in src/corrode/async_iterator.py 
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
async def filter_err_unordered(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int | None = None,
) -> AsyncIterator[E]:
    """
    Await coroutines or tasks concurrently, yielding ``Err`` values as they complete.

    ``Ok`` values are silently skipped.
    Values are yielded in completion order, not input order.

    *concurrency* limits how many run at the same time.
    ``None`` means unlimited — all are scheduled at once.

    If the consumer stops iterating early (``break``, exception, ``aclose()``),
    all in-flight tasks are cancelled and unconsumed coroutines are closed.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        ```python
        async for err in filter_err_unordered([fetch(1), fetch(2), fetch(3)]):
            print(err)
        ```

    """
    it = iter(iterable)
    pending: set[asyncio.Task[Result[T, E]]] = {
        asyncio.ensure_future(item)
        for item in (it if concurrency is None else itertools.islice(it, concurrency))
    }

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            results, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            for result in results:
                match result:
                    case Ok():
                        pass
                    case Err(e):
                        yield e
                next_item = next(it, None)
                if next_item is not None:
                    pending.add(asyncio.ensure_future(next_item))
    finally:
        await _cancel_all(pending, it)

filter_ok(iterable, *, concurrency) async

Await coroutines or tasks concurrently, yielding Ok values in input order.

Err values are silently skipped. Values are yielded in input order — later-completing tasks are buffered until all earlier ones have been yielded.

concurrency controls the size of the sliding window of in-flight tasks. Unlike filter_ok_unordered, None is not accepted because the reorder buffer would be unbounded.

If the consumer stops iterating early (break, exception, aclose()), all in-flight tasks are cancelled and unconsumed coroutines are closed.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

async for user in filter_ok([fetch(1), fetch(2), fetch(3)], concurrency=4):
    print(user)
Source code in src/corrode/async_iterator.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
async def filter_ok(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int,
) -> AsyncIterator[T]:
    """
    Await coroutines or tasks concurrently, yielding ``Ok`` values in input order.

    ``Err`` values are silently skipped.
    Values are yielded in input order — later-completing tasks are buffered until
    all earlier ones have been yielded.

    *concurrency* controls the size of the sliding window of in-flight tasks.
    Unlike ``filter_ok_unordered``, ``None`` is not accepted because the
    reorder buffer would be unbounded.

    If the consumer stops iterating early (``break``, exception, ``aclose()``),
    all in-flight tasks are cancelled and unconsumed coroutines are closed.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        ```python
        async for user in filter_ok([fetch(1), fetch(2), fetch(3)], concurrency=4):
            print(user)
        ```

    """
    it = iter(iterable)
    pending, next_idx = _make_pending_indexed(it, concurrency)
    buf: dict[int, Result[T, E]] = {}
    next_yield = 0

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            batch, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            for idx, result in batch:
                buf[idx] = result
                next_item = next(it, None)
                if next_item is not None:
                    pending.add(_wrap_indexed(next_idx, next_item))
                    next_idx += 1

            while next_yield in buf:
                match buf.pop(next_yield):
                    case Ok(value):
                        yield value
                    case Err():
                        pass
                next_yield += 1
    finally:
        await _cancel_all(pending, it)

filter_err(iterable, *, concurrency) async

Await coroutines or tasks concurrently, yielding Err values in input order.

Ok values are silently skipped. Values are yielded in input order — later-completing tasks are buffered until all earlier ones have been yielded.

concurrency controls the size of the sliding window of in-flight tasks. Unlike filter_err_unordered, None is not accepted because the reorder buffer would be unbounded.

If the consumer stops iterating early (break, exception, aclose()), all in-flight tasks are cancelled and unconsumed coroutines are closed.

Exceptions: if any coroutine raises, all remaining tasks are cancelled and every exception — including any raised while those tasks were being cancelled — propagates as a single ExceptionGroup, even when only one task failed. One failure is a group of one: the exception type you catch never depends on timing. Handle with except*.

Examples:

async for err in filter_err([fetch(1), fetch(2), fetch(3)], concurrency=4):
    print(err)
Source code in src/corrode/async_iterator.py 
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
async def filter_err(
    iterable: Iterable[_CoroOrTask[Result[T, E]]],
    *,
    concurrency: int,
) -> AsyncIterator[E]:
    """
    Await coroutines or tasks concurrently, yielding ``Err`` values in input order.

    ``Ok`` values are silently skipped.
    Values are yielded in input order — later-completing tasks are buffered until
    all earlier ones have been yielded.

    *concurrency* controls the size of the sliding window of in-flight tasks.
    Unlike ``filter_err_unordered``, ``None`` is not accepted because the
    reorder buffer would be unbounded.

    If the consumer stops iterating early (``break``, exception, ``aclose()``),
    all in-flight tasks are cancelled and unconsumed coroutines are closed.

    **Exceptions**: if any coroutine raises, all remaining tasks are cancelled and
    every exception — including any raised while those tasks were being cancelled —
    propagates as a single ``ExceptionGroup``, even when only one task failed.
    One failure is a group of one: the exception type you catch never depends
    on timing. Handle with ``except*``.

    Examples:
        ```python
        async for err in filter_err([fetch(1), fetch(2), fetch(3)], concurrency=4):
            print(err)
        ```

    """
    it = iter(iterable)
    pending, next_idx = _make_pending_indexed(it, concurrency)
    buf: dict[int, Result[T, E]] = {}
    next_yield = 0

    try:
        while pending:
            done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            batch, excs = _split_done(done)
            if excs:
                excs.extend(await _drain(pending, it))
                pending = set()
                raise BaseExceptionGroup(_GROUP_MSG, excs)
            for idx, result in batch:
                buf[idx] = result
                next_item = next(it, None)
                if next_item is not None:
                    pending.add(_wrap_indexed(next_idx, next_item))
                    next_idx += 1

            while next_yield in buf:
                match buf.pop(next_yield):
                    case Ok():
                        pass
                    case Err(e):
                        yield e
                next_yield += 1
    finally:
        await _cancel_all(pending, it)

try_reduce(iterable, initial, f) async

Await each coroutine or task sequentially, folding with f and short-circuiting on Err.

Unlike the async collect / partition family, tasks run one at a time — each awaited value is passed to f before the next is awaited, because the accumulator depends on the previous step.

On short-circuit — and on any exception — remaining tasks are cancelled and unconsumed coroutines are closed.

Exceptions: unlike the concurrent functions, execution is sequential — only one coroutine can fail — so exceptions propagate bare, without an ExceptionGroup.

Examples:

>>> import asyncio
>>> async def fetch(i: int) -> int:
...     return i
>>> def safe_add(acc: int, x: int) -> Result[int, str]:
...     return Err(f"negative: {x}") if x < 0 else Ok(acc + x)
>>> asyncio.run(try_reduce([fetch(1), fetch(2)], 0, safe_add))
Ok(3)
>>> asyncio.run(try_reduce([fetch(1), fetch(-1), fetch(3)], 0, safe_add))
Err('negative: -1')
Source code in src/corrode/async_iterator.py 
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
async def try_reduce(
    iterable: Iterable[_CoroOrTask[T]],
    initial: U,
    f: Callable[[U, T], Result[U, E]],
) -> Result[U, E]:
    """
    Await each coroutine or task sequentially, folding with *f* and short-circuiting on ``Err``.

    Unlike the async ``collect`` / ``partition`` family, tasks run one at a time —
    each awaited value is passed to *f* before the next is awaited, because the
    accumulator depends on the previous step.

    On short-circuit — and on any exception — remaining tasks are cancelled and
    unconsumed coroutines are closed.

    **Exceptions**: unlike the concurrent functions, execution is sequential —
    only one coroutine can fail — so exceptions propagate bare, without an
    ``ExceptionGroup``.

    Examples:
        >>> import asyncio
        >>> async def fetch(i: int) -> int:
        ...     return i
        >>> def safe_add(acc: int, x: int) -> Result[int, str]:
        ...     return Err(f"negative: {x}") if x < 0 else Ok(acc + x)
        >>> asyncio.run(try_reduce([fetch(1), fetch(2)], 0, safe_add))
        Ok(3)
        >>> asyncio.run(try_reduce([fetch(1), fetch(-1), fetch(3)], 0, safe_add))
        Err('negative: -1')

    """
    it = iter(iterable)
    acc: U = initial
    try:
        for item in it:
            value: T = await item
            match f(acc, value):
                case Ok(new_acc):
                    acc = new_acc
                case Err() as err:
                    return err
    finally:
        for remaining in it:
            if isinstance(remaining, asyncio.Task):
                remaining.cancel()
            else:
                remaining.close()
    return Ok(acc)