如何在Python 3中嵌套异步上下文管理器
在Python 3中，可以使用asyncio模块来实现异步编程。异步上下文管理器是一种特殊的上下文管理器，它允许在异步代码中使用with语句来管理资源的获取和释放。
要在Python 3中嵌套异步上下文管理器，可以按照以下步骤进行操作：
- 导入
asyncio模块：首先，需要导入asyncio模块来使用异步编程的功能。
import asyncio- 创建异步上下文管理器：使用
async关键字定义一个异步上下文管理器，并在其中实现__aenter__和__aexit__方法。__aenter__方法用于获取资源，__aexit__方法用于释放资源。
class AsyncContextManager:
async def __aenter__(self):
# 获取资源的异步操作
return resource
async def __aexit__(self, exc_type, exc_val, exc_tb):
# 释放资源的异步操作
pass- 嵌套使用异步上下文管理器：在异步代码中使用
with语句来嵌套使用异步上下文管理器。可以使用async with语句来异步获取和释放资源。
async def nested_async_context_manager():
async with AsyncContextManager() as resource1:
# 使用 resource1 的异步操作
async with AsyncContextManager() as resource2:
# 使用 resource2 的异步操作在上述代码中，AsyncContextManager是一个自定义的异步上下文管理器，可以根据实际需求进行定义和实现。在nested_async_context_manager函数中，使用async with语句嵌套使用两个异步上下文管理器，分别获取和释放资源。
需要注意的是，异步上下文管理器只能在异步函数中使用，即函数定义时需要使用async def关键字声明为异步函数。
关于腾讯云相关产品和产品介绍链接地址，可以参考腾讯云官方文档或咨询腾讯云官方客服获取更详细的信息。
相关·内容
2回答
我有一群上下文管理人员，我想把他们联系起来。乍一看，contextlib.nested看起来是一个合适的解决方案。但是，该方法在文档中被标记为不推荐使用，该文档还指出，最新的with语句直接允许这样做：
从2.7版开始就不推荐了:with-语句现在直接支持这个功能(没有容易出错的令人困惑的怪癖)。
但是，我无法让Python3.4.3使用上下文管理器的动态迭代：
class Foo():
def __enter__(self):
print('entering:', self.name)
return self
def _
浏览 1提问于2015-06-22得票数 5
回答已采纳
我一直在使用许多上下文管理器作为组合各种设置/拆卸情况的干净方法。因为我的部署目标是Python2.6，所以这意味着使用contextlib.nested。
最近，我对用相同的代码库支持Python2.x和Python3很感兴趣。这在中是可能的，但是我在上下文管理器方面遇到了麻烦，因为：
Python3不支持contextlib.nested
在2.6中不支持Python-3风格的嵌套上下文管理器(例如with aa() as a, bb() as b: ...)。
这里有一个基本的句法不兼容。由于我无法控制的各种原因，2.7目前可能很难投入生产，但我希望将来尽可能多地验证代码
浏览 3提问于2013-10-01得票数 2
回答已采纳
我正试图在我们学校的计算机集群上部署地理空间联合脚本。不幸的是，他们有python2.6(Cloudera的标准)，我不得不相应地重写我的代码。
读取pandas.from_sql()需要建立SQLAlchemy连接。但是，他们提出的方法在python2.6中不起作用：
engine = create_engine('sqlite:///%s' % path)
with engine.connect() as conn, conn.begin():
chunks = pd.read_sql_table('tweets', conn, columns=c
浏览 5提问于2016-03-03得票数 0
回答已采纳
我有一行python代码，如下所示：
with long_name_function(p) as a, other_long_name_function():
我想把它分成多行，因为太长了，我可以使用反斜杠，但它们被认为是一种糟糕的做法。我也可以使用contextlib.nested，但已经被弃用了，有没有其他选择？
浏览 0提问于2013-04-18得票数 22
回答已采纳
4回答
在Python2.6中，我们用这种方式格式化嵌套的上下文管理器：
with nested(
context1,
context2
) as a, b:
pass
从Python2.7和on开始，nested是不可取的。我在单行上看到了很多多上下文管理器的例子，但是我找不到允许它们出现在多行上的语法。你会怎么做？
# That's working fine
with context1 as a, context2 as b:
pass
# But how do we make it multine?
# These are not working
wit
浏览 4提问于2015-07-22得票数 32
回答已采纳
1回答
我有异步脚本，它每40秒用aio_pika库连接到rabbitmq，并检查是否有任何消息并打印出来，然后永远重复。但是，通常在运行2天左右之后，我将开始接收无休止的连接异常错误，这些错误只能通过重新启动脚本来解决。也许在我的异步脚本的逻辑中有一些明显的错误，而我错过了这些错误吗？
#!/usr/bin/python3
import time
import async_timeout
import asyncio
import aio_pika
async def got_message(message: aio_pika.IncomingMessage):
with message.
浏览 3提问于2019-04-01得票数 1
回答已采纳
1回答
这里来自python3.4的新手开发人员。
我天真的理解是，只有当我看到协同线是上下文管理器时才使用关键字async with？
浏览 0提问于2018-01-21得票数 12
回答已采纳
我在Ubuntu 16.04上使用aiohttp V2.0.7和Python 3.5.2，在代码中嵌套块有问题，如下所示：
while True:
try:
async with aiohttp.ClientSession() as session:
with contextlib.closing(MyServerSession(loop, session)) as ss:
# this was simplified for testing!
done, pending = awai
浏览 1提问于2017-05-15得票数 0
1回答
在python中可以组合异步上下文管理器吗？类似于asyncio.gather，但能够与上下文管理器一起使用。就像这样：
async def foo():
async with asyncio.gather_cm(start_vm(), start_vm()) as vm1, vm2:
await vm1.do_something()
await vm2.do_something()
这是目前的可能吗？
浏览 1提问于2018-10-29得票数 10
回答已采纳
我有一些这样的代码：
with SomeContext(args):
<statement1>
.
.
<statementN>
我希望这段代码的行为像这样：
if some_condition(args):
f1()
else:
<statement1>
.
.
<statementN>
else块中的语句还需要访问args。
但是我想对这个抽象的用户隐藏f1和some_condition，所以使用if-else块是不可能的。另外，我不想强制要求用户将所有语句包装在一个函数中。有没有可能在允许这样做的with
浏览 0提问于2012-11-30得票数 2
回答已采纳
提到：
async with EXPR as VAR:
BLOCK
在语义上相当于：
mgr = (EXPR)
aexit = type(mgr).__aexit__
aenter = type(mgr).__aenter__
VAR = await aenter(mgr)
try:
BLOCK
except:
if not await aexit(mgr, *sys.exc_info()):
raise
else:
await aexit(mgr, None, None, None)
但是，VAR = await aenter(mgr)不在t
浏览 1提问于2021-01-25得票数 1
回答已采纳
假设我有两个
@contextmanager
def foo():
[...]
@contextmanager
def bar():
[...]
在Python2.7中，我可以在下面这样的代码块中使用这两种方法：
def do_something():
with foo(), bar():
[...]
但是，我如何将这两者组合成一个单一的上下文管理器(为了保持干爽/清晰)：
@contextmanager
def foobar():
[???]
def do_something():
with foobar():
[.
浏览 4提问于2014-11-27得票数 2
给定以下ContextManager类，下面是下面示例中__exit__调用的顺序--反向声明顺序。
有趣的是，__exit__可以被多次调用。在2.7版的Python的所有版本中，这是否都是确定性的？如果不是，在中，这个顺序是否可靠地确定为的哪个版本？
a = A()
b = B()
c = C()
with a, b, c, a, a:
print("Start")
测试线束
from typing import ContextManager
class A(ContextManager):
def __exit__(self, __exc_type,
浏览 2提问于2021-07-04得票数 4
回答已采纳
1回答
我使用Python2.7，我知道我可以这样写：
with A() as a, B() as b:
do_something()
我想提供一个兼具两者的方便帮手。这个助手的用法应该如下所示：
with AB() as ab:
do_something()
现在AB()应该同时做这两件事:创建上下文A()和创建上下文B()。
我不知道如何写这个方便助手。
浏览 0提问于2017-08-09得票数 23
回答已采纳
2回答
我有以下代码可以使用async和websokets python库与websocket交互：
#!/usr/bin/env python3
import sys, json
import asyncio
from websockets import connect
class AsyncWebsocket():
async def __aenter__(self):
self._conn = connect('wss://ws.myws.com/v2')
self.websocket = await self._conn.__aent
浏览 5提问于2017-02-06得票数 6
回答已采纳
2回答
我希望创建一个使用aiohttp进行API调用的调度器类。我试过这个：
import asyncio
import aiohttp
class MySession:
def __init__(self):
self.session = None
async def __aenter__(self):
async with aiohttp.ClientSession() as session:
self.session = session
return self
async def __
浏览 0提问于2018-11-09得票数 6
回答已采纳
Python3.5用新的函数定义语法大大扩展了对异步编程的支持。而异步函数以前只是“有好处的生成器”：
def generate_numbers():
"""
Generator function that lazily returns 1 - 100
"""
for i in range 100:
yield i
generate_async = asyncio.coroutine(generate_numbers)
generate_async.__doc__ = """
浏览 1提问于2016-11-13得票数 3
回答已采纳
1回答
我有一个简单的python异步程序，它从特定的URL中检索gzipped文件。我已经将aiohttp用于异步请求。根据aiohttp ()，我在我的测试方法中使用了他们在“流响应内容”下的例子来编写数据。
async def main(url):
async with aiohttp.ClientSession() as session:
await asyncio.gather(test(session, url))
async def test(session, url):
async with session.get(url=url) as r:
浏览 9提问于2022-02-27得票数 0
回答已采纳
我有以下几点：
with open("c:\xml1.txt","r") as f1, open('c:\somefile.txt','w') as f2:
这给出了一个语法错误：
with open("c:\xml1.txt","r") as f1, open('c:\somefile.txt','w') as f2:
^
SyntaxError: mismatched input '
浏览 0提问于2013-05-16得票数 4
回答已采纳
我想创建可选的异步信号量。
在asyncio.Semaphore不支持None值的情况下，如果指定了连接限制，我决定创建asyncio.Semaphore，否则就创建某种虚拟对象。
有一个contextlib.nullcontext，但它只支持同步的with。
我创造了自己的假人：
@contextlib.asynccontextmanager
async def asyncnullcontext():
yield None
有没有默认的异步空上下文管理器？
浏览 2提问于2020-04-28得票数 7
回答已采纳
我正在尝试理解python中的with语句。无论我往哪里看，它都在谈论打开和关闭一个文件，它的目的是取代try-finally块。有没有人可以再贴一些其他的例子。我只是在试用flask，里面有很多语句。一定要请某人提供一些清晰的信息。
浏览 0提问于2010-09-28得票数 4
回答已采纳
我试图使用背景队列解析XML文档并将其数据存储在核心数据存储中。我正在使用在UIManagedObjectContext 5中引入的新的嵌套iOS概念。
我向我的NSManagedObject公司添加了一个新的类别，它处理公司特定数据的解析：
- (void)parseAttributesFrom:(NSString*)xmlStr
inManagedObjectContext:(NSManagedObjectContext*)managedObjectContext
{
NSManagedObjectContext * context = [[NSMan
浏览 3提问于2012-06-24得票数 1
回答已采纳
我正在为多线程应用程序使用threading.RLock。我希望让线程尝试获取一个锁，如果不成功，请重试一些次，前提是重试所花费的时间低于某个超时阈值。
这是我拥有的
>>> import threading
>>> lock = threading.RLock()
>>> def worker():
with lock.acquire(timeout=5):
print('acquired')
>>> lock.acquire()
>>> t2 =
浏览 0提问于2017-10-09得票数 0
回答已采纳
这是在对的回答中提到的，但我想问一个关于这个问题的新问题。
如何不使用上下文管理器连接到websocket？当需要使用它的协同线实际被调用时，我在关闭websocket时遇到了麻烦。
ws = websockets.connect('ws://192.168.1.32:8765')
print(ws)
提供<websockets.py35.client.Connect object at 0x0076CAD0>，而不是我从使用上下文管理器获得的<websockets.client.WebSocketClientProtocol object at 0x04A9
浏览 2提问于2017-03-22得票数 2
回答已采纳
5回答
我不想浪费你的时间，但是:你是否也想到过，在使用Python的with 语句时，它真的与“Python的禅宗”的第5行“平面比嵌套好”是相反的？有任何开明的Python大师能与我分享他们对此的一些见解吗？
(我总是发现，每当我使用with而不是f.close()时，代码中就会出现更多的缩进.而且我也不是不打算使用try: ... finally: ...，因此，with的好处仍然没有实现，即使我越来越喜欢和理解Python .)
@glglgl (对不起，我找不到在注释中编写代码的方法)：是的，但是如果您采用with方式，您的代码将变成：
try:
with file(...) as
浏览 2提问于2012-07-07得票数 10
回答已采纳
目前，我需要编写一个python脚本，它应该分析通过MQTT提供的一些数据。为此，我必须使用的方法需要一个生成器/迭代器作为参数。遗憾的是，python中的paho-mqtt lib似乎只能通过on_message回调方法来访问消息，而仅仅在该回调中添加一个“that”应该是行不通的。有没有办法以生成器的形式访问发布的消息，或者有没有可能把它们放到一个生成器中(可能通过多线程)，或者有没有其他的包可以用来做这件事？我对python不是很熟悉，也找不到解决方案。希望有人能有个主意。为尼克拉斯干杯
浏览 12提问于2020-02-10得票数 0
1回答
我想为异步函数创建一个上下文管理器，它在每次执行“移动”到另一个上下文时调用一个函数
例如。
import os
import asyncio
class AsyncContextChangeDir:
def __init__(self, newdir):
self.curdir = os.getcwd()
self.newdir = newdir
async def __aenter__(self):
os.chdir(self.newdir)
async def __aexit__(self, exc_type,
浏览 1提问于2017-04-05得票数 1
1回答
我想通过asyncio和websockets连接到websocket，格式如下所示。我怎样才能做到这一点呢？
from websockets import connect
class EchoWebsocket:
def __init__(self):
self.websocket = self._connect()
def _connect(self):
return connect("wss://echo.websocket.org")
def send(self, message):
sel
浏览 4提问于2016-05-22得票数 23
回答已采纳
我有一个带有with语句的python类A和带有with语句的类B。现在它的工作原理如下：
with A() as a:
with B() as b:
do_things()
我如何构建一个包装类A和类B的类C，这样我就可以像这样调用它：
with C() as c:
do_things()
具有相同的功能
浏览 0提问于2018-10-10得票数 1
我现在有这个核心，如果有人能帮上忙，那就太棒了
import aiohttp
from discord import Webhook, AsyncWebhookAdapter
a_file = open("path......")
lines_to_read = [0, 2] # line indexs starting with 0 so first line = 0 second = 1...
for position, line in enumerate(a_file):
if position in lines_to_read:
print(
浏览 4提问于2022-01-22得票数 0
Python指出，上下文管理器可以是单一用途、可重用或可重入的。可重入语句可以用于多个with语句，包括嵌套语句；可重用但不能重入的语句可以用于多个with语句，而不是嵌套语句。这里提到了几个例子。
不过，其他上下文管理器的文档并不总是提到它们是什么。例如，patch上下文管理器在unittest.mock中的文档根本没有提到这一点。
通常，您会在源代码中查看哪些内容来确定上下文管理器是单一用途、可重用还是可重入？
浏览 1提问于2014-07-30得票数 12
回答已采纳
我需要运行一个与其他代码并行的重代码，这样用户就不必等待。我正在尝试Task.Run(() =>来完成我的任务。我的任务是使用依赖注入从服务中调用方法。
我收到以下错误：
{“解决依赖关系失败，类型= \"Company.CacheProvider.CacheEmployeeProvider\"，名称=\”(无)\“.\r\nException发生时:进行解析。\r\nException是: InvalidOperationException - PerRequestLifetimeManager只能在InvalidOperationException请求的上下文中使用
浏览 0提问于2018-03-16得票数 1
回答已采纳
我编写了一个简单的TimeManager：一个上下文管理器，它在输入上下文时触发threading.Timer，在上下文退出时取消它。如果计时器在退出上下文之前停止，则会引发异常：
import threading
class TimeManager(object):
def __init__(self):
self._timeout = 1
def _timeoutHandler(self):
raise Exception("Timeout!")
def __enter__(self):
self
浏览 1提问于2015-03-25得票数 4
回答已采纳
1回答
我是新来的异步。最近我遇到了AIOFiles ()，并在文档中看到了它支持的“异步与”和“异步”。我想了解它，但没有多少好的报道，除了PEP 492，没有太多的细节。
PEP 492相关章节的快捷方式：
如果有人不介意回答的话，我有很多问题要问：
异步迭代器/上下文管理器的唯一明显好处是，您可以在它们所需的神奇方法的实现中使用awaitables。对于异步上下文管理器，PEP 492中的说：“异步上下文管理器是能够在其输入和退出方法中挂起执行的上下文管理器。”这是指使用等待调用协同线吗？
浏览 2提问于2019-11-06得票数 1
回答已采纳
1回答
如何在UWP中找到UI线程
、、、、
我试图了解线程在UWP中是如何工作的。
据我所知：
1)每个应用程序窗口都有一个UI线程。和许多工作者(或背景)线程。
2) UI元素只能通过UI线程访问。可以在后台线程中执行所有内存密集型的计算，只需将值分配给UI线程，以保持UI响应性。
为此，使用了CoreApplication.MainView.CoreWindow.Dispatcher.RunAsync(()=>方法)。
3)要检查线程是UI线程还是非UI线程，可以使用CoreApplication.MainView.CoreWindow.Dispatcher.HasThreadAccess。(如前面提到的 )。
但,
当我尝试
浏览 5提问于2017-02-24得票数 10
我正在学习Python，我发现有两种类型的文件打开操作。第一个是，
myreadfile = open("bear.txt", "r")
content = myreadfile.read()
第二种方法是
with open("bear.txt") as file:
content = file.read()
我想知道这两种方法是否有区别，哪一种最适合使用。
浏览 1提问于2021-07-16得票数 3
回答已采纳
1回答
我在本地使用dask客户端。问题是，当任务完成时，即使我使用本地调度程序，内存中仍然有很多python实例。这就是我要做的：
x.compute(get=dask.get)
from dask.distributed import Client
client = Client() # Starts local cluster
x.compute()
如果我运行这几次，我将在我的进程中(在windows上)得到几个python.exe。有什么办法防止这种情况发生吗？
谢谢
浏览 2提问于2017-09-15得票数 1
回答已采纳
在这里，我试图通过ansible剧本调用python脚本。当我运行包含下面提到的代码的剧本时，它将成功地调用脚本。但是ansible脚本继续运行，直到Python脚本完成它的所有任务。
- hosts: localhost
gather_facts: false
vars:
username: 'User'
password: '1234@345'
tasks:
- name: Invoking Python script
script: data_pull_push.py
args:
浏览 0提问于2020-04-16得票数 1
回答已采纳
根据的说法，在特定的情况下，异步/等待可能会出现死锁。
但是，如果我始终使用ConfigureAwait(false)，那么仍然会出现死锁吗？
例如：task.ConfigureAwait(false).GetAwaiter().GetResult()
而不是：task.GetAwaiter().GetResult()
浏览 3提问于2019-03-06得票数 3
回答已采纳
我的问题如下：
list = [1,2,3] #unknown length
with (item for item in list):
...
我知道python 3.10支持在上下文管理器中使用元组，但我无法使用该版本。
浏览 1提问于2022-07-15得票数 0
在中
class MyPaintWidget(Widget):
def on_touch_down(self, touch):
userdata = touch.ud
with self.canvas:
Color(1, 1, 0)
d = 30.
Ellipse(pos=(touch.x - d/2, touch.y - d/2), size=(d, d))
userdata['line'] = Line(points=(touch.x, t
浏览 0提问于2012-10-02得票数 0
回答已采纳
我有一个读入.csv文件的python脚本，这个文件在Python2.7上工作得很好，但在Python2.4上就不行了。抛出的错误在行上
with open(sys.argv[1], 'rb') as csvfile:
就在这里，它给了我一个语法错误，所以我的问题是'with‘做什么(或者在2.4中抛出语法错误的任何部分)。我在任何地方都找不到关于这个函数的文档，部分原因是它的通用名称。
浏览 6提问于2013-06-28得票数 2
回答已采纳
1回答
我经常看到自己试图遵循自己的代码，以找出异常的来源。
典型的例子是当某些解析失败时，我捕捉到了这个请求。然后，我花费了大量荒谬的时间使用断点和步骤/分步/步骤-来查看准确的解析失败。因为许多方法和许多不同的调用都使用这个特定的解析，所以我不确定是哪一个导致了它。
现在，如果它的生产代码是“坚实的”，每隔一段时间就会有一个bug出现，我真的很难找到它，只需要一个一个地复制和跳过每一行。不管它的价值是什么，即使它总是成功的，它就是如此乏味。
我做错什么了吗？也许我的建筑是错的？我真的没有任何代码可以展示，但如果其他人有同样的感觉或知道这种情况发生的经验，我会很高兴了解一些东西！
浏览 0提问于2016-02-11得票数 1
回答已采纳
我试图教会自己Python的异步功能。为此，我构建了一个异步web刮刀。我想限制我一次打开的连接的总数，以便成为服务器上的好公民。我知道信号量是一个很好的解决方案，异步库有一个内置的类。我的问题是，在将yield from和await语法组合在一起时，Python在async函数中使用yield时会发出抱怨。下面是我使用的确切语法..。
import asyncio
import aiohttp
sema = asyncio.BoundedSemaphore(5)
async def get_page_text(url):
with (yield from sema):
浏览 3提问于2016-11-28得票数 20
我试图使用responses模块来模拟HTTP服务器，当调用请求是多线程的时候，我觉得它被破坏了。
请考虑以下几点：
import json
import responses
import requests
import threading
try:
import urlparse
except ModuleNotFoundError:
import urllib.parse as urlparse
server_url = 'http://server'
headers_json = {'content-type': 'applica
浏览 7提问于2019-10-14得票数 1
回答已采纳
给出以下最起码的例子：
@asynccontextmanager
async def async_context():
try:
yield
finally:
await asyncio.sleep(1)
print('finalize context')
async def async_gen():
try:
yield
finally:
await asyncio.sleep(2)
# will never be called if timeo
浏览 3提问于2021-02-24得票数 1
回答已采纳
我正在尝试从一个api获得一些响应，而我正在使用一个包装器。
我寻找更多的错误，但他们没有解决我的问题。我对等待一个函数或协作不太了解。我关掉了is_asnyc，它成功了，但我需要那个选项。所以我不能关掉它。
import clashroyale, asyncio
token = "my token"
cr = clashroyale.official_api.Client(token=token, is_async=True)
async def top():
p = await cr.get_top_players()
return p
topplaye
浏览 2提问于2019-08-12得票数 3
回答已采纳
1回答
在中，我没有找到任何关于上下文服务器update的信息。
我需要原子上的以下内容：
context_metadata = contextvars.ContextVar("context_logger_metadata")
my_dict = context_metadata.get()
my_dict['appended'] = 'some_data'
context_metadata.set(my_dict)
python提供了类似于同步块的东西吗？如何确保GIL不会在get和set之间切换上下文？
浏览 3提问于2021-09-17得票数 0
回答已采纳
我是一个python新手，但我正在用python实现一个基准测试工具，例如，它将创建几组相互依赖的资源。当程序超出范围时，我希望以正确的顺序清理资源。 我有C++背景，在C++中我知道我可以用RAII (构造函数，析构函数)做到这一点。 此问题在模式中的等效模式是什么？有没有办法用python做RAII，或者有更好的办法来解决这个问题？
浏览 44提问于2020-04-17得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
