我如何编写一个高阶生成器泛型而不是同步？

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

我看到red functions 我想要他们painted black
这不是解决方案，这只是一个肮脏的黑客攻击

这是一个使用装饰器的解决方案。这有点像 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 唯一可以是可迭代的给定和其他有效同步的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]。

Answer 2

您可以将处理移至函数并将其作为参数传递给process()：

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


def process_normal(iterable, transform):
    for i in iterable:
        yield transform(i)


async def process_async(iterable, transform):
    async for i in iterable:
        yield transform(i)

您也可以将所有生成器重构为异步生成器并仅使用 process_async()，但如果没有更深入的知识，我不知道它是否适合您的需求。

最后尝试在处理之前将同步生成器转换为异步生成器。这也需要在转换后使用异步代码处理来自普通生成器的结果。

async def to_async(iterable):
    for x in iterable:
        yield x


def process(iterable):
    if not inspect.isasyncgen(iterable):
        iterable = to_async(iterable)
    return process_async(iterable)


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