我现在正在使用 AMQP pika 库。我想要开发的是上下文管理器或装饰器,以确保与 Rabbit 建立连接。问题是不可能在装饰器中使用生成器,因为它们必须返回被调用的函数。以下示例引发异常:
def ensure_conn(func):
@gen.coroutine
def wrapper(self, *args, **kwargs):
yield self.connection.ensure_connection()
return func(*args, **kwargs)
return wrapper
上下文管理器也存在几乎相同的问题。不可能使用yield两次。以下示例引发生成器未停止的异常。
@contextlib.contextmanager
@gen.coroutine
def ensure_conn(self):
yield self.ensure_connection()
yield
请推荐另一种方法?当然,我对简单的协程调用已经很满意了。谢谢。
最佳答案
实际上,有两种方法可以创建上下文管理器,这可以为您提供一些保证。就我而言,它是与 AMQP 的连接。第一种方法是重写 concurrent.futures.Future
result()
方法,强制其返回由 contextlib.contextmanager
修饰的生成器函数。 ajdavis在他漂亮的图书馆中使用了这种方法TORO 。您可以通过导航至 this line 来查看它.
但是,如果您不想覆盖 concurrent.futures.Future
对象,那么我建议您查看以下代码片段:
@gen.coroutine
def ensure_connection(*args, **kwargs):
res = yield _make_connection(timeout=kwargs.pop('timeout', 5), *args, **kwargs)
raise gen.Return(res)
@gen.coroutine
def ensure(*args, **kwargs):
res = yield ensure_connection(*args, **kwargs)
@contextlib.contextmanager
def func(res):
try:
yield # your wrapped code
finally:
pass
return func(res)
@gen.coroutine
def simple_usage(conn):
with (yield conn.ensure()) as res:
# Do your target staff here
conn = ...
IOLoop.add_callback(callback=lambda : simple_usage(conn))
关于python - Tornado 上下文管理器在 gen.coroutine 中调用,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/18866607/