标签: python-asyncio

有很多库使用其自定义版本的Future。kafka和s3transfer只是两个例子：它们所有自定义的类似未来的类都具有object超类。

毫不奇怪，您不能直接调用asyncio.wrap_future()此类对象，也不能await与它们一起使用。

包装此类 future 以便与 asyncio 一起使用的正确方法是什么？

所以我设法让 asyncio / Google CSE API 一起工作......当我在 PyCharm 上运行我的代码时，我能够打印我的结果。然而，在打印内容的最后是错误“TypeError: 'NoneType' object is not callable”。

我怀疑这与我的列表有关，也许循环试图搜索另一个术语，即使我位于列表的末尾......

另外..这是我的第一个问题帖子，所以请随时提供有关如何更好地提出问题的建议

想法？

searchterms = ['cheese',
    'hippos',
    'whales',
    'beluga']

async def sendQueries(queries, deposit=list()):
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        loop = asyncio.get_event_loop()
        futures = [
            loop.run_in_executor(
                executor,
                searching(queries)
            )
        ]
        for response in await asyncio.gather(*futures):
            deposit.append(response.json())
        return deposit

def running():
     loop = asyncio.get_event_loop()
     loop.run_until_complete(loop.create_task(sendQueries(searchterms)))
     loop.close()

print(running())
print(str(time.time() - x))

我的错误可以追溯到“等待 asyncio.gather(*futures) 中的响应：”

供您参考，搜索（查询）只是我的 Google CSE API 调用的函数。

我不确定我在这里做错了什么，我试图拥有一个包含队列并使用协程来消耗该队列上的项目的类。问题是事件循环正在一个单独的线程中运行（在该线程中我是为了loop.run_forever()让它运行）。

但我看到的是，用于消费项目的协程从未被触发：

import asyncio
from threading import Thread

import functools

# so print always flushes to stdout
print = functools.partial(print, flush=True)


def start_loop(loop):
    def run_forever(loop):
        print("Setting loop to run forever")
        asyncio.set_event_loop(loop)
        loop.run_forever()
        print("Leaving run forever")

    asyncio.set_event_loop(loop)
    print("Spawaning thread")
    thread = Thread(target=run_forever, args=(loop,))
    thread.start()


class Foo:
    def __init__(self, loop):
        print("in foo init")
        self.queue = asyncio.Queue()
        asyncio.run_coroutine_threadsafe(self.consumer(self.queue), loop)

    async def consumer(self, queue):
        print("In consumer")
        while True:
            message = await queue.get()
            print(f"Got message {message}")
            if message == "END OF QUEUE":
                print(f"exiting …

我有几个速率限制器类（其中一个未显示），我想为其创建 ABC。该request方法是一个异步上下文管理器。通过下面显示的代码我得到

“请求”的签名与超类型“RateLimiterInterface”不兼容

如果我尝试用 装饰抽象方法@asynccontextmanager，则会出现输入错误：

“asynccontextmanager”的参数 1 具有不兼容的类型“Callable[[RateLimiterInterface], Coroutine[Any, Any, AsyncIterator[Any]]]”；预期“Callable[...，AsyncIterator[]]”

我怎样才能做到这一点？

class RateLimiterInterface(abc.ABC):
    @abc.abstractmethod
    async def request(self) -> AsyncIterator:
        pass

class LeakyBucketRateLimiter(RateLimiterInterface):
    def __init__(self, max_tokens: Optional[int] = None, rate: float = 60) -> None:
        self.max_tokens = max_tokens
        self.rate = rate
        self._bucket = max_tokens
        self._last_added_at = time.time()

    @contextlib.asynccontextmanager
    async def request(self) -> AsyncIterator:
        if self._bucket is None:
            yield
            return
        while not self._bucket:
            await asyncio.sleep(0)
            self._add_tokens(int((time.time() - self._last_added_at) * self.rate))
        self._bucket -= 1
        yield
        return

    def _add_tokens(self, …

我正在测试我的消费者，它使用scope['url_route']但使用HttpCommunicatoror ApplicationCommunicator，该参数未设置。我该如何设置这个参数？测试文档非常有限，并且没有关于此https://channels.readthedocs.io/en/latest/topics/testing.html的任何文档。

测试.py

from channels.testing import HttpCommunicator
import pytest

from my_app import consumers

@pytest.mark.asyncio
async def test_consumers():
    communicator = HttpCommunicator(consumers.MyConsumer, 'GET', '/project/1')
    response = await communicator.get_response()
    assert response['status'] == 400

my_app.py

import json

from channels.db import database_sync_to_async
from channels.generic.http import AsyncHttpConsumer

from projects import model

class MyConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        _id = int(self.scope['url_route']['kwargs'].get('project_id', 1))
        record = await database_sync_to_async(models.Project.objects.get)(id=_id)
        data = json.dumps({"id": _id})
        await self.send_response(200, data.encode('utf8'))

我搭建了一个websocket服务器，它的简化版本如下所示：

import websockets, subprocess, asyncio, json, re, os, sys
from multiprocessing import Process

def docker_command(command_words):
    return subprocess.Popen(
        ["docker"] + command_words,
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)

async def check_submission(websocket:object, submission:dict):
    exercise=submission["exercise"]
    with docker_command(["exec", "-w", "badkan", "grade_exercise", exercise]) as proc:
        for line in proc.stdout:
            print("> " + line)
            await websocket.send(line)

async def run(websocket, path):
    submission_json = await websocket.recv()   # returns a string
    submission = json.loads(submission_json)   # converts the string to a python dict

    ####
    await check_submission(websocket, submission)


websocketserver = websockets.server.serve(run, '0.0.0.0', 8888, origins=None)
asyncio.get_event_loop().run_until_complete(websocketserver) …

我有一些代码可以使用asyncio和向远程站点发出请求aiohttp。

import aiohttp, asyncio

async def f():
    session = aiohttp.ClientSession()
    async with session.get('https://httpbin.org/get') as response:
        pass

asyncio.run(f())

当我运行它时，它会产生

Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f56796b3d68>
Unclosed connector
connections: ['[(<aiohttp.client_proto.ResponseHandler object at 0x7f56793836a8>, 0.0)]']
connector: <aiohttp.connector.TCPConnector object at 0x7f56796b3f98>

现在，不推荐我提出请求的方式（出于各种原因），这就是我收到这些消息的原因。但是，他们肯定不会阻止我，所以我只想再也见不到他们。

我合理地认为它们是正常的警告，并尝试通过-Wignore命令行选项禁用它们，以及将以下代码放在脚本的开头

import warnings
warnings.filterwarnings('ignore')

但消息不断出现。怎么会？我怎样才能压制他们？

如何在 asyncio 中重用服务器的套接字？而不是为每个查询创建一个新连接？

这是我的代码；

async def lookup(server, port, query, sema):
    async with sema as sema:
        try:
            reader, writer = await asyncio.open_connection(server, port)
        except:
            return {}
        writer.write(query.encode("ISO-8859-1"))
        await writer.drain()
        data = b""
        while True:
            d = await reader.read(4096)
            if not d:
                break
            data += d
        writer.close()
        data = data.decode("ISO-8859-1")
        return data

Python 3.6.6

这是代码：

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


executor_processes = ProcessPoolExecutor(2)


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    loop_ = asyncio.get_event_loop()
    loop_.run_in_executor(executor_processes, calculate)
    await asyncio.sleep(1)
    print("finish sleep")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_method())
    print("main_thread is finished")

输出：

while
finish sleep
main_thread 已完成
while
while
...

我希望子进程将被终止，就像当使用守护进程属性生成进程时一样：

import asyncio
import time
import multiprocessing


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    proc = multiprocessing.Process(target=calculate)
    proc.daemon = True
    proc.start()
    await asyncio.sleep(1)
    print("finish sleep")

if …

我运行一个由网络事件运行的 python 程序，并且不能在不处理心跳的情况下运行 10-15 秒。（更具体地说，我使用 discord.py 处理大量事件）

在一种可能的情况下，我可以使用命令将大量数据存储到数据库中，这可能会花费超过 10 到 15 秒的时间，并且会发生阻塞。

这些是数千个小型数据库调用，如果需要，我可以在这些调用之间使异步事件循环“运行其进程”。在这种情况下我怎样才能让Python“await无缘无故”呢？

类似的 hack 是等待PromiseJavaScript 中的解决，这会将进程扔回到事件循环中，首先解决更紧迫的事件。