Even if you haven't heard of Python's Context manager, according to the introduction, you already know that it is a replacement for the try/finally
block. It is implemented using the commonly used statement with
when opening a file. Same as try/finally
, this pattern was introduced to guarantee that certain operations are performed at the end of the block, even if an exception occurs or the program terminates.
On the surface, the Context Management Protocol is just a statement surrounding a with
block of code. In fact, it consists of 2 special ( dunder ) methods - __enter__
and __exit__
, which help to start and stop respectively.
When a with
statement is encountered in the code, the __enter__
method will be triggered and its return value will be put into the variable after the as
qualifier middle. with
After the block body is executed, call the __exit__
method to stop - completing the role of the finally
block.
# 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)
The code above shows a version using try/finally
and a more elegant version using the with
statement to implement a simple timer. As mentioned above, implementing such a context manager requires __enter__
and __exit__
, but how would we create them? Let’s take a look at the code for this Timer
class:
# 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()
This code snippet shows the Timer# that implements the
__enter__ and
__exit__ methods ##kind. The
__enter__ method simply starts the timer and returns
self,
self will be used as
some_var## in with ...
. #Assignment, with
After the statement body is completed, the __exit__
method will be called with 3 parameters - exception type, exception value and traceback. If all goes well in the body of the with
statement, these are equal to None
. If an exception is thrown, these will be populated with exception data, which we can handle in the __exit__
method. In this case we omit exception handling and just stop the timer and calculate the elapsed time and store it in a property of the context manager. We've seen the implementation and example usage of the
statement here, but to get a more intuitive understanding of what actually
happens, let's see how to do it without Python syntax sugar for calling these special methods: manager = Timer()
manager.__enter__() # Setup
time.sleep(3) # Body
manager.__exit__(None, None, None) # Teardown
print(manager.elapsed)
to with
statements. The first benefit is that the entire starting and stopping are under the control of the context manager object. This prevents errors and reduces boilerplate code, making the API safer and easier to use. Another reason to use it is that
blocks highlight key sections and encourage you to reduce the amount of code in that section, which is generally a good practice as well. Finally - last but not least - it's a great refactoring tool that breaks out common start and stop code and moves it to one location - i.e. __enter__
and __exit__
methods. With that said, I hope I can convince you to start using context managers instead of
, even if you haven't used them before. So, now let’s look at some cool and useful context managers that you should start including in your code! @contextmanager
and __exit__
methods. This is simple, but we can make it even simpler using contextlib
and more specifically @contextmanager
.
is a decorator that can be used to write self-contained context management functions. Therefore, we don't need to create the entire class and implement the __enter__
and __exit__
methods, we just need to create a generator: <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">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)</pre><div class="contentsignin">Copy after login</div></div>
This code snippet implements the same as the previous one The
class in the section is very similar to the context manager. This time, however, we need much less code. This code is divided into two parts, one part is before yield
, and the other part is after yield
. yield
The previous code assumes the work of the __enter__
method, and yield
itself is the return
statement of the __enter__
method . Everything after yield
is part of the __exit__
method. <p data-id="p838747a-t4jwrOVA">正如你在上面看到的,像这样使用单个函数创建上下文管理器需要使用使用<code>try/finally
语句,因为如果在语句withy
体中发生异常,它将在yield
行被引发,我们需要在对应于__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...
在本文的开头,我们正在使用计时代码块。我们在这里尝试的是将超时设置为with
语句包围的块:
import 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)
上面的代码为这个上下文管理器声明了一个名为timeout
的类,因为这个任务不能在单个函数中完成。为了能够实现这种超时,我们还需要使用信号-更具体地说是SIGALRM
。我们首先使用signal.signal(...)
将处理程序设置为SIGALRM
,这意味着当内核引发SIGALRM
时,将调用处理程序函数。对于这个处理程序函数(_timeout_handler
),它所做的只是引发TimeoutError
,如果没有及时完成,它将停止with
语句体中的执行。处理程序就位后,我们还需要以指定的秒数开始倒计时,这由signal.alarm(self.seconds)
完成。
对于__exit__
方法,如果上下文管理器的主体设法在时间到期之前完成,SIGALRM
则将被取消,而signal.alarm(0)
和程序可以继续。另一方面 - 如果由于超时而引发信号,那么_timeout_handler
将引发TimeoutError
,这将__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
上面的代码演示了不带和带上下文管理器的选项。第二个选项显然更短,更具可读性。它还考虑了临时上下文,使其不易出错。
在使用@contextmanager
时,我们已经窥探了contextlib
,但我们可以使用更多的东西——作为第一个示例,让我们看看redirect_stdout
和redirect redirect_stderr
:
import 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)
如果你有一个工具或函数,默认情况下将所有数据输出到stdout
或stderr
,但你希望它将数据输出到其他地方——例如文件。那么这两个上下文管理器可能非常有用。与前面的示例一样,这大大提高了代码的可读性,并消除了不必要的视觉干扰。
contextlib
的另一个方便的方法是suppress
上下文管理器,它将抑制任何不需要的异常和错误:
import os from contextlib import suppress try: os.remove('file.txt') except FileNotFoundError: pass with suppress(FileNotFoundError): os.remove('file.txt')
当然,正确处理异常是更好的,但有时你只需要消除令人讨厌的DeprecationWarning
警告,这个上下文管理器至少会使它可读。
我将提到的contextlib
中的最后一个实际上是我最喜欢的,它叫做closing
:
# 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()
调用,并帮助你定义此特定模拟的明确范围。这个测试的好处是它可以与unittest
以及pytest
一起使用,即使它是标准库的一部分(因此也是unittest
)。
说到pytest
,让我们也展示一下这个库中至少一个非常有用的上下文管理器:
import pytest, os with pytest.raises(FileNotFoundError, message="Expecting FileNotFoundError"): os.remove('file.txt')
这个例子展示了pytest.raises
的非常简单的用法,它断言代码块引发提供的异常。如果没有,则测试失败。这对于测试预期会引发异常或失败的代码路径非常方便。
从pytest
转到另一个伟大的库——requests
。通常,你可能需要在HTTP请求之间保留cookie,需要保持TCP连接活动,或者只想对同一主机执行多个请求。requests
提供了一个很好的上下文管理器来帮助应对这些挑战,即管理会话:
import requests with requests.Session() as session: session.request(method=method, url=url, **kwargs)
除了解决上述问题之外,这个上下文管理器还可以帮助提高性能,因为它将重用底层连接,因此避免为每个请求/响应对打开新连接。
最后但同样重要的是,还有用于管理SQLite事务的上下文管理器。除了使代码更干净之外,此上下文管理器还提供了在异常情况下回滚更改的能力,以及在with
语句体成功完成时自动提交的能力:
import 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
上下文管理器的良好使用,它有助于处理不再使用的连接对象,这进一步简化了代码,并确保我们不会让任何连接挂起。
The above is the detailed content of How to use Python context manager. For more information, please follow other related articles on the PHP Chinese website!