标签: python-asyncio

import asyncio

f = open('filename.txt', 'w')

@asyncio.coroutine
def fun(i):
    print(i)
    f.write(i)
    # f.flush()

def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.as_completed([fun(i) for i in range(3)]))
    f.close()

main()

我正在尝试使用python3的新库。asyncio但是我收到此错误，不知道为什么。任何帮助，将不胜感激。

我来自Twisted背景,因此我对Twisted实施的协议和工厂有了深刻的理解.但是,我正在切换到asyncio,我在理解工厂如何集成到这个特定框架时遇到了一些麻烦.

在官方文档中,我们有一个服务器asyncio.Protocol类定义的示例.它没有用户定义的__init__功能,所以我们可以简单地调用loop.create_server(EchoServerClientProtocol, addr, port).

如果我们Protocol需要实现一些初始化逻辑会发生什么？例如,考虑这个设置最大缓冲区大小的示例:

import asyncio
from collections import deque

class BufferedProtocolExample(asyncio.Protocol):
    def __init__(self, buffsize=None):
        self.queue = deque((), buffsize)

    # ...

在Twisted中,您将创建一个Factory类来保存所有配置值,然后将其传递给初始化连接的函数.Asyncio 似乎以相同的方式工作,但我找不到任何文档.

我可以使用functools.partial,但处理这种情况的正确方法是什么？

我想要运行多个ssh使用的连接python2.7,asyncio和asyncssh.试图找出如何安装这些软件包,我发现我的python版本应该python3或更高,我需要安装trollius以运行asyncio函数python2.7.如果我需要使用asyncssh函数python2.7,我应该遵循哪些步骤,我需要处理哪些依赖项？

我的aiohttp网络服务器使用随时间变化的全局变量:

from aiohttp import web    

shared_item = 'bla'

async def handle(request):
    if items['test'] == 'val':
        shared_item = 'doeda'
    print(shared_item)

app = web.Application()
app.router.add_get('/', handle)
web.run_app(app, host='somewhere.com', port=8181)

结果是:

UnboundLocalError:赋值前引用的局部变量'shared_item'

我如何正确使用共享变量shared_item？

我是从Sanic开始的......

Sanic是一款类似Flask的Python 3.5+ Web服务器,可以快速编写.(...)除了像Flask一样,Sanic支持异步请求处理程序.这意味着您可以使用Python 3.5中新的闪亮的async/await语法,使您的代码无阻塞且快速.

......到目前为止,关于如何使用他的例子很少,而且文档也不是那么好.

按照文档基本示例,我们有

from sanic import Sanic
from sanic.response import json

app = Sanic()

@app.route("/")
async def test(request):
    return json({"test": True})

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)

例如,如何使用自定义状态代码返回自定义响应？

我的代码如下:

import asyncio
import aiohttp

urls = [
    'http://www.163.com/',
    'http://www.sina.com.cn/',
    'https://www.hupu.com/',
    'http://www.csdn.net/'
]

async def get_url_data(u):
    """
    read url data
    :param u:
    :return:
    """
    print('running ', u)
    resp = await aiohttp.ClientSession().get(url=u)
    headers = resp.headers
    print(u, headers)
    return headers


async def request_url(u):
    """
    main func
    :param u:
    :return:
    """
    res = await get_url_data(u)
    return res

loop = asyncio.get_event_loop()
task_lists = asyncio.wait([request_url(u) for u in urls])
loop.run_until_complete(task_lists)
loop.close()

当我运行我的代码时,它会显示一条警告消息:未 关闭的客户端会话

有人可以给我一些解决方案吗？

非常感谢

有没有很好的方法或受支持的库在python3中合并异步迭代器？

所需的行为与在reactx中合并可观察对象的行为基本相同。

也就是说，在正常情况下，如果要合并两个异步迭代器，则希望生成的异步迭代器按时间顺序产生结果。迭代器之一中的错误应使合并的迭代器出轨。

（来源：http : //reactivex.io/documentation/operators/merge.html）

这是我的最佳尝试，但似乎可能存在以下标准解决方案：

async def drain(stream, q, sentinal=None):
    try:
        async for item in stream:
            await q.put(item)
        if sentinal:
            await q.put(sentinal)
    except BaseException as e:
        await q.put(e)


async def merge(*streams):

    q = asyncio.Queue()
    sentinal = namedtuple("QueueClosed", ["truthy"])(True)

    futures = {
        asyncio.ensure_future(drain(stream, q, sentinal)) for stream in streams
    }

    remaining = len(streams)
    while remaining > 0:
        result = await q.get()
        if result is sentinal:
            remaining -= 1
            continue
        if isinstance(result, BaseException):
            raise result
        yield result


if …

我百思不得其解既不的发现typing.Awaitable，也没有collections.abc.Awaitable涵盖发电的协程是一个awaitable定义小号

• https://www.python.org/dev/peps/pep-0492/#await-expression

从Python 3.6开始，一些asyncioAPI（例如sleep()和open_connection()实际上返回基于生成器的协程）。我通常有应用没有问题await关键字来的返回值，但我要处理正常的价值观和的混合物awaitable S和我需要找出哪些需要await产生的实际值。

所以这是我的问题，isinstance(c, ???) == True对于基于任意生成器的协程，满足什么条件c？我不是坚持isinstance要为此目的而使用，也许getattr()可以作为解决方案...

背景

我正在开发一个基于https://blog.miguelgrinberg.com/post/unit-testing-asyncio-code的微型函数实用程序，用于异步功能的单元测试，该函数内部具有一个模拟的asyncio.Queue()返回值，并且我想增强我的实用程序，以便队列可以包含等待元素，每个元素都会触发await操作。我的代码看起来像

async def case01(loop):
  f = AsyncMock()
  f.side_effect = loop, []
  await f()  # blocks forever on an empty queue

async def case02(loop):
  f = AsyncMock()
  f.side_effect = loop, ['foo']
  await f() # => 'foo'
  await f() …

我很难理解更改后的回显服务器的行为，该服务器试图利用python 3的asyncio模块。

本质上，我有一个无限循环（可以说，在建立连接后，我想无限地将一些数据从服务器流式传输到客户端），例如MyServer.py：

#! /usr/bin/python3
import asyncio
import os
import time

class MyProtocol(asyncio.Protocol):

    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def connection_lost(self, exc):
        asyncio.get_event_loop().stop()

    def data_received(self, data):
        i = 0
        while True:
            self.transport.write(b'>> %i' %i)
            time.sleep(2)
            i+=1

loop = asyncio.get_event_loop()
coro = loop.create_server(MyProtocol, 
    os.environ.get('MY_SERVICE_ADDRESS', 'localhost'), 
    os.environ.get('MY_SERVICE_PORT', 8100))
server = loop.run_until_complete(coro)

try:
    loop.run_forever()
except:
    loop.run_until_complete(server.wait_closed())
finally:
    loop.close()

接下来，当我连接nc ::1 8100并发送一些文本（例如“测试”）时，我得到以下信息：

user@machine$ nc ::1 8100
*** Connection from('::1', 58503, 0, 0) …

为什么CancelledError没有在这个例子中被捕获？

import asyncio

q = asyncio.Queue()

async def getter():
    try:
        v = await q.get()
        print(f"getter got {v}")
    except asyncio.CancelledError:
        print("getter cancelled")

async def test():
    task = asyncio.ensure_future(getter())
    task.cancel()
    await task

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

if __name__ == '__main__':
    main()

我想要获取"getter cancelled"消息,但收到了堆栈跟踪:

Traceback (most recent call last):
  File "ce.py", line 22, in 
    main()
  File "ce.py", line 19, in main
    loop.run_until_complete(test())
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 468, in run_until_complete
    return future.result()
concurrent.futures._base.CancelledError

Task.cancel声明:

这将安排CancelledError通过事件循环在下一个循环中抛出到包装的协同程序中.然后协程有机会使用try/except/finally清理甚至拒绝请求.