标签: python-asyncio

我正在尝试将此Python2.7代码重写为新的异步世界顺序:

def get_api_results(func, iterable):
    pool = multiprocessing.Pool(5)
    for res in pool.map(func, iterable):
        yield res

map()阻塞直到计算完所有结果,所以我试图将其重写为异步实现,一旦准备就会产生结果.同样map(),返回值必须按照与之相同的顺序返回iterable.我试过这个(我需要requests因为传统的身份验证要求):

import requests

def get(i):
    r = requests.get('https://example.com/api/items/%s' % i)
    return i, r.json()

async def get_api_results():
    loop = asyncio.get_event_loop()
    futures = []
    for n in range(1, 11):
        futures.append(loop.run_in_executor(None, get, n))
    async for f in futures:
        k, v = await f
        yield k, v

for r in get_api_results():
    print(r)

但是使用Python 3.6我得到了:

  File "scratch.py", line 16, in <module>
    for r in …

我试图确保函数参数是异步函数。所以我在玩以下代码：

async def test(*args, **kwargs):
    pass

def consumer(function_: Optional[Coroutine[Any, Any, Any]]=None):
    func = function_

consumer(test)

但它不起作用。

在 pyCharm 中进行类型检查时出现以下错误：

Expected type 'Optional[Coroutine]', got '(args: Tuple[Any, ...], kwargs: Dict[str, Any]) -> Coroutine[Any, Any, None]' instead

谁能给我一些提示如何解决这个问题？

我正在编写一个使用 aiohttp 执行 http 请求的类。根据文档，我不应该为每个请求创建一个 ClientSession，所以我想重用同一个会话。

代码：

class TestApi:
   def __init__(self):
      self.session = aiohttp.ClientSession()

   # async defs methods from here 

做的时候

TestApi()

我收到错误：Unclosed client session。

持久化ClientSession对象的解决方案是什么？

等待多个异步函数并不是真正的异步工作。

例如，我期望下面运行多个await函数的代码在大约 6 秒内运行，但它像同步代码一样运行并在大约 10 秒内执行。

当我在 中尝试时asyncio.gather，它在大约 6 秒内执行。

有人可以解释为什么会这样吗？

 # Not working concurrently
 async def async_sleep(n):
    await asyncio.sleep(n+2)
    await asyncio.sleep(n)

start_time = time.time()
asyncio.run(async_sleep(4))
end_time = time.time()
print(end_time-start_time)

# Working concurrently 
async def async_sleep(n):
    await asyncio.gather(
        asyncio.sleep(n+2),
        asyncio.sleep(n)
    )

我们可以使用这两个函数来同步运行任何异步函数：

import asyncio
from asgiref.sync import async_to_sync

asyncio.run(asyncio.sleep(1))
async_to_sync(asyncio.sleep)(1)

有什么不同？我们可以总是使用asyncio.run代替async_to_sync吗？

使用装饰器为函数的返回添加类型提示的正确方法是什么@asynccontextmanager？这是我所做的两次尝试，但都失败了。

from contextlib import asynccontextmanager
from typing import AsyncContextManager


async def caller():
    async with return_str() as i:
        print(i)

    async with return_AsyncContextManager() as j:
        print(j)


@asynccontextmanager
async def return_str() -> str:
    yield "hello"


@asynccontextmanager
async def return_AsyncContextManager() -> AsyncContextManager[str]:
    yield "world"

对于 vscode 中的 Pylancei和j Pylance显示类型Any。我考虑过的其他想法：

• 我想也许我可以将类型信息传递给装饰器本身（例如@asynccontextmanager(cls=str)，但我找不到任何示例，或者我可以传递的参数的任何描述。
• async with return_str() as i: # type: str也不行。即使确实如此，我也宁愿暗示函数定义，而不是在每次调用时都暗示。类型注释不是很 DRY。
• 我尝试创建一个具有和函数AsyncContextManager的类，但没有成功。如果它有效的话我会回到那个，但我更喜欢让装饰器工作，因为它更简洁。__aenter()____aexit()__

这是我将鼠标悬停在该return_AsyncContextManager()函数上的屏幕截图，并显示 Pylance 弹出窗口显示它返回AsyncContextManager[_T]

我正在尝试使用 asyncio 进行 python 协程编程。这是我的代码。

import asyncio

async def coro_function():
    return 2 + 2

async def get():
    return await coro_function()

print(asyncio.iscoroutinefunction(get))

loop = asyncio.get_event_loop()
a1 = loop.create_task(get)
loop.run_until_complete(a1)

但是当我执行它时，它给了我错误

True
Traceback (most recent call last):
  File "a.py", line 13, in <module>
    a1 = loop.create_task(get)
  File "/home/alie/anaconda3/lib/python3.7/asyncio/base_events.py", line 405, in create_task
    task = tasks.Task(coro, loop=self)
TypeError: a coroutine was expected, got <function get at 0x7fe1280b6c80>

怎么解决呢？

我所有的代码都是在没有考虑 asyncio 的情况下编写的；但是，我使用一个异步函数（由另一位开发人员编写；出于我的目的，它是一个黑匣子）。我们就这样称呼吧func_1。我需要从另一个函数中调用这个函数，调用它func_2（它本身可以在任意长的函数链中调用func_3，func_4等等......）。

由于func_1是异步的，我需要等待它，但由于我在 中调用它func_2，所以我也需要进行func_2异步（我不能在非异步函数中等待）。这样的情况一直持续下去；我需要将整个函数链func_2, func_3,转换func_4为异步函数。

有办法避免这种情况吗？我只想调用func_1，等待它完成，然后在我的正常 python 代码的其余部分中使用结果。我可以创建一个包装器func_1来允许这样做吗？

我想要的基本上是以下内容，但不起作用：

# This is the function defined by someone else
async def func_1(*args):
    return something(*args)

# This is my wrapper
def func_1_wrapper(*args):
    return await func_1(*args)

# So that I can call it like normal within the rest of my code
def func_2(*args):
    # do something
    a …

这是我的代码的简化版本：

main是一个在第二次迭代后停止的协程。
get_numbers是一个异步生成器，它生成数字，但位于异步上下文管理器中。

import asyncio


class MyContextManager:
    async def __aenter__(self):
        print("Enter to the Context Manager...")
        return self

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        print(exc_type)
        print("Exit from the Context Manager...")
        await asyncio.sleep(1)
        print("This line is not executed")  # <-------------------
        await asyncio.sleep(1)


async def get_numbers():
    async with MyContextManager():
        for i in range(30):
            yield i


async def main():
    async for i in get_numbers():
        print(i)
        if i == 1:
            break


asyncio.run(main())

输出是：

Enter to the Context Manager...
0
1
<class 'asyncio.exceptions.CancelledError'>
Exit …

当我在Python 3.11.0上使用@asyncio.coroutine装饰器运行下面的代码时：

import asyncio

@asyncio.coroutine # Here
def test():
    print("Test")

asyncio.run(test())