python - 如何在同步上编写高阶生成器泛型？

Question

我有一堆具有相同模式的代码:

def process(iterable):
    if inspect.isasyncgen(iterable):
        return process_async(iterable)
    else:
        return process_normal(iterable)

def process_normal(iterable):
    for i in iterable:
        do_something
        yield something

async def process_async(iterable):
    async for i in iterable:
        do_something
        yield something

这允许我使用相同的代码处理正常和异步迭代，如 process()返回与它获得的相同类型的可迭代对象。然而，当循环内的处理代码不重要时，这会导致丑陋的代码重复，这并不总是可以提取，所以我想知道，是否可以更简洁地编写它，或者用一些 stdlib 替换其中的一些事物？

Answer 1

I see red functions and I want them painted black
It’s no solution, this is just a dirty hack

这是使用装饰器的解决方案。这有点像 hack，因为它深入研究了异步函数和生成器是如何在 Python 中实际实现的；这些机制记录在 PEP 492 中和 PEP 525 ，所以它们不仅仅是 CPython 的实现细节，但这仍然依赖于 Python 实现从不添加虚假的悬挂点，即使它们被判断为“无害”，这是一个有点不可靠的假设。

import functools, inspect

def generic_over_async(process_async):
    def process_sync(iterable, /, *args, **kwargs):
        async def iterable_async():
            for item in iterable:
                yield item
        agen = process_async(iterable_async(), *args, **kwargs)
        sent = None
        while True:
            try:
                gen = agen.asend(sent)
                gen.send(None)
            except StopIteration as e:
                sent = yield e.value
            except StopAsyncIteration:
                return
            else:
                gen.throw(RuntimeError,
                    f"synchronously-called function '{process_async.__name__}' has blocked")
    @functools.wraps(process_async)
    def process(iterable, /, *args, **kwargs):
        if inspect.isasyncgen(iterable):
            return process_async(iterable, *args, **kwargs)
        return process_sync(iterable, *args, **kwargs)
    return process

上面定义了一个装饰器，它采用异步生成器并添加以下逻辑:

当使用异步生成器调用时，它会按原样传递给装饰的异步生成器；

当使用常规可迭代对象调用时，它会转换为异步生成器并传递给装饰的异步生成器。然后手动驱动生成的异步生成器完成，生成所有生成的值。

成功使用这个装饰器需要被装饰的函数唯一可以做的事情 await on 是给定的迭代和其他 async有效同步的函数(即它们从不实际阻塞)；否则装饰器会抛出 RuntimeError .确保这永远不会发生，留给读者作为练习。
测试用例:

import asyncio

@generic_over_async
async def process(iterable):
    async for i in iterable:
        yield i * 2

async def blow_up(iterable):
    """ Turns an iterable into an asynchronous iterable by adding dummy suspension points """
    await asyncio.sleep(0)
    for item in iterable:
        yield item
        await asyncio.sleep(0)
        
async def main():
    print(list(process(range(5))))
    print([item async for item in process(blow_up(range(5)))])

asyncio.get_event_loop().run_until_complete(main())

以上将打印[0, 2, 4, 6, 8]两次。