Python 上下文管理器模塊--contextlib

時間 2021-02-03

標籤 html python sql 數據庫編程 flask session 異步 ide 函數欄目 Python 简体版

原文原文鏈接

在 Python 處理文件的時候咱們使用 with 關鍵詞來進行文件的資源關閉,可是並非只有文件操做才能使用 with 語句。今天就讓咱們一塊兒學習 Python 中的上下文管理 contextlib。html

上下文管理器

上下文，簡而言之，就是程式所執行的環境狀態，或者說程式運行的情景。既然說起上下文，就不可避免的涉及 Python 中關於上下文的魔法。上下文管理器(context manager)是 Python2.5 開始支持的一種語法，用於規定某個對象的使用範圍。一旦進入或者離開該使用範圍，會有特殊操做被調用。它的語法形式是 with...as...，主要應用場景資源的建立和釋放。例如，文件就支持上下文管理器，能夠確保完成文件讀寫後關閉文件句柄。python

with open('password.txt', 'wt') as f:
    f.write('contents go here')
# 文件會自動關閉

_enter__和_exit

with 方法的實現涉及到兩個魔法函數_enter__和_exit。
執行流進入 with 中的代碼塊時會執行__enter__方法，它會返回在這個上下文中使用的一個對象。執行流離開 with 塊時，則調用這個上下文管理器的__exit__方法來清理所使用的資源。sql

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__
Doing work in context
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 則會拋出異常。flask

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=}") #Python3.8 的語法等價於 f"exc_type={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()

輸出session

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 類,實現對常規上下文管理器類的支持，其不只能夠做爲上下文管理器，也能夠做爲函數修飾符。ide

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('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 原理：
一、由於 func()已是個生成器了嘛，因此運行__enter__()的時候，contextmanager 調用 self.gen.next()會跑到 func 的 yield 處，停住掛起，這個時候已經有了 t1=time.time()
二、而後運行 with 語句體裏面的語句，也就是 a+b=300
三、跑完後運行__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:
                # Need to force instantiation so we can reliably
                # tell if we get the same exception back
                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 的文檔，那裏有不少本文沒有覆蓋到的知識。

這篇文章其實主要爲了後面的異步上文管理器使用的。
更多內容請關注公衆號:python學習開發