标签: python-asyncio

使用asyncio可以在超时时执行协程,以便在超时后取消:

@asyncio.coroutine
def coro():
    yield from asyncio.sleep(10)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(coro(), 5))

上面的示例按预期工作(它在5秒后超时).

但是,当协程不使用asyncio.sleep()(或其他asyncio协同程序)时,它似乎没有超时.例:

@asyncio.coroutine
def coro():
    import time
    time.sleep(10)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(coro(), 1))

运行时间超过10秒,因为time.sleep(10)未取消.在这种情况下是否可以强制取消协程？

如果应该使用asyncio来解决这个问题,我该怎么做？

我找不到aiohttp与登录页面结合的工作代码.目标很简单:使用用户名和密码进行基于表单的身份验证,我希望在随后的aiohttp async fetch调用中使用该cookie.

似乎整个Session概念在版本之间的aiohttp中发生了变化,所以我很好奇如何在最新版本中实现它.我不确定如何获取cookie一次,然后在异步事件中使用它.

我真的很想看到一个完整的例子,因为不幸的是我无法使用我在任何地方找到的片段.

我想这可能是一个开始,但我不确定,我当然不知道如何将所有内容连接到它(我还需要一个aiohttp.TCPConnector吗？) http://aiohttp.readthedocs.org/en/latest/ client_reference.html#aiohttp.client.ClientSession

使用mechanize在Python 2中我的非异步版本的示例(虽然我自然地使用Python 3进行asyncio等):

import mechanize
import urllib

class MyClass()
    def __init__(self):
        self.data = {'username' : 'me', 'password' : 'pw'}
        self.login_url = 'http://example.com/login'
        self.login()

    def call(self, url):
        request2 = mechanize.Request(url)
        self.cookie_jar.add_cookie_header(request2)
        response2 = mechanize.urlopen(request2).read()
        return response2    

    def login(self):
        request = mechanize.Request(self.login_url)
        # 'username' and 'password' keys are actually the name of the <input>
        logInfo = urllib.urlencode({'username' : self.data['username'], 
                                    'password' : self.data['password']})
        response = mechanize.urlopen(request, data = logInfo) …

我正在尝试async使用asyncio事件循环创建一种对绑定方法的计时器回调.现在的问题是绑定的异步方法不应该保存对实例的强引用,否则后者永远不会被删除.定时器回调应该只与父实例一样长.我找到了一个解决方案,但我认为它并不漂亮:

import asyncio
import functools
import weakref

class ClassWithTimer:
    def __init__(self):
        asyncio.ensure_future(
            functools.partial(
                ClassWithTimer.update, weakref.ref(self)
            )()
        )

    def __del__(self):
        print("deleted ClassWithTimer!")

    async def update(self):
        while True:
            await asyncio.sleep(1)
            if self() is None: break
            print("IN update of object " + repr(self()))

async def run():
    foo = ClassWithTimer()
    await asyncio.sleep(5)
    del foo

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

有没有更好,更pythonic的方式来做到这一点？计时器回调确实需要异步.没有asyncio,weakref.WeakMethod可能是要走的路.但是asyncio.ensure_future需要一个协程对象,所以在这种情况下它不起作用.

有人可以提供一个示例代码，这些代码示例使用asynio以非阻塞方式监听按键并在每次单击时将其放入控制台吗？

这不是一些图形工具包的问题

如何找出事件循环中有多少活动任务？在文档中我只找到asyncio.Task.all_tasks()但它是简单的增量计数器:

import asyncio


async def coro():
    await asyncio.sleep(1)


async def main():
    tasks = []
    print('Tasks count: ', len(asyncio.Task.all_tasks()))
    for idx in range(3):
        task = asyncio.ensure_future(coro())
        tasks.append(task)
        print('Tasks count: ', len(asyncio.Task.all_tasks()))
    await asyncio.gather(*tasks)
    print('Tasks count: ', len(asyncio.Task.all_tasks()))


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

输出:

Tasks count:  1
Tasks count:  2
Tasks count:  3
Tasks count:  4
Tasks count:  4

预期产量:

Tasks count:  1
Tasks count:  2
Tasks count:  3
Tasks count:  4
Tasks count:  1

我在asyncio事件循环中运行了一些Python3代码.

我想使用它的功能os.chmod(...),但理想情况下会像这样的非阻塞版本,以便我可以使用await os.chmod(...),并避免进行阻塞系统调用.

我不相信有任何可用的库提供此功能,至少从我能看到的.

我如何os.chmod(...)从头开始实现非阻塞？更好的是,是否存在预先存在的解决方案？

我正在尝试在 python 异步循环中接收 UDP 数据包。我对 asyncio 很陌生，所以我可能做错了什么，因为回调永远不会被调用：

import asyncio

class DiscoveryProtocol(asyncio.DatagramProtocol):
    def __init__(self):
        super().__init__()
    def connection_made(self, transport):
        self.transport = transport
    def datagram_received(self, data, addr):
        print(data)

def start_discovery():
    loop = asyncio.get_event_loop()
    t = loop.create_datagram_endpoint(DiscoveryProtocol,local_addr=('',5006))
    loop.run_until_complete(t)
    loop.run_forever()

我可以使用普通的旧套接字接收数据包（没有异步）。

我究竟做错了什么？

我正在尝试从FTP服务器下载超过30,000个文件,并且在使用异步IO进行一些谷歌搜索之后似乎是一个好主意.但是,下面的代码无法下载任何文件并返回超时错误.我真的很感激任何帮助!谢谢!

class pdb:
    def __init__(self):
        self.ids = []
        self.dl_id = []
        self.err_id = []


    async def download_file(self, session, url):
        try:
            with async_timeout.timeout(10):
                async with session.get(url) as remotefile:
                    if remotefile.status == 200:
                        data = await remotefile.read()
                        return {"error": "", "data": data}
                    else:
                        return {"error": remotefile.status, "data": ""}
        except Exception as e:
            return {"error": e, "data": ""}

    async def unzip(self, session, work_queue):
        while not work_queue.empty():
            queue_url = await work_queue.get()
            print(queue_url)
            data = await self.download_file(session, queue_url)
            id = queue_url[-11:-7]
            ID = id.upper()
            if …

这是简单的测试代码和结果.

import asyncio

async def test():
    await asyncio.sleep(1)

if __name__ == '__main__':

    asyncio.set_event_loop(None)      # Clear the main loop.
    loop = asyncio.new_event_loop()   # Create a new loop.
    loop.run_until_complete(test())   # Run coroutine over the new loop

Traceback (most recent call last):
  File "test_test.py", line 11, in <module>
    loop.run_until_complete(test())
  File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
    return future.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 239, in _step
    result = coro.send(None)
  File "test_test.py", line 5, in test
    await asyncio.sleep(1)
  File …

我有一个 bog 标准同步 python 程序，需要能够从 websockets 读取数据并使用数据更新 GUI。然而，asyncio 的蠕动总是让我绊倒。

我如何制作一个模块：

1. 接受多个来源的多个订阅
2. 每当有数据时就向请求者发送更新
3. 每个 URL 只打开一个 websocket 连接
4. 如果 websocket 关闭则重置它

这是我已经拥有的，但它在很多方面都失败了：

1. run_forever()意味着循环在订阅完成之前被卡住，然后handle()卡在 falseywhile循环中
2. 它似乎不想在套接字关闭时重新启动套接字，因为 websockets 对象没有属性connected（没有 s 的 websocket 有，但我不清楚其中的差异，也无法在网上找到信息）
3. 我绝对不确定我的方法是否正确。

已经为此奋斗了几个星期。希望得到一些指点。

class WSClient():
    subscriptions = set()
    connections = {}
    started = False

    def __init__(self):
        self.loop = asyncio.get_event_loop()

    def start(self):
        self.started = True
        self.loop.run_until_complete(self.handle())
        self.loop.run_until_forever()  # problematic, because it does not allow new subscribe() events

    async def handle(self):
        while len(self.connections) > 0:
            # …