标签: python-asyncio

在阅读 ASGI 规范时，我注意到WSGI 兼容性段落中的这句话：

\n

WSGI 应用程序是同步的，必须在线程池中运行才能获得服务，但否则它们的运行时会映射到 HTTP 连接作用域\xe2\x80\x99s 生命周期。

\n

我不明白为什么它必须在线程池中运行？

我最近开始研究 Python 中的异步编程。假设我们想要异步运行一个函数，如下所示：

async def print_i_async(no):
    print("Async: Preparing print of " + str(no))
    await asyncio.sleep(1)
    print(str(no))

async def main_async(no):
    await asyncio.gather(*(print_i_async(i) for i in range(no)))

asyncio.run(main_async(no))

这将按预期异步工作。然而，我不清楚如果不使用asyncio.gather(). 例如：

def print_i_serial(no):
    print("Serial: Preparing print of " + str(no))
    time.sleep(1)
    print(str(no))

for i in range(5):
    print_i_serial(i)

for i in range(5):
    asyncio.run(print_i_async(i))

这两个函数产生相同的结果。我错过了什么吗？考虑到这就是我们实际获取异步结果的方式，我们是否有任何理由使用async defif 我们不使用？asyncio.gather()

这是代码，我认为程序会因为未捕获的异常而立即崩溃。然而，当主要任务coro2完成时，它等待了 10 秒。

import asyncio

@asyncio.coroutine
def coro1():
    print("coro1 primed")
    yield
    raise Exception("abc")

@asyncio.coroutine
def coro2(loop):
    try:
        print("coro2 primed")
        ts = [asyncio.Task(coro1(),loop=loop) for _ in range(2)]
        res = yield from asyncio.sleep(10)
        print(res)
    except Exception as e:
        print(e)
        raise

loop= asyncio.get_event_loop()
loop.run_until_complete(coro2(loop))

我认为这是一个严重的问题，因为在更复杂的程序中，这会使进程永远卡住，而不是因异常信息而崩溃。

except此外，我在源代码中的块中设置了断点run_until_complete，但它没有被触发。我感兴趣的是哪一段代码在 python asyncio 中处理了该异常。

对于由p以下创建的类对象asyncio.subprocess.Process

p = await asyncio.create_subprocess_exec(...)

如何以非阻塞方式确定系统调用是否仍在运行？

我有一个代码可以读取文件目录并将文件上传到 Azure Blob 存储。它运行良好并且文件上传成功。但是，我需要帮助来修改此代码以运行异步操作以实现并发上传。

import os
import asyncio
import yaml
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__


def load_config():
    path_root = os.path.dirname(os.path.abspath(__file__))
    with open(path_root + "/config.yaml", "r") as configfile:
        return yaml.load(configfile, Loader=yaml.FullLoader)


def read_files(dir):
    with os.scandir(dir) as files:
        for filename in files:
            if filename.is_file() and not filename.name.startswith('.'):
                yield filename


def upload(files, connection_string, container_name):
    container_client = ContainerClient.from_connection_string(connection_string, container_name)
    print("Uploading images to remote blob storage")

    for file in files:
        blob_client = container_client.get_blob_client(file.name)
        with open(file.path, "rb") as data:
            blob_client.upload_blob(data)
            print(f"{file.name} upload to remote …

我需要为 aiohttp 创建工作队列。

现在我使用 asyncio.gather，但它的工作方式错误：

这就是我想做的：

第一个可以用以下代码实现：

async def some_stuff(_):
    pass

tasks = []
for i in data:
    tasks.append(do_stuff(i))

asyncio.run(asyncio.gather(*tasks))

我需要的例子

from aiohttp import ClientSession
import asyncio
import threading
import time     

 class Query():

    def execute_async(self):
    asyncio.run(self.call_requests())

    async def call_requests(self):
        self.api_request_list = []

        self.PROMETHEUS_URL = "http://www.randomnumberapi.com/api/v1.0/random?min=100&max=1000&count=1"

        self.api_request_list.append(self.api_request(self.PROMETHEUS_URL, 1))
        self.api_request_list.append(self.api_request(self.PROMETHEUS_URL, 1))
        self.api_request_list.append(self.api_request(self.PROMETHEUS_URL, 1))

        while True:
            timer_start = time.perf_counter()

            await asyncio.gather(*self.api_request_list)

            timer_stop = time.perf_counter()
            print(f"Performed all requests in... {timer_stop - timer_start:0.4f} seconds")

    async def api_request(self, query, sleep_duration):
        await asyncio.sleep(sleep_duration)
        async with ClientSession() as session:
            async with session.get(self.PROMETHEUS_URL) as response:
            response = await response.text()
            random_int = int(response[2])
            print('Request response... {0}'.format(random_int))

if __name__ == '__main__':
    query …

我在使用 HTTPX 模块时多次遇到此错误。我相信我知道这意味着什么，但我不知道如何解决。

在下面的示例中，我有一个异步函数 Gather_players()，它将 get 请求发送到我正在使用的 API，然后返回指定 NBA 球队的所有球员的列表。在 teamRoster() 内部，我使用 asyncio.run() 来启动 Gather_players() ，这就是产生此错误的行：RuntimeError: The connection pool was closed while 6 HTTP requests/responses were still in-flight

async def gather_players(list_of_urlCodes):

    async def get_json(client, link):
        response = await client.get(BASE_URL + link)

        return response.json()['league']['standard']['players']

    async with httpx.AsyncClient() as client:

        tasks = []
        for code in list_of_urlCodes:
            link = f'/prod/v1/2022/teams/{code}/roster.json'
            tasks.append(asyncio.create_task(get_json(client, link)))
        
        list_of_people = await asyncio.gather(*tasks)
        
        return list_of_people

def teamRoster(list_of_urlCodes: list) -> list:
        list_of_personIds = asyncio.run(gather_players(list_of_urlCodes))

        finalResult = []
        for person …

以下作品；这是一个成功使用异步生成器作为固定装置的测试。

from collections.abc import AsyncGenerator

import pytest

@pytest.fixture()
async def fixture() -> AsyncGenerator[str, None]:
    yield "a"

@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

但是，假设我想fixture()返回AsyncGenerator由其他函数生成的结果。在这种情况下，我收到错误：

from collections.abc import AsyncGenerator

import pytest

async def _fixture() -> AsyncGenerator[str, None]:
    yield "a"

@pytest.fixture()
async def fixture() -> AsyncGenerator[str, None]:
    return _fixture()

@pytest.mark.asyncio
async def test(fixture: str):
    assert fixture[0] == "a"

错误是：

>     assert fixture[0] == "a"
E     TypeError: 'async_generator' object is not subscriptable

我缺少什么？

我试图做一个简单的命令,我可以调用discord,让我看到目录中有多少文件,但由于某种原因,我的代码一旦运行命令就会一直遇到以下错误:

 `Traceback (most recent call last):
  File "/home/pi/MusicToaster/musicbot/bot.py", line 1995, in on_message
    response = await handler(**handler_kwargs)
  File "/home/pi/MusicToaster/musicbot/bot.py", line 1822, in cmd_audiocache
    stdout=asyncio.subprocess.PIPE)
  File "/usr/local/lib/python3.5/asyncio/subprocess.py", line 212, in create_subprocess_exec
    stderr=stderr, **kwds)
  File "/usr/local/lib/python3.5/asyncio/base_events.py", line 970, in subprocess_exec
    bufsize, **kwargs)
  File "/usr/local/lib/python3.5/asyncio/unix_events.py", line 184, in _make_subprocess_transport
    **kwargs)
  File "/usr/local/lib/python3.5/asyncio/base_subprocess.py", line 40, in __init__
    stderr=stderr, bufsize=bufsize, **kwargs)
  File "/usr/local/lib/python3.5/asyncio/unix_events.py", line 635, in _start
    universal_newlines=False, bufsize=bufsize, **kwargs)
  File "/usr/local/lib/python3.5/subprocess.py", line 950, in __init__
    restore_signals, start_new_session)
  File "/usr/local/lib/python3.5/subprocess.py", line 1540, in _execute_child
    raise child_exception_type(errno_num, …