如何捕获current.futures._base.TimeoutError - python

我正在尝试捕获run_until_complete中引发的异常,但是无论如何,我似乎都无法正确捕获它们。

这是我的最新尝试(请注意,我正在使用Pypputeer,它是Python中的Puppeteer的一个分支,它使用asyncio):

import asyncio
from pyppeteer.launcher import launch

async def test(instance):
    page = await instance.newPage()
    await page.goto('http://www.google.com', {'waitUntil': 'load', 'timeout': 1})
    await page.pdf({'path': 'example.pdf'})


async def test2():
    instance = launch(headless=True)
    try:
        task = asyncio.ensure_future(test(instance))
        print(task)
        await task
    except:
        print("Caught!")

    instance.close()


def __main__():
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(test2())
    except:
        print("ERROR")
    return 'ok'

我在这段代码中遇到的问题有两个:

如果我改用asyncio.get_event_loop,则会出现以下错误:

线程“ Thread-1”中没有当前事件循环。

如果将超时更改为适当的值,则会收到以下错误(在loop.run_until_complete(test2())处):

RuntimeError:此事件循环已在运行

如果将超时设置为1(以强制执行错误),则会出现以下指示的异常,并在控制台中显示,并显示文本“ ERROR”。 (但未发现)。

这是堆栈跟踪:

Exception in callback NavigatorWatcher.waitForNavigation.<locals>.watchdog_cb(<Task finishe...> result=None>) at /home/user/www/project/api/env/lib/python3.6/site-packages/pyppeteer/navigator_watcher.py:49
handle: <Handle NavigatorWatcher.waitForNavigation.<locals>.watchdog_cb(<Task finishe...> result=None>) at /home/user/www/project/api/env/lib/python3.6/site-packages/pyppeteer/navigator_watcher.py:49>
Traceback (most recent call last):
  File "/usr/lib64/python3.6/asyncio/events.py", line 145, in _run
    self._callback(*self._args)
  File "/home/user/www/project/api/env/lib/python3.6/site-packages/pyppeteer/navigator_watcher.py", line 52, in watchdog_cb
    self._timeout)
  File "/home/user/www/project/api/env/lib/python3.6/site-packages/pyppeteer/navigator_watcher.py", line 40, in _raise_error
    raise error
concurrent.futures._base.TimeoutError: Navigation Timeout Exceeded: 1 ms exceeded

那么,TLDR,如何捕获在异步的run_until_complete调用中引发的异常?

非常感谢!

python大神给出的解决方案

您无法捕获此错误,因为它没有在事件循环完成之前就发生:

loop.run_until_complete(test2())
print('TEST !!!')  # You will see this line, because there was no exception before

但是,如果您查看整个追溯,您将看到:

Error in atexit._run_exitfuncs:

这意味着异常发生在使用atexit处理程序注册的Pypputeer函数之一中。您应该搜索在那里捕获异常的方法,但是我不确定是否可能。

如果在执行退出处理程序期间引发异常,则
打印回溯(除非引发SystemExit),并且例外
信息已保存。在所有退出处理程序都有机会运行之后
重新提出最后一个例外。

不相关,但是never do这样的事情。

except:

子条件的python条件覆盖 - python

我试图找到一个python代码覆盖率工具,该工具可以衡量语句中是否包含子表达式:例如,我想看看下面的示例是否涵盖了condition1 / condition2 / condtion3?if condition1 or condition2 or condition3: x = true_value python大神给出的解决方案 对此的唯一合理答案是:当前…

USB设备发行 - python

我目前正在使用PyUSB。由于我不熟悉USB,所以我不知道如何执行以下操作。我已经从Python PyUSB成功连接到我的USB设备硬件。在代码中,我需要重置USB设备硬件。通过向硬件发送命令来完成。现在,在硬件重置后,我想从Python PyUSB释放当前的USB设备。然后,我想在重置后将其重新连接到USB设备硬件。请让我知道,如何释放USB设备连接和接口…

Python-熊猫描述了抛出错误:无法散列的类型“ dict” - python

更新:我正在使用“ Socrata开源API”中的一些示例代码。我在代码中注意到以下注释:# First 2000 results, returned as JSON from API / converted to Python # list of dictionaries by sodapy. 我不熟悉JSON。我已经下载了一个数据集,并创建了一个包含大量…

在Pytorch中重复张量的特定列 - python

我有一个大小为X的pytorch张量m x n和一个长度为num_repeats的非负整数n列表(假定sum(num_repeats)> 0)。在forward()方法中,我想创建一个大小为X_dup的张量m x sum(num_repeats),其中i的列X重复num_repeats[i]次。张量X_dup将在forward()方法的下游使用,因此需…

在屏幕上打印错误,但继续执行代码 - python

我有一些代码可以通过一系列URL进行迭代。如果由于其中一个URL不包含有效的JSON正文而导致我的代码中出现错误,我希望将生成的错误打印到屏幕上,然后将代码移至下一个迭代。我的代码的简单版本是:for a in myurls: try: #mycode except Exception as exc: print traceback.format_exc()…