标签: python-asyncio

龙卷风（或 asyncio）中是否有一种行为/模式等待 a 中的任何而不是所有 Futures list？

yield any_of([future1, future2, future3])

说future2准备好了那么结果应该是：

[None, <result>, None]

我经常看到类似于以下代码的代码(类似于aiohttp文档中的示例).

@asyncio.coroutine
def init(loop):
    srv = yield from loop.create_server(web.Application().make_handler(), '0.0.0.0', 8080)
    return srv

在下面的1行中有没有任何优点/缺点,假设你不想srv在获取它和返回之间对对象做任何事情？

@asyncio.coroutine
def init(loop):
    return (yield from loop.create_server(web.Application().make_handler(), '0.0.0.0', 8080))

为什么这行不通：

try:
    async with asyncio.wait_for(aiohttp.get(url), 2) as resp:
        print(resp.text())
except asyncio.TimeoutError as e:
    pass

给

async with asyncio.wait_for(aiohttp.get(url), 2) as resp:
AttributeError: __aexit__

据我了解，asyncio.wait_for()将传递 , 的未来aiohttp.get()，它有一个__aenter__and__aexit__方法（正如有效的事实所证明的那样async with aiohttp.get()）。

我有一个代码，正在使用侦听WebSocket上的消息aiohttp。

看起来像：

async for msg in ws:
    await self._ws_msg_handler.handle_message(ws, msg, _services)

（原始代码）ws的实例在哪里aiohttp.web.WebSocketResponse()

在我的测试中，我模拟了WebSocketResponse()它及其__aiter__方法：

def coro_mock(**kwargs):
    return asyncio.coroutine(mock.Mock(**kwargs))


@pytest.mark.asyncio
@mock.patch('aiojsonrpc.request_handler.WebSocketMessageHandler')
async def test_rpc_websocket_handler(
    MockWebSocketMessageHandler,
    rpc_websocket_handler
):

    ws_response = 'aiojsonrpc.request_handler.WebSocketResponse'
    with mock.patch(ws_response) as MockWebSocketResponse:
        MockRequest = mock.MagicMock()
        req = MockRequest()

        ws_instance = MockWebSocketResponse.return_value
        ws_instance.prepare = coro_mock()
        ws_instance.__aiter__ = coro_mock(return_value=iter(range(5)))
        ws_instance.__anext__ = coro_mock()

        handle_msg_result = 'Message processed'
        MockWebSocketMessageHandler.handle_message.side_effect = Exception(
            handle_msg_result)
        msg_handler = MockWebSocketMessageHandler()

        with pytest.raises(Exception) as e:
            await request_handler.RpcWebsocketHandler(msg_handler)(req)
        assert str(e.value) …

我正在研究 Python 3.5.1，我希望能够判断一个函数是否作为协程对象返回，但我找不到协程类型的定义位置，最近我一直在使用下面的代码片段来获取通过用函数实例化协程来确定类型。

async def _f():
    pass
COROUTINE_TYPE = type(_f())

必须有更好的方法来做到这一点，我的问题是这种类型在哪里定义以便我可以直接使用它？

我使用的是uvloop与websockets作为

import uvloop
coro = websockets.serve(handler, host, port)  # creates new server
loop = uvloop.new_event_loop()
loop.create_task(coro)
loop.run_forever()

一切正常，我只是想知道如果不将全局asyncio政策设置为，是否可以遇到一些意想不到的问题uvloop。据我了解，只要不使用全局asyncio方法，只要不使用全局方法，就可以设置全局策略，而直接使用传递的事件循环。那是对的吗？

我在asyncio上的TCP客户端和服务器上找到了这些示例：tcp server example。但是如何连接它们以获取将接收数据并将其发送到其他地址的TCP代理服务器？

将异步用于套接字服务器的推荐方法是：

import asyncio

async def handle_client(reader, writer):
    request = (await reader.read(100)).decode()
    response = "Data received." 
    writer.write(response.encode())

async def main():
    loop.create_task(asyncio.start_server(handle_client, 'localhost', 15555))

loop = asyncio.get_event_loop()
loop.create_task(main())
loop.run_forever()

这可以正常工作，但是现在我需要接收适当的客户端请求，然后使用aiohttp库从第三方的Restful API获取数据。

这需要创建一个会话变量，如下所示：

from aiohttp import ClientSession

session = ClientSession()

但这也应该在协程本身内部，因此我将其放入main中：

async def main():
    session = ClientSession()
    loop.create_task(asyncio.start_server(handle_client, '', 55555))

现在，我需要将会话变量传递给aiohttp get coroutine来获取其余的API数据：

async with session.get(url, params=params) as r:
    try:
        return await r.json(content_type='application/json')
    except aiohttp.client_exceptions.ClientResponseError:
        ....

我的问题是，如果它坚持只具有读取器，写入器参数，而全局变量对协程内部必须存在会话却没有帮助，那么如何将会话变量传递给handle_client协程？

我想创建一个使用aiohttp进行API调用的调度程序类。我尝试了这个：

import asyncio
import aiohttp

class MySession:
    def __init__(self):
        self.session = None

    async def __aenter__(self):
        async with aiohttp.ClientSession() as session:
            self.session = session
            return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

async def method1():
    async with MySession() as s:
        async with s.session.get("https://www.google.com") as resp:
            if resp.status == 200:
                print("successful call!")

loop = asyncio.get_event_loop()
loop.run_until_complete(method1())
loop.close()

但这只会导致错误：RuntimeError: Session is closed。该__aenter__功能的第二种方法：

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

效果很好。这是一个好结构吗？它不遵循如何使用aiohttp的示例。还想知道为什么第一种方法不起作用？

我了解如何向将来添加回调方法，并在将来完成后调用它。但是，当您已经可以从协程内部调用函数时，这为何有用呢？

回调版本：

def bar(future):
    # do stuff using future.result()
    ...

async def foo(future):
    await asyncio.sleep(3)
    future.set_result(1)

loop = asyncio.get_event_loop()
future = loop.create_future()
future.add_done_callback(bar)
loop.run_until_complete(foo(future))

选择：

async def foo():
    await asyncio.sleep(3)
    bar(1)

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

什么时候第二个版本不可用/不合适？