小编Mik*_*mov的帖子

有时需要发生一些非关键的异步操作,但我不想等待它完成.在Tornado的协程实现中,您可以通过简单地省略yield关键字来"触发并忘记"异步功能.

我一直试图弄清楚如何使用Python 3.5中发布的新async/ await语法来"解雇" .例如,简化的代码段:

async def async_foo():
    print("Do some stuff asynchronously here...")

def bar():
    async_foo()  # fire and forget "async_foo()"

bar()

但是会发生什么,bar()从不执行,而是我们得到运行时警告:

RuntimeWarning: coroutine 'async_foo' was never awaited
  async_foo()  # fire and forget "async_foo()"

PEP 0492增加了新的__await__魔法.实现此方法的对象变为类似未来的对象,可以等待使用await.很明显:

import asyncio


class Waiting:
    def __await__(self):
        yield from asyncio.sleep(2)
        print('ok')

async def main():
    await Waiting()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

好的,但如果我想调用一些已async def定义的函数而不是asyncio.sleep？我不能使用await因为__await__不是async函数,我无法使用yield from因为本机协同程序需要await表达式:

async def new_sleep():
    await asyncio.sleep(2)

class Waiting:
    def __await__(self):
        yield from new_sleep()  # this is TypeError
        await new_sleep()  # this is SyntaxError
        print('ok')

我该如何解决？

这是代码:

def g():
    try:
        yield 1
        yield 2
        yield 3
    except GeneratorExit:
        yield 4

gen = g()
print(gen.__next__())

如果你运行它,你会看到:

1
Exception ignored in: <generator object g at 0x00000216BB546A98>
RuntimeError: generator ignored GeneratorExit

我理解发生了什么,但我找不到这个警告是什么类型的消息.

看起来它不是logging警告(我无法删除它设置日志记录级别).它也没有warnings警告(它看起来不像我们在通话时得到的warnings.warn).

我认为它可以与异常相关联,但我无法理解它sys.excepthook,而msg本身就消失了:

import sys

def hook(exc_type, exc_val, tb):
    print(exc_type, exc_val, tb)  # Nothing prints, while msg disappear

sys.excepthook = hook

我怎么能抓住这个警告？

如何手动创建此类警告？

QtWebKit进程的内存大小随着每个新页面加载而增加.清理内存缓存没有帮助.有谁知道如何解决它？

这个简单的例子在运行一段时间后崩溃:

from PyQt5.QtCore import QUrl
from PyQt5.QtWidgets import QApplication
from PyQt5.QtWebKitWidgets import QWebView
from PyQt5.QtWebKit import QWebSettings

class Crawler(QWebView):
    def __init__(self):
        QWebView.__init__(self)

        self.settings().setMaximumPagesInCache(0)
        self.settings().setObjectCacheCapacities(0, 0, 0)
        self.settings().setOfflineStorageDefaultQuota(0)
        self.settings().setOfflineWebApplicationCacheQuota(0)
        self.settings().setAttribute(QWebSettings.AutoLoadImages, False)

        self.loadFinished.connect(self._result_available)

    def start(self):
        self.load(QUrl('http://stackoverflow.com/'))

    def _result_available(self, ok):
        print('got it!')

        self.settings().clearMemoryCaches() # it doesn't help
        self.settings().clearIconDatabase()

        self.start() # next try

if __name__ == '__main__':
    app = QApplication([])
    crawler = Crawler()
    crawler.start()
    app.exec_()

当我们装饰功能时,我们使用functools.wraps使装饰功能看起来像原始.

当我们想要装饰类时,有没有wat做同样的事情？

def some_class_decorator(cls_to_decorate):
    class Wrapper(cls_to_decorate):
        """Some Wrapper not important doc."""
        pass
    return Wrapper


@some_class_decorator
class MainClass:
    """MainClass important doc."""
    pass


help(MainClass)

输出:

class Wrapper(MainClass)
 |  Some Wrapper not important doc.
 |  
 |  Method resolution order:
 |      Wrapper
 |      MainClass
 |      builtins.object
 |  
 # ... no MainClass important doc below.

我尝试根据functools.wraps源代码编写类装饰器的包装,但我的实现不正确:

import functools


def wraps_cls(original_cls):
    def wrapper(wrapper_cls):
        """Update wrapper_cls to look like original_cls."""
        for attr in functools.WRAPPER_ASSIGNMENTS:
            try:
                value = getattr(original_cls, attr)
            except AttributeError:
                pass
            else: …

看起来aiohttp.ProxyConnector 不支持socks代理.这有什么解决方法吗？我会很感激任何建议.

我试图用以下代码标记未来的超时完成：

import asyncio


@asyncio.coroutine
def greet():
    while True:
        print('Hello World')
        yield from asyncio.sleep(1)

@asyncio.coroutine
def main():
    future = asyncio.async(greet())
    loop.call_later(3, lambda: future.set_result(True))
    yield from future
    print('Ready')

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

“计时器”loop.call_later 将结果设置为 3 秒后的未来。它有效，但我也遇到异常：

Hello World
Hello World
Hello World
Ready
Exception in callback <bound method Task._wakeup of Task(<greet>)<result=True>>(Future<result=None>,)
handle: Handle(<bound method Task._wakeup of Task(<greet>)<result=True>>, (Future<result=None>,))
Traceback (most recent call last):
  File "C:\Python33\lib\site-packages\asyncio\events.py", line 39, in _run
    self._callback(*self._args)
  File "C:\Python33\lib\site-packages\asyncio\tasks.py", line 337, in _wakeup
    self._step(value, None)
  File "C:\Python33\lib\site-packages\asyncio\tasks.py", line 267, …

我想实现方法链接,但不是通常的函数 - 对于asyncio协同程序.

import asyncio


class Browser:
    @asyncio.coroutine
    def go(self):
        # some actions
        return self

    @asyncio.coroutine
    def click(self):
        # some actions
        return self

调用链的"直观"方式不起作用,因为单个方法返回coroutine(生成器),而不是self:

@asyncio.coroutine
def main():
    br = yield from Browser().go().click()  # this will fail

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

正确的呼叫链方式是:

br = yield from (yield from Browser().go()).click()

但它看起来很难看,并且在链条增长时变得难以理解.

有没有办法做得更好？欢迎任何想法.

我正在尝试使用streamz来管理图像处理管道。我正在使用“相机”类模拟相机，并在下面设置了管道。

import time
from skimage.io import imread
from skimage.color import rgb2gray


class Camera(object):
    imgs = [imread(f + '.jpg') for f in ['1', '2', '3']]

    @staticmethod
    def grab_one():
        return Camera.imgs[int(time.time()) % 3]

def mc_streamz():
    from streamz import Stream
    def np2jpeg(np_im):
        from six import BytesIO
        from PIL import Image
        pim = Image.fromarray(np_im)
        io = BytesIO()
        pim.save(io,format='jpeg')
        io.seek(0)
        return io.read()

    def focusmetric(np_im):
        from skimage.filters import laplace
        return laplace(np_im).var()

    def asjpegframe(b_im):
        return b'--frame\r\n'b'Content-Type: image/jpeg\r\n\r\n' + b_im + b'\r\n'

    from dask.distributed import Client
    client = …

I'm having a problem to install PycURL on the following environment

Environment

Python 3.7.0
Windows 10 

Problem

C:\>pip install pycurl
Collecting pycurl
  Using cached https://files.pythonhosted.org/packages/e8/e4/0dbb8735407189f00b33d84122b9be52c790c7c3b25286826f4e1bdb7bde/pycurl-7.43.0.2.tar.gz
    Complete output from command python setup.py egg_info:
    Please specify --curl-dir=/path/to/built/libcurl

    ----------------------------------------
Command "python setup.py egg_info" failed with error code 10 in C:\Users\user01\AppData\Local\Temp\pip-install-xrandomx\pycurl\

C:\>

According to the official site ... http://pycurl.io/docs/latest/install.html#official-packages

Currently official PycURL packages are built against the following Python versions:

2.7.10
3.2.5
3.3.5
3.4.3
3.5.2
3.6.0

Run Code Online (Sandbox Code Playgroud)

Does it mean I have to uninstall current …