首頁  >  文章  >  後端開發  >  Python上下文管理器怎麼使用

Python上下文管理器怎麼使用

WBOY
WBOY轉載
2023-05-21 09:16:051412瀏覽

什麼是上下文管理器?

即使你沒有聽說過 Python 的上下文管理器,根據介紹,你也已經知道,它是try/finally區塊的替代品。它是使用開啟檔案時常用的語句with來實現的。與try/finally相同,引入此模式是為了確保在區塊末尾執行某些操作,即使發生異常或程式終止。

從表面上看,上下文管理協定只是圍繞著with程式碼區塊的語句。實際上,它包含 2 個特殊的 ( dunder ) 方法 -__enter____exit__組成,分別有助於啟動和停止。

當程式碼中遇到with語句時,將觸發__enter__方法並將其傳回值放入as限定符後面的變數中。 with區塊體執行完畢後,呼叫__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/finally的版本和使用with語句來實現簡單的計時器的更優雅的版本。如上所述,實作這樣的上下文管理器需要__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__方法只啟動計時器並傳回selfself將在with ....中作為some_var賦值,with語句體完成後,將使用3 個參數呼叫__exit__方法- 異常類型、異常值和回溯。如果with語句正文中一切順利,則這些都等於None。如果引發異常,這些將填入異常數據,我們可以在__exit__方法中處理這些數據。在這種情況下,我們省略了異常處理,只是停止計時器併計算經過的時間,並將其儲存在上下文管理器的屬性中。

我們已經在這裡看到了with語句的實作和範例用法,但是為了更直觀地了解實際發生的情況,讓我們看看如何在沒有Python 語法糖的情況下調用這些特殊方法:

manager = Timer()
manager.__enter__()  # Setup
time.sleep(3)  # Body
manager.__exit__(None, None, None)  # Teardown
print(manager.elapsed)

現在我們已經確定了什麼是上下文管理器,它是如何工作的以及如何實現它,讓我們看看使用它的好處——只是為了有更多的動力從try/finally切換到with語句。

第一個好處是整個啟動和停止都在上下文管理器物件的控制下進行。這可以防止錯誤並減少樣板程式碼,從而使 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)

此程式碼段實作了與上一節中的Timer類別非常相似的上下文管理器。然而,這一次,我們需要的程式碼要少得多。這段程式碼分成兩個部分,一部分是在yield之前,另一部分是yield之後。 yield之前的程式碼承擔了__enter__方法的工作,而yield本身就是__enter__方法的return語句。 yield之後的都是__exit__方法的一部份。

正如你在上面看到的,像这样使用单个函数创建上下文管理器需要使用使用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

上面的代码演示了不带和带上下文管理器的选项。第二个选项显然更短,更具可读性。它还考虑了临时上下文,使其不易出错。

contextlib

在使用@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)

如果你有一个工具或函数,默认情况下将所有数据输出到stdoutstderr,但你希望它将数据输出到其他地方——例如文件。那么这两个上下文管理器可能非常有用。与前面的示例一样,这大大提高了代码的可读性,并消除了不必要的视觉干扰。

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 事务

最后但同样重要的是,还有用于管理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上下文管理器的良好使用,它有助于处理不再使用的连接对象,这进一步简化了代码,并确保我们不会让任何连接挂起。

以上是Python上下文管理器怎麼使用的詳細內容。更多資訊請關注PHP中文網其他相關文章!

陳述:
本文轉載於:yisu.com。如有侵權,請聯絡admin@php.cn刪除