在 Python 處理文件的時候我們使用 with 關鍵詞來進行文件的資源關閉,但是並不是只有文件操作才能使用 with 語句。今天就讓我們一起學習 Python 中的上下文管理 contextlib。
上下文管理器上下文，簡而言之，就是程式所執行的環境狀態，或者說程式運行的情景。既然提及上下文，就不可避免的涉及 Python 中關於上下文的魔法。上下文管理器(context manager)是 Python2.5 開始支持的一種語法，用於規定某個對象的使用範圍。一旦進入或者離開該使用範圍，會有特殊操作被調用。它的語法形式是 with…as…，主要應用場景資源的創建和釋放。例如，文件就支持上下文管理器，可以確保完成文件讀寫後關閉文件句柄。
with open('password.txt', 'wt') as f:
f.write('contents go here')
with 方法的實現涉及到兩個魔法函數__enter__和__exit__。
執行流進入 with 中的代碼塊時會執行__enter__方法，它會返回在這個上下文中使用的一個對象。執行流離開 with 塊時，則調用這個上下文管理器的__exit__方法來清理所使用的資源。
class Context:
def __init__(self):
print("int __init__")
def __enter__(self):
print("int __enter__")
def __exit__(self, exc_type, exc_val, exc_tb):
print("in __exit__")
if __name__ == '__main__':
with Context():
print('start with ')
輸出
int __init__
int __enter__
start witht
in __exit__
相對於使用 try:finally 塊，使用 with 語句代碼看起來更緊湊，with 代碼塊執行的時候總會調用__exit__方法,即使出現了異常。
如果給 with 語句的 as 子句指定一個別名，那麼__enter__方法可以返回與這個名關聯的任何對象。
import requests
class Request:
def __init__(self):
self.session = requests.session()
def get(self, url, headers=None):
if headers is None:
headers = {}
response = self.session.get(url)
return response
class Context:
def __init__(self):
print("int __init__")
def __enter__(self):
print("int __enter__")
return Request()
def __exit__(self, exc_type, exc_val, exc_tb):
print("in __exit__")
if __name__ == '__main__':
with Context() as t:
req = t.get("https://wwww.baidu.com")
print(req.text)
這裡的 t 就是__enter__方法返回的 Request 對象的實例,然後我們可以調用該實例的一些方法。
如果上下文中出現異常，可以通過修改__exit__方法，如果返回值為 true，則可以把異常列印出來，如果為 false 則會拋出異常。
class Context:
def __init__(self,flag):
self.flag = flag
def __enter__(self):
print("int __enter__")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("in __exit__")
print(f"{exc_type=}")
print(f"{exc_val=}")
print(f"{exc_val=}")
return self.flag
if __name__ == '__main__':
with Context(True) as t:
raise RuntimeError()
print("-華麗的分割線----")
with Context(False) as t:
raise RuntimeError()
輸出
int __enter__
in __exit__
exc_type=<class 'RuntimeError'>
exc_val=RuntimeError()
exc_val=RuntimeError()
-華麗的分割線----
int __enter__
in __exit__
exc_type=<class 'RuntimeError'>
exc_val=RuntimeError()
exc_val=RuntimeError()
Traceback (most recent call last):
File "demo.py", line 26, in <module>
raise RuntimeError()
RuntimeError
可以看到__exit__方法接收一些參數，其中包含 with 塊中產生的異常的詳細信息。
上下文管理器作為裝飾器通過繼承 contextlib 裡面的 ContextDecorator 類,實現對常規上下文管理器類的支持，其不僅可以作為上下文管理器，也可以作為函數修飾符。
import contextlib
class Context(contextlib.ContextDecorator):
def __init__(self, how_used):
self.how_used = how_used
print(f'__init__({how_used})')
def __enter__(self):
print(f'__enter__({self.how_used})')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print(f'__exit__({self.how_used})')
@Context('這是裝飾器方式')
def func(message):
print(message)
print("")
func('作為裝飾器運行')
print("\n--華麗的分割線---\n")
with Context('上下文管理器方式'):
print('emmmm')
看一下 ContextDecorator 的源碼就了解了
class ContextDecorator(object):
"A base class or mixin that enables context managers to work as decorators."
def _recreate_cm(self):
"""Return a recreated instance of self.
Allows an otherwise one-shot context manager like
_GeneratorContextManager to support use as
a decorator via implicit recreation.
This is a private interface just for _GeneratorContextManager.
See issue #11647 for details.
"""
return self
def __call__(self, func):
@wraps(func)
def inner(*args, **kwds):
with self._recreate_cm():
return func(*args, **kwds)
return inner
有時候我們的代碼只有很少的上下文要管理，此時再使用上面的形式寫出 with 相關的魔法函數
就顯得比較囉嗦了，在這種情況下，我們可以使用 contextmanager 修飾符將一個生成器轉換為上下文管理器。
import contextlib
@contextlib.contextmanager
def make_context():
print("enter make_context")
try:
yield {}
except RuntimeError as err:
print(f"{err=}")
print("Normal")
with make_context() as value:
print("in with")
print("RuntimeError")
with make_context() as value:
raise RuntimeError("runtimeerror")
print("Else Error")
with make_context() as value:
raise ZeroDivisionError("0 不能作為分母")
輸出結果
enter make_context
in with
existing
RuntimeError:
enter make_context
err=RuntimeError(
existing
Else Error:
enter make_context
existing
Traceback (most recent call last):
File "demo.py", line 24, in <module>
raise ZeroDivisionError("0 不能作為除數")
ZeroDivisionError: 0 不能作為除數
需要主要的是這個函數必須是個裝飾器
被裝飾器裝飾的函數分為三部分:
with 語句中的代碼塊執行前執行函數中 yield 之前代碼
yield 返回的內容複製給 as 之後的變量
with 代碼塊執行完畢後執行函數中 yield 之後的代碼
再簡單理解就是
yield 前半段用來表示__enter__()
yield 後半段用來表示__exit__()
在這裡，我們從 contextlib 模塊中引入 contextmanager，然後裝飾我們所定義的 make_context 函數。這就允許我們使用 Python 的 with 語句來調用 make_context 函數。在函數中通過 yield 一個空字典，將其傳遞出去，最終主調函數可以使用它。
一旦 with 語句結束，控制就會返回給 make_context 函數，它繼續執行 yield 語句後面的代碼,這個最終會執行 finally 語句列印 existing。如果我們遇到了 RuntimeError 錯誤，它就會被捕獲，最終 finally 語句依然會列印 existing。
因為 contextmanager 繼承自 ContextDecorator，所以也可以被用作函數修飾符
import contextlib
@contextlib.contextmanager
def make_context():
print("enter make_context")
try:
yield {}
except RuntimeError as err:
print(f"{err=}")
finally:
print("existing")
@make_context()
def normal():
print("in normal")
@make_context()
def raise_error(err):
raise err
if __name__ == '__main__':
print("Normal:")
normal()
print("RuntimeError:")
raise_error(RuntimeError("runtime error"))
print("Else Error")
raise_error(ValueError("value error"))
輸出
Normal:
enter make_context
in normal
existing
RuntimeError:
enter make_context
err=RuntimeError('runtime error')
existing
Else Error
enter make_context
existing
Traceback (most recent call last):
File "demo.py", line 33, in <module>
raise_error(ValueError("value error"))
File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/contextlib.py", line 75, in inner
return func(*args, **kwds)
File "demo.py", line 21, in raise_error
raise err
ValueError: value error
這是 contextmanager 原理：
1、因為 func()已經是個生成器了嘛，所以運行enter_()的時候，contextmanager 調用 self.gen.next()會跑到 func 的 yield 處，停住掛起，這個時候已經有了 t1=time.time()2、然後運行 with 語句體裡面的語句，也就是 a+b=3003、跑完後運行exit()的時候，contextmanager 調用 self.gen.next()會從 func 的 yield 的下一句開始一直到結束。這個時候有了 t2=time.time()，t2-t1 從而實現了統計 cost_time 的效果，完美。
源碼
class GeneratorContextManager(object):
"""Helper for @contextmanager decorator."""
def __init__(self, gen):
self.gen = gen
def __enter__(self):
try:
return self.gen.next()
except StopIteration:
raise RuntimeError("generator didn't yield")
def __exit__(self, type, value, traceback):
if type is None:
try:
self.gen.next()
except StopIteration:
return
else:
raise RuntimeError("generator didn't stop")
else:
if value is None:
value = type()
try:
self.gen.throw(type, value, traceback)
raise RuntimeError("generator didn't stop after throw()")
except StopIteration, exc:
return exc is not value
except:
if sys.exc_info()[1] is not value:
raise
def contextmanager(func):
@wraps(func)
def helper(*args, **kwds):
return GeneratorContextManager(func(*args, **kwds))
return helper
並不是所有的類都支持上下文管理器的 API，有一些遺留的類會使用一個 close 方法。
為了確保關閉句柄，需要使用 closing 為他創建一個上文管理器。
該模塊主要是解決不能支持上下文管理器的類。
import contextlib
class Http():
def __init__(self):
print("int init:")
self.session = "open"
def close(self):
"""
關閉的方法必須叫 close
"""
print("in close:")
self.session = "close"
if __name__ == '__main__':
with contextlib.closing(Http()) as http:
print(f"inside session value:{http.session}")
print(f"outside session value:{http.session}")
with contextlib.closing(Http()) as http:
print(f"inside session value:{http.session}")
raise EnvironmentError("EnvironmentError")
print(f"outside session value:{http.session}")
輸出
int init:
inside session value:open
in close:
outside session value:close
int init:
inside session value:open
in close:
Traceback (most recent call last):
File "demo.py", line 23, in <module>
raise EnvironmentError("EnvironmentError")
OSError: EnvironmentError
可以看到即使程序出現了錯誤，最後也會執行 close 方法的內容。
看一眼下面的源碼就知道 closing 幹嘛了
class closing(object):
"""Context to automatically close something at the end of a block.
Code like this:
with closing(<module>.open(<arguments>)) as f:
<block>
is equivalent to this:
f = <module>.open(<arguments>)
try:
<block>
finally:
f.close()
"""
def __init__(self, thing):
self.thing = thing
def __enter__(self):
return self.thing
def __exit__(self, *exc_info):
self.thing.close()
這個 contextlib.closing()會幫它加上__enter__()和__exit__()，使其滿足 with 的條件。然後 exit 裡執行的就是對應類的 close 方法。
巧妙的迴避錯誤contextlib.suppress(*exceptions)
另一個工具就是在 Python 3.4 中加入的 suppress 類。這個上下文管理工具背後的理念就是它可以禁止任意數目的異常。假如我們想忽略 FileNotFoundError 異常。如果你書寫了如下的上下文管理器，那麼它不會正常運行。
>>> with open("1.txt") as fobj:
... for line in fobj:
... print(line)
...
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
FileNotFoundError: [Errno 2] No such file or directory: '1.txt'
正如你所看到的，這個上下文管理器沒有處理這個異常，如果你想忽略這個錯誤，你可以按照如下方式來做，
>>> from contextlib import suppress
>>> with suppress(FileNotFoundError):
... with open("1.txt") as fobj:
... for line in fobj:
... print(line)
在這段代碼中，我們引入 suppress，然後將我們要忽略的異常傳遞給它，在這個例子中，就是 FileNotFoundError。如果你想運行這段代碼，你將會注意到，文件不存在時，什麼事情都沒有發生，也沒有錯誤被拋出。請注意，這個上下文管理器是可重用的。
例子資料庫的自動提交和回滾在編程中如果頻繁的修改資料庫, 一味的使用類似 try:… except..: rollback() raise e 其實是不太好的.
try:
gift = Gift()
gift.isbn = isbn
...
db.session.add(gift)
db.session.commit()
except Exception as e:
db.session.rollback()
raise e
為了達到使用 with 語句的目的, 我們可以重寫 db 所屬的類:
from flask_sqlalchemy import SQLAlchemy as _SQLALchemy
class SQLAlchemy(_SQLALchemy):
@contextmanager
def auto_commit(self):
try:
yield
self.session.commit()
except Exception as e:
db.session.rollback()
raise e
這時候, 在執行數據的修改的時候便可以:
with db.auto_commit():
gift = Gift()
gift.isbn = isbndb.session.add(gift)
db.session.add(gift)
with db.auto_commit():
user = User()
user.set_attrs(form.data)
db.session.add(user)
上下文管理器很有趣，也很方便。我經常在自動測試中使用它們，例如，打開和關閉對話。現在，你應該可以使用 Python 內置的工具去創建你的上下文管理器。你還可以繼續閱讀 Python 關於 contextlib 的文檔，那裡有很多本文沒有覆蓋到的知識。
這篇文章其實主要為了後面的異步上文管理器使用的。
參考資料https://www.cnblogs.com/zhbzz2007/p/6158125.html
https://blog.csdn.net/emaste_r/article/details/78105713
https://blog.csdn.net/weixin_42359464/article/details/80742387#three
https://docs.python.org/zh-cn/3/library/contextlib.html?highlight=contextlib#module-contextlib
點擊閱讀原文獲取Python 爬蟲面試題 170 道：2019 版