标签: python-asyncio

我的模型后处理是使用post_save信号:

from django.core.signals import request_finished
from django.dispatch import receiver
from models import MyModel
from pipeline import this_takes_forever


@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    this_takes_forever(sender)

该this_takes_forever程序确实IO,所以我想推迟,以避免阻塞的要求太多了.

我认为这是新的asyncio模块的一个很好的用例.但是我很难在整个过程中思考.

我想我应该能够像这样调整信号接收器:

@receiver(post_save, sender=MyModel)
def my_callback(sender, **kwargs):
    loop = asyncio.get_event_loop()
    loop.run_until_complete(this_takes_forever(sender))
    loop.close()

提供this_takes_forever的还适用于协程.

@coroutine
def this_takes_forever(instance):
    # do something with instance
    return instance

这听起来太神奇了.事实上它停止了AssertionError:

AssertionError at /new/
There is no current event loop in thread 'Thread-1'.

我不知道在这种情况下我应该在哪里开始循环.有人试过这样的事吗？

我主要在IPython的交互式shell中工作.有时我会从Internet复制粘贴代码来测试它并使用示例.

如果我粘贴本教程中的以下代码:

import asyncio

async def speak_async():  
    print('OMG asynchronicity!')

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

我将关闭循环.该文件说,要使用在一个事件循环的任何方法,它已经关闭了.并且async.get_event_loop()仍然会返回那个闭环.

如果我意外关闭事件循环该怎么办？我宁愿不重启解释器.

我有一个协程如下:

async def download():
    downloader = DataManager()
    downloader.download()

DataManager.download() 方法看起来像:

def download(self):
    start_multiple_docker_containers()
    while True:
        check_containers_statuses()
        sleep(N)  # synchronous sleep from time module

这是一个好习惯吗？如果没有,我怎么能使用asyncio.sleep的download()？

或许这样的代码结构在概念上是错误的？

所以在添加try/catch块之前,当进程运行不到5分钟时,我的事件循环正常关闭,但是在添加try/catch块后,当进程超过5分钟时我开始收到此错误

async def run_check(shell_command):
    p = await asyncio.create_subprocess_shell(shell_command,
                    stdin=PIPE, stdout=PIPE, stderr=STDOUT)
    fut = p.communicate()
    try:
        pcap_run = await asyncio.wait_for(fut, timeout=5)
    except asyncio.TimeoutError:
        pass

def get_coros():
    for pcap_loc in print_dir_cointent():
        for pcap_check in get_pcap_executables():
            tmp_coro = (run_check('{args}'
            .format(e=sys.executable, args=args)))
            if tmp_coro != False:
                coros.append(tmp_coro)
     return coros

async def main(self):
    p_coros = get_coros()
    for f in asyncio.as_completed(p_coros):
        res = await f




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

追溯:

Exception ignored in: <bound method BaseSubprocessTransport.__del__ of 
    <_UnixSubprocessTransport closed pid=171106 running stdin=
    <_UnixWritePipeTransport closing fd=8 …

我想运行一个使用协同程序和多线程请求URL的服务.但是我无法将协同程序传递给执行程序中的工作程序.有关此问题的最小示例,请参阅下面的代码:

import time
import asyncio
import concurrent.futures

EXECUTOR = concurrent.futures.ThreadPoolExecutor(max_workers=5)

async def async_request(loop):
    await asyncio.sleep(3)

def sync_request(_):
    time.sleep(3)

async def main(loop):
    futures = [loop.run_in_executor(EXECUTOR, async_request,loop) 
               for x in range(10)]

    await asyncio.wait(futures)

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

导致以下错误:

Traceback (most recent call last):
  File "co_test.py", line 17, in <module>
    loop.run_until_complete(main(loop))
  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 "co_test.py", line 10, in main
    futures …

如何定义异步装置并在异步测试中使用它们？

以下代码全部在同一个文件中,失败了.是否由测试跑者明确地称之为夹具并且没有等待？

@pytest.fixture
async def create_x(api_client):
    x_id = await add_x(api_client)
    return api_client, x_id

async def test_app(create_x, auth):
    api_client, x_id = create_x
    resp = await api_client.get(f'my_res/{x_id}', headers=auth)
    assert resp.status == web.HTTPOk.status_code

生产

==================================== ERRORS ====================================
_____________ ERROR at setup of test_app[pyloop] ______________

api_client = <aiohttp.test_utils.TestClient object at 0x7f27ec954f60>

    @pytest.fixture
    async def create_x(api_client):
>       x_id = await add_x(api_client)
...
... cannot show the full trace and pathnames sorry
...    

in __await__
    ret = yield from self._coro /home/mbb/.pyenv/versions/3.6.3/envs/mr/lib/python3.6/site-packages/aiohttp/test_utils.py:245: in request
    method, self.make_url(path), *args, …

作为一项学习练习，我试图修改aiohttp的快速入门示例，以使用单个ClientSession提取多个URL（文档建议通常应为每个应用程序创建一个ClientSession）。

import aiohttp
import asyncio

async def fetch(session, url):
  async with session.get(url) as response:
    return await response.text()

async def main(url, session):
  print(f"Starting '{url}'")
  html = await fetch(session, url)
  print(f"'{url}' done")

urls = (
  "https://python.org",
  "https://twitter.com",
  "https://tumblr.com",
  "https://example.com",
  "https://github.com",
)

loop = asyncio.get_event_loop()
session = aiohttp.ClientSession()
loop.run_until_complete(asyncio.gather(
  *(loop.create_task(main(url, session)) for url in urls)
))
# session.close()   <- this doesn't make a difference

但是，在协程之外创建ClientSession显然不是可行的方法：

？python 1_async.py
1_async.py:30：用户警告：在协程之外创建客户端会话是一个非常危险的想法
  会话= aiohttp.ClientSession（）
在协程之外创建客户端会话
client_session： 
开始'https://python.org'
开始“ https://twitter.com”
开始'https://tumblr.com'
开始“ https://example.com”
开始'https://github.com'
'https://twitter.com'完成
“ https://example.com”完成
'https://github.com'完成 …

在尝试从异步生成器聚合结果时，如下所示：

async def result_tuple():
    async def result_generator():
        # some await things happening in here
        yield 1
        yield 2
    return tuple(num async for num in result_generator())

我得到一个

类型错误：“async_generator”对象不可迭代

执行该async for行时。

但PEP 530似乎暗示它应该是有效的：

异步理解

我们建议允许将 async 用于内部列表、集合和字典推导式。等待 PEP 525 批准，我们还可以允许创建异步生成器表达式。

例子：

• 集合理解：{i async for i in agen()};
• 列表理解：[i async for i in agen()];
• 字典理解：{i: i ** 2 async for i in agen()};
• 生成器表达式：（i ** 2 async for i in agen()）。

发生了什么，我如何将异步生成器聚合成一个tuple？

我正在尝试创建两个线程,每个线程都有自己的asyncio事件循环.

我尝试了以下代码,但它似乎不起作用:

import asyncio
from threading import Thread

def hello(thread_name):
    print('hello from thread {}!'.format(thread_name))

event_loop_a = asyncio.new_event_loop()
event_loop_b = asyncio.new_event_loop()

def callback_a():
    asyncio.set_event_loop(event_loop_a)
    asyncio.get_event_loop().call_soon_threadsafe(lambda: hello('a'))

def callback_b():
    asyncio.set_event_loop(event_loop_b)
    asyncio.get_event_loop().call_soon_threadsafe(lambda: hello('b'))

thread_a = Thread(target=callback_a, daemon=True)
thread_b = Thread(target=callback_b, daemon=True)
thread_a.start()
thread_b.start()

我的用例是调用Tornado web框架的websocket_connect异步函数.

我有一个异步应用程序,它通过aiohttp提供请求并执行其他异步任务(与数据库交互,处理消息,将请求本身作为HTTP客户端).我想监视事件循环的繁忙程度,也就是说它花费了多少时间来执行代码而不是等待select完成.

有没有办法用标准库事件循环或其他第三方循环(uvloop)来衡量这个？

具体来说,我想要一个持续的百分比衡量饱和度,而不仅仅是这个问题似乎要解决的二元"忙碌" .