Python上下文管理器怎么使用
这篇文章主要介绍了Python上下文管理器怎么使用的相关知识，内容详细易懂，操作简单快捷，具有一定借鉴价值，相信大家阅读完这篇Python上下文管理器怎么使用文章都会有所收获，下面我们一起来看看吧。
什么是上下文管理器？
即使你没有听说过 Python 的上下文管理器，根据介绍，你也已经知道，它是
try/finallywithtry/finally从表面上看，上下文管理协议只是围绕
with__enter____exit__当代码中遇到
with__enter__aswith__exit__finally# Using try/finally
import time
start = time.perf_counter() # Setup
try: # Actual body
time.sleep(3)
finally: # Teardown
end = time.perf_counter()
elapsed = end - start
print(elapsed)
# Using Context Manager
with Timer() as t:
time.sleep(3)
print(t.elapsed)上面的代码显示了使用
try/finallywith__enter____exit__Timer# Implementation of above context manager
class Timer:
def __init__(self):
self._start = None
self.elapsed = 0.0
def start(self):
if self._start is not None:
raise RuntimeError('Timer already started...')
self._start = time.perf_counter()
def stop(self):
if self._start is None:
raise RuntimeError('Timer not yet started...')
end = time.perf_counter()
self.elapsed += end - self._start
self._start = None
def __enter__(self): # Setup
self.start()
return self
def __exit__(self, *args): # Teardown
self.stop()此代码片段显示了实现
__enter____exit__Timer__enter__selfselfwith ...some_varwith__exit__withNone__exit__我们已经在这里看到了
withmanager = Timer()
manager.__enter__() # Setup
time.sleep(3) # Body
manager.__exit__(None, None, None) # Teardown
print(manager.elapsed)现在我们已经确定了什么是上下文管理器，它是如何工作的以及如何实现它，让我们看看使用它的好处——只是为了有更多的动力从
try/finallywith第一个好处是整个启动和停止都在上下文管理器对象的控制下进行。这可以防止错误并减少样板代码，从而使 API 更安全、更易于使用。使用它的另一个原因是
with__enter____exit__话虽如此，我希望我能说服你开始使用上下文管理器，而不是
try/finally@contextmanager
在上一节中，我们探讨了如何使用
__enter____exit__contextlib@contextmanager@contextmanager__enter____exit__from contextlib import contextmanager
from time import time, sleep
@contextmanager
def timed(label):
start = time() # Setup - __enter__
print(f"{label}: Start at {start}")
try:
yield # yield to body of `with` statement
finally: # Teardown - __exit__
end = time()
print(f"{label}: End at {end} ({end - start} elapsed)")
with timed("Counter"):
sleep(3)
# Counter: Start at 1599153092.4826472
# Counter: End at 1599153095.4854734 (3.00282621383667 elapsed)此代码段实现了与上一节中的
Timeryieldyieldyield__enter__yield__enter__returnyield__exit__正如你在上面看到的，像这样使用单个函数创建上下文管理器需要使用使用
try/finallywithyyield__exit__finally正如我已经提到的，这可以用于自包含的上下文管理器。但是，它不适合需要成为对象一部分的上下文管理器，例如连接或锁。
尽管使用单个函数构建上下文管理器会迫使你使用
try/finally现实生活中的例子
现在让我们从理论转向实用且有用的上下文管理器，你可以自己构建它。
记录上下文管理器
当需要尝试查找代码中的一些bug时，你可能会首先查看日志以找到问题的根本原因。但是，这些日志可能默认设置为错误或警告级别，这可能不足以用于调试。更改整个程序的日志级别应该很容易，但更改特定代码部分的日志级别可能会更复杂 - 不过，这可以通过以下上下文管理器轻松解决：
import logging
from contextlib import contextmanager
@contextmanager
def log(level):
logger = logging.getLogger()
current_level = logger.getEffectiveLevel()
logger.setLevel(level)
try:
yield
finally:
logger.setLevel(current_level)
def some_function():
logging.debug("Some debug level information...")
logging.error('Serious error...')
logging.warning('Some warning message...')
with log(logging.DEBUG):
some_function()
# DEBUG:root:Some debug level information...
# ERROR:root:Serious error...
# WARNING:root:Some warning message...超时上下文管理器
在本文的开头，我们正在使用计时代码块。我们在这里尝试的是将超时设置为
withimport signal
from time import sleep
class timeout:
def __init__(self, seconds, *, timeout_message=""):
self.seconds = int(seconds)
self.timeout_message = timeout_message
def _timeout_handler(self, signum, frame):
raise TimeoutError(self.timeout_message)
def __enter__(self):
signal.signal(signal.SIGALRM, self._timeout_handler) # Set handler for SIGALRM
signal.alarm(self.seconds) # start countdown for SIGALRM to be raised
def __exit__(self, exc_type, exc_val, exc_tb):
signal.alarm(0) # Cancel SIGALRM if it's scheduled
return exc_type is TimeoutError # Suppress TimeoutError
with timeout(3):
# Some long running task...
sleep(10)上面的代码为这个上下文管理器声明了一个名为
timeoutSIGALRMsignal.signal(...)SIGALRMSIGALRM_timeout_handlerTimeoutErrorwithsignal.alarm(self.seconds)对于
__exit__SIGALRMsignal.alarm(0)_timeout_handlerTimeoutError__exit__with使用已有的
除了上面的上下文管理器，标准库或其他常用库（如request或sqlite3）中已经有很多有用的上下文管理程序。那么，让我们看看我们可以在那里找到什么。
临时更改小数精度
如果你正在执行大量数学运算并需要特定的精度，那么你可能会遇到需要临时更改十进制数精度的情况：
from decimal import getcontext, Decimal, setcontext, localcontext, Context
# Bad
old_context = getcontext().copy()
getcontext().prec = 40
print(Decimal(22) / Decimal(7))
setcontext(old_context)
# Good
with localcontext(Context(prec=50)):
print(Decimal(22) / Decimal(7)) # 3.1428571428571428571428571428571428571428571428571
print(Decimal(22) / Decimal(7)) # 3.142857142857142857142857143上面的代码演示了不带和带上下文管理器的选项。第二个选项显然更短，更具可读性。它还考虑了临时上下文，使其不易出错。
contextlib
在使用
@contextmanagercontextlibredirect_stdoutredirect_stderrimport sys
from contextlib import redirect_stdout
# Bad
with open("help.txt", "w") as file:
stdout = sys.stdout
sys.stdout = file
try:
help(int)
finally:
sys.stdout = stdout
# Good
with open("help.txt", "w") as file:
with redirect_stdout(file):
help(int)如果你有一个工具或函数，默认情况下将所有数据输出到
stdoutstderrcontextlibsuppressimport os
from contextlib import suppress
try:
os.remove('file.txt')
except FileNotFoundError:
pass
with suppress(FileNotFoundError):
os.remove('file.txt')当然，正确处理异常是更好的，但有时你只需要消除令人讨厌的
DeprecationWarning我将提到的
contextlibclosing# Bad
try:
page = urlopen(url)
...
finally:
page.close()
# Good
from contextlib import closing
with closing(urlopen(url)) as page:
...此上下文管理器将关闭作为参数传递给它的任何资源（在上面的示例中），即
page.close()try/finally用于更好测试的上下文管理器
若你们想让人们使用、阅读或维护你们所写的测试，你们必须让他们可读，易于理解和模仿。
mock.patch# Bad
import requests
from unittest import mock
from unittest.mock import Mock
r = Mock()
p = mock.patch('requests.get', return_value=r)
mock_func = p.start()
requests.get(...)
# ... do some asserts
p.stop()
# Good
r = Mock()
with mock.patch('requests.get', return_value=r):
requests.get(...)
# ... do some asserts使用
mock.patch.start().stop()unittestpytestunittest说到
pytestimport pytest, os
with pytest.raises(FileNotFoundError, message="Expecting FileNotFoundError"):
os.remove('file.txt')这个例子展示了
pytest.raises跨请求持久化会话
从
pytestrequestsrequestsimport requests
with requests.Session() as session:
session.request(method=method, url=url, **kwargs)除了解决上述问题之外，这个上下文管理器还可以帮助提高性能，因为它将重用底层连接，因此避免为每个请求/响应对打开新连接。
管理 SQLite 事务
最后但同样重要的是，还有用于管理SQLite事务的上下文管理器。除了使代码更干净之外，此上下文管理器还提供了在异常情况下回滚更改的能力，以及在
withimport sqlite3
from contextlib import closing
# Bad
connection = sqlite3.connect(":memory:")
try:
connection.execute("INSERT INTO employee(firstname, lastname) values (?, ?)", ("John", "Smith",))
except sqlite3.IntegrityError:
...
connection.close()
# Good
with closing(sqlite3.connect(":memory:")) as connection:
with connection:
connection.execute("INSERT INTO employee(firstname, lastname) values (?, ?)", ("John", "Smith",))在本例中，你还可以看到
closing相关文章