标签: python-asyncio

我正在调试 FastAPI 应用程序，并且遇到了与本文中提到的问题类似的问题：asyncio.wait_for对应该超时的调用却没有超时：

try:
    await wait_for(completion_event.wait(), 1.0)
except TimeoutError:
    logging.info("timeout")
    return SubmissionResult(post_id=post_id, language_check_pending=True)

此代码片段是 FastAPI 的 POST 请求处理程序的一部分。这里，completion_event是一个asyncio.Event对象。我可以在带有 的行上放置一个断点wait_for，观察它卡住超过 1 秒的时间，然后直接移过该except块。毫无疑问，这wait_for并没有达到预期的效果。

我不知道为什么它会这样。此时，我开始怀疑 FastAPI 的内部结构，因为它使用uvloop作为“更快的异步替代品”。但我不知道如何检验这个假设，更不知道如何解决这个问题（如果确实如此）。

有什么建议么？

我希望我的 FastAPI 应用程序能够访问始终实际bot_data的python-telegram-bot。我需要这个，所以当我调用 FastAPI 中的某个端点时，可以将消息发送到所有聊天，存储在bot_data.

据我了解这个问题：bot.run_polling()并uvicorn.run(...)启动两个独立的异步循环。我需要将它们整合在一起运行。

UPD-1：
感谢@MatsLindh，我创建了下一个函数，我将其传递给主块，但它的工作不一致。有时bot.run_polling()（获得正确的循环并且一切正常，但其他时候并因存在不同循环的错误而中断）：

import asyncio
from uvicorn import Config, Server
# --snip--
def run(app: FastAPI, bot:Application):
    # using get_event_loop leads to:
    # RuntimeError: Cannot close a running event loop
    # I guess it is because bot.run_polling()
    # calls loop.run_until_complete() different tasks
    # loop = asyncio.get_event_loop()
    loop = asyncio.new_event_loop()
    server = Server(Config(app=app, port=9001))
    loop.create_task(server.serve())

    t = Thread(target=loop.run_forever)
    t.start()

    bot.run_polling()

    t.join() …

下面的测试通过了，但我怀疑我是否正确使用了 asyncio：

• 该代码混合了异步和线程
• 测试通过但从未退出（可能是因为“loop.run_until_complete”永远不会结束）

import asyncio
import threading
import pytest
import websockets

async def echo(websocket):
    async for message in websocket:
        await websocket.send(message)

async def websocket_server():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()

def _run_server():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(websocket_server())
    loop.close()

@pytest.fixture
def run_server():
    thread = threading.Thread(target=_run_server)
    thread.start()
    yield thread
    # no idea how to stop the loop here
    thread.join()


@pytest.mark.asyncio
async def test_websocket(run_server):
    async with websockets.connect("ws://localhost:8765") as websocket:
        await websocket.send("Hello!")
        response = await websocket.recv()
        assert response == "Hello!"

（注意：为了停止循环，我尝试了此处提出的解决方案（How …

我有一个Python 2.7程序,它从网站提取数据并将结果转储到数据库.它遵循消费者生产者模型,并使用线程模块编写.

只是为了好玩,我想用新的asyncio模块(从3.4)重写这个程序,但我无法弄清楚如何正确地做到这一点.

最关键的要求是程序必须按顺序从同一网站获取数据.例如,对于网址' http://a-restaurant.com ',它应首先获得' http://a-restaurant.com/menu/0 ',然后' http://a-restaurant.com/menu/ 1 ',然后' http://a-restaurant.com/menu/2 ',...如果没有提取它们,网站将完全停止提供页面,你必须从0开始.

然而另一个网站(' http://another-restaurant.com ')的另一个获取可以(并且应该)同时运行(其他网站也有sequantial限制).

线程模块非常适合这种情况,因为我可以为每个网站创建单独的线程,并且在每个线程中它可以等到一个页面完成加载,然后再获取另一个页面.

这是一个来自线程版本(Python 2.7)的极为简化的代码片段:

class FetchThread(threading.Threading)
    def __init__(self, queue, url)
        self.queue = queue
        self.baseurl = url
    ...
    def run(self)
        # Get 10 menu pages in a sequantial order
        for food in range(10):
            url = self.baseurl + '/' + str(food)
            text = urllib2.urlopen(url).read()
            self.queue.put(text)
            ...
def main()
    queue = Queue.Queue()
    urls = ('http://a-restaurant.com/menu', 'http://another-restaurant.com/menu') …

我在asyncio事件循环中遇到了一些CPU密集型任务的问题.我在处理维护传入数据缓冲区和从中构建数据包时遇到的麻烦.我已经尝试使用执行程序来执行CPU绑定的东西,但是当从中删除数据包时,无法维护缓冲区的顺序.

我正在寻找一种最佳实践方法来实现以下功能,而无需在事件循环中执行CPU绑定任务.

import asyncio
import struct

class Reader(asyncio.Protocol):
    def __init__(self):
        self.extra = bytearray()

    def data_received(self, data):
        self.extra.extend(data)
        packet = get_packet(bytes(self.extra))
        if packet:
            del self.extra[:len(packet)]
            if verify_hash(packet):  # CPU intensive
                asyncio.async(distribute(packet))  # Some asyncio fan-out callback


def get_packet(data):  # CPU intensive
    if len(data) > HEADER_SIZE:
        payload_size, = struct.unpack_from(HEADER_FORMAT, data)
        if len(data) >= HEADER_SIZE + payload_size:
            return data[:HEADER_SIZE + payload_size]
    return None

loop = asyncio.get_event_loop()
loop.run_until_complete(loop.create_server(Reader, '0.0.0.0', 8000))
loop.run_forever()

我正在研究当前的玩具代码,试图理解asyncio模块.

import asyncio
import os, sys, traceback
from time import time

os.environ['PYTHONASYNCIODEBUG'] = '1'
print(sys.version)

def timed_fib(n):
    def fib(n):
        return fib(n - 1) + fib(n - 2) if n > 1 else n
    a = time()
    return fib(n), time() - a

def process_input():
    text = sys.stdin.readline()
    n = int(text.strip())
    print('fib({}) = {}'.format(n, timed_fib(n)))


@asyncio.coroutine
def print_hello():
    while True:
        print("{} - Hello world!".format(int(time())))
        yield from asyncio.sleep(3)

def main():
    loop = asyncio.get_event_loop()
    loop.add_reader(sys.stdin, process_input)
    loop.run_until_complete(print_hello())


if __name__ == '__main__':
    main()

然而,尝试运行它会产生下面令人难以置信的神秘回溯.如您所见,调试环境变量在上面代码的第五行中设置,但是,回溯仍然非常无益,如下所示:

3.4.3rc1 …

我正在使用Discord.py创建一个机器人,并且在尝试发送带有嵌入的消息时我一直收到错误.

这是我得到的错误:

Traceback (most recent call last):
  File "C:\Users\pc\Documents\Storage\python\NanoBot\bot.py", line 101, in on_message
    await client.send_message(message.channel, embed=embed)
  File "C:\Users\pc\AppData\Local\Programs\Python\Python36-32\lib\site-packages\discord\client.py", line 1152, in send_message
data = yield from self.http.send_message(channel_id, content, guild_id=guild_id, tts=tts, embed=embed)
  File "C:\Users\pc\AppData\Local\Programs\Python\Python36-32\lib\site-packages\discord\http.py", line 198, in request
raise HTTPException(r, data)
discord.errors.HTTPException: BAD REQUEST (status code: 400)

我的代码:

embed = discord.Embed(color=target.color)
embed.set_thumbnail(url=target.avatar_url)
embed.set_author(name=str(target.name), url="Playing " + str(target.game))
embed.set_footer(text="!!userinfo command")
embed.add_field(name="Status", value=str(target.status))
embed.add_field(name="Nickname", value=str(target.nick))
embed.add_field(name="Account Created", value=str(target.created_at))
embed.add_field(name="Roles", value=str(roles))
embed.add_field(name="Joined at", value=str(target.joined_at))
await client.send_message(message.channel, embed=embed)

我有一些代码向一些API发出请求序列.我想为所有人设置通用日志记录,我该如何设置它？

假设我的代码看起来像这样

import aiohttp
import asyncio

async def fetch(client):
    async with client.get('http://httpbin.org/get') as resp:
        assert resp.status == 200
        return await resp.text()

async def post_data(client):
    async with client.post('http://httpbin.org/post', data={'foo': 'bar'}) as resp:
        assert resp.status == 200
        return await resp.text()

async def main(loop):
    async with aiohttp.ClientSession(loop=loop) as client:
        html = await fetch(client)
        print(html)
        other_html = await post_data(client)
        print(other_html)

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

现在我想查看所有请求的状态代码,网址,标题和所有请求,因此日志中的输出如下所示:

2017-08-09 08:44:30 DEBUG (200) <GET http://httpbin.org/get>
2017-08-09 08:44:30 DEBUG (200) <POST http://httpbin.org/post>

我知道我可以在每次请求后添加logger.log()调用,但这将是重复的.如果我有更多请求,我将不得不在调用logger.log的每个请求下编写重复的代码.似乎效率低下.

有一个aiohttp.client记录器,但没有详细说明如何设置它.

我正试图像这样设置它

logger = …

我正在尝试使用async从URL列表（由ID标识）中获取HTML。我需要使用代理。

我正在尝试将aiohttp与以下代理一起使用：

import asyncio
import aiohttp
from bs4 import BeautifulSoup

ids = ['1', '2', '3']

async def fetch(session, id):
    print('Starting {}'.format(id))
    url = f'https://www.testing.com/{id}'

    async with session.get(url) as response:
        return BeautifulSoup(await response.content, 'html.parser')

async def main(id):
    proxydict = {"http": 'xx.xx.x.xx:xxxx', "https": 'xx.xx.xxx.xx:xxxx'}
    async with aiohttp.ClientSession(proxy=proxydict) as session:
        soup = await fetch(session, id)
        if 'No record found' in soup.title.text:
            print(id, 'na')


loop = asyncio.get_event_loop()
future = [asyncio.ensure_future(main(id)) for id in ids]


loop.run_until_complete(asyncio.wait(future))

根据这里的问题：https : //github.com/aio-libs/aiohttp/pull/2582，似乎ClientSession(proxy=proxydict)应该工作。

但是，我收到一个错误 "__init__() …

我正在尝试使用asyncio队列运行这个简单的代码,但是捕获异常,甚至嵌套异常.

我希望得到一些帮助,使asyncio中的队列正常工作:

import asyncio, logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("asyncio").setLevel(logging.WARNING)


num_workers = 1
in_queue = asyncio.Queue()
out_queue = asyncio.Queue()
tasks = []


async def run():
    for request in range(1):
        await in_queue.put(request)

    # each task consumes from 'input_queue' and produces to 'output_queue':
    for i in range(num_workers):
        tasks.append(asyncio.create_task(worker(name=f'worker-{i}')))
    # tasks.append(asyncio.create_task(saver()))

    print('waiting for queues...')
    await in_queue.join()
    # await out_queue.join()
    print('all queues done')

    for task in tasks:
        task.cancel()
    print('waiting until all tasks cancelled')
    await asyncio.gather(*tasks, return_exceptions=True)
    print('done')


async def worker(name):
    while True:
        try:
            print(f"{name} started")
            num = …