标签: python-asyncio

我有一个大型 (1M) 数据库结果集，我想为每一行调用 REST API。

API 可以接受批量请求，但我不确定如何对rows生成器进行切片，以便每个任务处理一个行列表，例如 10 行。我宁愿不预先读取所有行并坚持使用生成器。

在一个 http 请求中发送my_function一个列表是很容易的，但是呢asyncio.gather？也许其中之一itertools可以提供帮助。

请参阅下面的通用伪代码进行说明：

async def main(rows):
    async with aiohttp.ClientSession() as session:
        tasks = [my_function(row, session) for row in rows]
        return await asyncio.gather(*tasks)

rows = <generator of database rows>
results = asyncio.run(main(rows))

注意：它们results很小，基本上是每行的确认值。

顺便说一句，

• asyncio.gather()可以（有效）处理的任务数量是否有限制？
• 当前gather()将所有请求/任务加载到内存中，消耗 50GB（！）。如何即时读取和传递行和任务以减少内存使用？这是asyncio.BoundedSemaphore()用来做什么的吗？
• TCP 连接限制为 500，因为 REST Web 服务器可以接受这个数量。如果信号量发挥作用，该值应该是多少，即设置信号量> TCP连接限制是否有意义？

aiohttp很棒asyncio但很难理解 - 我同意这篇文章：

asyncio 一直在变化，所以要警惕旧的 Stack Overflow …

我目前正在设计一个蜘蛛来抓取特定的网站。我可以同步完成，但我正在尝试了解 asyncio 以使其尽可能高效。我尝试了很多不同的方法，yield但chained functions我queues无法使其发挥作用。

我最感兴趣的是设计部分和解决问题的逻辑。不是必需的可运行代码，而是强调 assyncio 最重要的方面。我无法发布任何代码，因为我的尝试不值得分享。

使命：

example.com（我知道，应该是 example.com）有以下设计：

以同步方式，逻辑将是这样的：

for table in my_url_list:
    # Get HTML
    # Extract urls from HTML to user_list
    for user in user_list:
        # Get HTML
        # Extract urls from HTML to user_subcat_list
        for subcat in user_subcat_list:
            # extract content

但现在我想异步抓取网站。假设我们使用 5 个实例（pyppeteer 中的选项卡或 aiohttp 中的请求）来解析内容。我们应该如何设计它以使其最高效以及我们应该使用什么 asyncio 语法？

更新

感谢@user4815162342解决了我的问题。我一直在研究他的解决方案，如果其他人想使用 asyncio，我会在下面发布可运行的代码。

import asyncio
import random
 
my_url_list = ['exemple.com/table1', 'exemple.com/table2', 'exemple.com/table3']


# Random sleeps to simulate requests to the …

我想为异步函数编写一个装饰器。如何注释装饰器定义的类型？

这是我想做的一个例子：

from typing import TypeVar # will Awaitable help?
AsyncFn = TypeVar('AsyncFn') # how to narrow this type definition down to async functions?

def my_decorator(to_decorate: AsyncFn) -> AsyncFn:
    async def decorated(*args, **kwargs): # mypy keeps saying "Function is missing a type"
       return await to_decorate(*args, **kwargs)

@my_decorator
async def foo(bar: int) -> str:
    return f"async: {bar}"

@my_decorator
async def quux(spam: str, **eggs: str) -> None:
    return

foo(1) # should check
foo("baz") # mypy should yell
quux("spam") # should check
quux(1) …

尝试使我的代码更加异步并出现此错误：

find = (await loop.run_until_complete(a.finddb()[0])) \n TypeError: 'coroutine' 对象不可下标

from telethon import TelegramClient, events, Button, utils, sync\nimport pymongo\nfrom pymongo import TEXT\nimport re\nimport logging\nimport asyncio\n\nclass Search(): # search in mongodb\n    def __init__(self, search): # connect to mongoDB\n        self.search = search\n        self.myclient = pymongo.MongoClient(\n            "mongodb+srv://:@cluster0.ye4cx.mongodb.net/info?retryWrites=true&w=majority&ssl=true&ssl_cert_reqs=CERT_NONE")\n        self.mydb = self.myclient["info"]\n        self.mycol = self.mydb["comics"]\n\n\n    async def searchdb(self): # finds all comics by request\n        self.mycol.create_index([('title', TEXT)], default_language='english')\n        self.find = self.mycol.find({"$text": {"$search": self.search}})\n        if self.find.count() == 0:\n            return 0\n        else:\n            return (self.find)\n\n    async def finddb(self): …

我有以下aiohttpWebSocket 处理程序：

async def websocket_handler(request):
  ws = None
  
  if 'my-websocket-clients' not in request.app:
    request.app['my-websocket-clients'] = []
  
  print('Websocket connection starting', request.path, request.query)
  
  try:
    ws = aiohttp.web.WebSocketResponse(autoping=True, heartbeat=10.0, compress=True)
    await ws.prepare(request)
    request.app['my-websocket-clients'].append(ws)
    print('Websocket connection ready', len(request.app['my-websocket-clients']))
    async for msg in ws:
      await websocket_message(request, ws, msg)
  except asyncio.exceptions.CancelledError as e:
    print('Websocket connection was closed uncleanly ' + str(e))
    # ~~~~~~ re-raise here? ~~~~~~
  except:
    traceback.print_exc()
  finally:
    try:
      await ws.close()
    except:
      traceback.print_exc()
  
  if ws in request.app['my-websocket-clients']:
    request.app['my-websocket-clients'].remove(ws)
  
  print('Websocket connection closed', len(request.app['my-websocket-clients']))
  
  if …

我不知道如何在一项任务完成后停止循环。在示例中，当 WsServe 计数到 5 时，我预计循环会关闭。但相反，我得到了RuntimeError: Cannot close a running event loop

#!/usr/bin/env python
import asyncio

async def rxer():
    i=0
    while True:
        i+=1
        print ('Rxer ',i)
        await asyncio.sleep(1)

async def WsServe():
    for i in range(5):
        print ('WsServe',i)
        await asyncio.sleep(1)

    print ('Finish')
    loop.stop()
    loop.close()

loop=asyncio.get_event_loop()
loop.create_task(rxer())
loop.run_until_complete(WsServe())
loop.run_forever()

asyncio.sleep()\ 的阻塞表兄弟time.sleep()不能保证它会休眠所请求的时间。

\n

实际的挂起时间可能小于请求的时间，因为任何捕获的信号都将在执行该信号\xe2\x80\x99s 捕获例程后终止 sleep() 。

\n

asyncio.sleep()\ 的文档没有提到类似的限制。

是否asyncio.sleep()能够对其休眠时间做出更有力的保证？

PEP 492提到：

async with EXPR as VAR:
    BLOCK

在语义上等同于：

mgr = (EXPR)
aexit = type(mgr).__aexit__
aenter = type(mgr).__aenter__

VAR = await aenter(mgr)
try:
    BLOCK
except:
    if not await aexit(mgr, *sys.exc_info()):
        raise
else:
    await aexit(mgr, None, None, None)

然而，VAR = await aenter(mgr)它不在try区块中，所以我想知道是否__aenter__()允许失败。

例如，在此aiohttp片段中（摘自入门）：

import aiohttp
import asyncio

async def main():

    async with aiohttp.ClientSession() as session:
        async with session.get('http://python.org') as response:

            print("Status:", response.status)
            print("Content-type:", response.headers['content-type'])

            html = await response.text()
            print("Body:", html[:15], …

例子：

    with suppress(asyncio.CancelledError):
        [await t for t in asyncio.all_tasks(loop=self.loop)
            if t is not asyncio.current_task()]

为了避免Task was destroyed but it is pending!警告，我必须在取消后等待任务，但等待它们会导致终端被垃圾邮件发送CancelledError。我知道它被取消了，但我不需要看到它。

使用contextlib.suppress此处会对取消产生负面影响吗？我可以避免看到取消的错误（或任务被破坏的警告而无需等待）的唯一其他方法是使用asyncio.wait而不是启动我的初始任务asyncio.gather。由于某种原因，wait似乎抑制了异常。我用了return_when=asyncio.FIRST_EXCEPTION一个wait又return_exceptions=True一个gather。但似乎无论我如何设置它们的关键字参数，都会打印异常，而不会gather打印异常。wait

我想知道是否有某种方法可以让我轻松处理输入参数并将它们限制为 FASTAPI 中的多个值。

例如，如果我在这里有一个 hello-world 处理程序：

from fastapi import FastAPI

app = FastAPI()

@app.get(/)
async def root(name:str):
    return {"user_name_is": name}

而我想要实现的是，让用户只能输入以下名称之一作为参数 [ Bob, Jack] ，其他名称都是非法的。

编写一些进一步的检查代码即可达到预期结果并不复杂：

from fastapi import FastAPI

app = FastAPI()

@app.get(/)
async def root(name:str):
    if name in ['Bob' , 'Jack']:
         return {"user_name_is": name}
    else:
         raise HTTPException(status_code=403)

然而，编写代码仍然不够容易，尤其是当需要处理大量输入参数时。我想知道是否有一种方法可以使用类型提示和 pydantic 来达到相同的结果？

在文档中没有找到太多信息，需要帮助，谢谢。

=======

顺便说一句，如果我也有机会需要获取输入参数列表，有什么方法可以检查它们，就像下面的代码一样？

from fastapi import FastAPI
from typing import List

app = FastAPI()

@app.get(/)
async def root(names:List[str]):
    for name in names:
        if name not in …