标签: python-asyncio

我正在尝试为使用"async with"语句的方法编写测试(在这种情况下,aioredis的连接池),我想模拟与redis的连接,但我无法弄清楚如何.

这是我到目前为止所拥有的:

from asyncio import Future
from unittest.mock import MagicMock

import pytest

# The thing i'm trying to test
async def set_value(redis, value):
    # Do things
    async with redis.get() as conn:
        await conn.set("key", value)

#My Mock classes
class MockRedis():
    def get(self):
        return MockAsyncPool()


class MockAsyncPool(MagicMock):
    async def __aenter__(self):
        conn = MagicMock()
        f = Future()
        f.set_result(True)
        conn.set = MagicMock(return_value=f)
        return conn

    def __aexit__(self, exc_type, exc_val, exc_tb):
        pass


# The actual test
@pytest.mark.asyncio
async def test_get_token():
    redis = MockRedis()

    token = …

这个线程中的 aiohttp 服务器示例失败并显示RuntimeError: There is no current event loop in thread 'Thread-1'.错误：

import threading
from aiohttp import web


def aiohttp_server():
    def say_hello(request):
        return web.Response(text='Hello, world')

    app = web.Application(debug=True)
    app.add_routes([web.get('/', say_hello)])
    web.run_app(app)


t = threading.Thread(target=aiohttp_server)
t.start()

如何在线程中运行 aiohttp 服务器？

是一个函数，如：

async def f(x):
    time.sleep(x)

await f(5)

正确异步/非阻塞？

asyncio 提供的睡眠功能有什么不同吗？

最后，aiorequests 是请求的可行异步替代品吗？

（在我看来，它基本上将主要组件包装为异步）

https://github.com/pohmelie/aiorequests/blob/master/aiorequests.py

鉴于此示例 Starlette 应用程序具有开放的 websocket 连接，您如何关闭 Starlette 应用程序？我在 uvicorn 上运行。每当我按下Ctrl+C输出时，Waiting for background tasks to complete.它就会永远挂起。

from starlette.applications import Starlette

app = Starlette()

@app.websocket_route('/ws')
async def ws(websocket):
    await websocket.accept()

    while True:
        # How to interrupt this while loop on the shutdown event?
        await asyncio.sleep(0.1)

    await websocket.close()

我尝试在关闭事件上切换 bool 变量，但该变量永远不会更新。它总是False。

例如。

app.state.is_shutting_down = False


@app.on_event('shutdown')
async def shutdown():
    app.state.is_shutting_down = True


@app.websocket_route('/ws')
async def ws(websocket):
    await websocket.accept()

    while app.state.is_shutting_down is False:

我正在尝试创建一个 WebSocket 命令行客户端，它等待来自 WebSocket 服务器的消息但同时等待用户输入。

每秒定期轮询多个在线资源在服务器上运行良好，（localhost:6789在本示例中运行的那个），但sleep()它使用而不是使用 Python 的正常方法asyncio.sleep()，这是有道理的，因为睡眠和异步睡眠不是一回事，在至少不是在引擎盖下。

同样，等待用户输入和异步等待用户输入不是一回事，但我无法弄清楚如何以异步等待任意秒数的方式异步等待用户输入，所以客户端可以处理来自 WebSocket 服务器的传入消息，同时等待用户输入。

希望的else-clause下面的评论monitor_cmd()解释了我的意思：

import asyncio
import json
import websockets

async def monitor_ws():
    uri = 'ws://localhost:6789'
    async with websockets.connect(uri) as websocket:
        async for message in websocket:
            print(json.dumps(json.loads(message), indent=2, sort_keys=True))

async def monitor_cmd():
    while True:

        sleep_instead = False

        if sleep_instead:
            await asyncio.sleep(1)
            print('Sleeping works fine.')
        else:
            # Seems like I need the equivalent of:
            # line = await asyncio.input('Is this your line? ') …

我想编写在多个平台上mypy使用和工作的类型化代码asyncio。具体来说，我经常有显式绑定到事件循环的类和方法。我想为事件循环提供类型注释。

asyncio当我检查Linux 上的事件循环类型时，我得到：

>>> import asyncio
>>> type(asyncio.get_event_loop())
<class 'asyncio.unix_events._UnixSelectorEventLoop'>

这种类型显然与 Unix/Linux 平台相关。

现在，我可以编写使用以下类型显式键入事件循环的代码：

>>> import asyncio
>>> type(asyncio.get_event_loop())
<class 'asyncio.unix_events._UnixSelectorEventLoop'>

但您会注意到，我必须# type: ignore在导入中包含一个标签_UnixSelectorEventLoop，因为asyncio.unix_events没有类型存根。我还犹豫是否要导入一个私有方法，如类名开头的下划线所示。

作为替代方案，我可以使用AbstractEventLoop以下类型：

import asyncio
from asyncio.unix_events import _UnixSelectorEventLoop  # type: ignore
def func(loop: _UnixSelectorEventLoop) -> None:
    print(loop)
func(asyncio.get_event_loop())

这成功地通过了 mypy 类型检查。我犹豫是否用作AbstractEventLoop我的类型，因为它是一个抽象类型。

是否存在跨平台工作、不需要使用抽象类定义并通过 mypy 类型检查的替代类型签名？

我有一个用 python 和异步框架 Starlette ( https://www.starlette.io/ ) 编写的 Web 应用程序，我需要连接到 MongoDB，所以我安装了异步 pymongo 驱动程序： motor ( https://motor.readthedocs .io/en/stable/）版本 2.1.0

我有一些像这样的伪代码：

motor_client = motor.motor_asyncio.AsyncIOMotorClient(...)
pymongo_client = pymongo.MongoClient(...)

class BlogList(HTTPEndpoint):
    async def get(self, request):
        blog_list = [item async for item in motor_client.blog.find(condition)]
        # blog_list = list(pymongo_client.blog.find(condition))
        data = {
            "request": request,
            "blogs": blog_list,
        }
        return RenderPageResponse("index.html", data)

对于blog_list这里的行，我用以下方法对其进行了基准测试wrk -t10 -d10 -c100 http://localhost:8080/blog：

• 如果我使用motor_clientwith await，结果约为“1500 个请求/秒”
• 如果我将其更改为pymongo_client（同步模式），结果约为“1900 个请求/秒”

使用电机的异步模式不应该比使用 pymongo 的同步模式更快吗？我想知道为什么使用 motor（异步，1500 req/s）时的性能比 pymongo（sync，1900 …

从文档来看，启动异步应用程序的推荐方法似乎是使用asyncio.run()，所以我的应用程序如下所示：

async def async_main():
    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, ugly:
    while True:
        await asyncio.sleep(10000)

asyncio.run(async_main())

最后的无限循环async_main()感觉非常错误。在其他语言中，我会永远调用事件循环。所以我尝试了这个：

def main():
    loop = asyncio.get_event_loop()

    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, pretty:
    loop.run_forever()

main()

RuntimeError: no running event loop这里的问题是，当我在函数内部调用时asyncio.create_task()，即使事件循环是在线程上创建并注册的，这也会失败并出现各种错误。

什么是Pythonic，一种在asyncio事件循环中永远休眠的方法？

我正在使用 Python Asyncio 来执行大量 HTTP 请求。

我想知道 Asyncio 的默认并发级别是多少，或者在任何给定时间会并行发生多少个 HTTP 请求？

我用来执行 HTTP 请求的代码如下：

async def call_url(self, session, url):
    response = await session.request(method='GET', url=url)
    return response


async def main(self, url_list):
    async with aiohttp.ClientSession() as session:
        res = await asyncio.gather(*[self.call_url(session, url) for url in url_list])
        return res

pandas dataframe 中有一种apply方法允许应用一些同步功能，例如：

import numpy as np
import pandas as pd

def fun(x):
    return x * 2

df = pd.DataFrame(np.arange(10), columns=['old'])

df['new'] = df['old'].apply(fun)

fun2如果必须应用异步函数，执行类似操作的最快方法是什么：

import asyncio
import numpy as np
import pandas as pd

async def fun2(x):
    return x * 2

async def main():
    df = pd.DataFrame(np.arange(10), columns=['old'])
    df['new'] = 0    
    for i in range(len(df)):
        df['new'].iloc[i] = await fun2(df['old'].iloc[i])
    print(df)

asyncio.run(main())