Python context manager — więcej niż opakowanie na try/except/finally. Szczegółowe wytłumaczeni i 9 unikalnych przykładów

Wyrażenie with nie wnosi nic wyjątkowego w język Python. Można się bez niego zwyczajnie obejść, a to dlatego, że każde wyrażenie with można zawsze zapisać za pomocą try/except/finally. Dlaczego jednak znalazło się w standardowej bibliotece? 

Po pierwsze aby hermetyzować zachowania które muszą się wydarzyć przed i po danej funkcjonalności. Co więcej, logowanie i obsługa wyjątków często zaciemniają kod, a te mogą być zapisane w formie kontekst menadżera jednocześnie zwiększając czytelność — innymi słowy w tym wypadku kontekst menedżer redukuje wizualny szum. Oczywiście wiele zależy od preferencji jak chcemy, aby kod wyglądał, jednak cytując credo There should be one– and preferably only one — obvious way to do it.

Typowe zastosowanie kontekst menadżera to blokowanie lub chronienie zasobów (transakcja, wątek, plik, globalny stan). Nie mniej, społeczność wypracowała szersze zastosowanie, o czym przekonasz się w dalszej części.

W tym artykule:

  1. dowiesz się jak zaimplementować własny kontekst menedżer w bardzo zwięzłej formie, nie używając klas.
  2. zobaczysz jak dynamicznie tworzyć konteksty
  3. zapoznasz z się szerokim, często nieoczywistym, zastosowaniem wyrażenia with

Podstawowe API i pakiet contextlib

Wyrażenie with wprowadza określenie kontekstu wykonania kodu. Czym zapewnia, że niezależnie od tego co wydarzy się w ciele with, wcześniej zdefiniowane zachowanie poprzedzające (enter) i zamykające (exit) na pewno się wydarzy, nawet jeśli pojawi się wyjątek.

class CustomContext:
    def __enter__(self):
        print(f'Entering')
        
    def __exit__(self, type_, value, traceback):
        print(f'Exiting')
        
with CustomContext():
    print('Hello world')
    raise RuntimeError()

Jak widzisz, tworzenie własnego kontekst menadżera poprzez klasę, wymaga zdefiniowania metod __enter__ która nie definiuje parametrów i __exit__ która posiada trzy, opisujące ewentualny błąd który wystąpił w ciele kontekstu.

Ostatnia informacja zamykająca to krótkie wprowadzenie, to fakt że wartość którą zwróci __enter__ jest dostępna w kontekście przy użyciu słowa kluczowego as.

class BreakingBadContext:
    def __enter__(self):
        print('Say my name.')
        return self
        
    def __exit__(self, type_, value, traceback):
        print(f'You God damn right.')
        
    def answer(self):
        print('Walter White?')
        
with BreakingBadContext() as target:
    target.answer()
Say my name.
Walter White?
You God damn right.

Podsumowując, szablon wygląda następująco

class Context:
    def __enter__(self):
        <executed before with body>
        return <value which can be accessed using "as" keyword>
    def __exit__(self, type_, value, traceback):
        <executes regardless of what happens in with body>
        

Skrócona forma zapisu

Dekorator contextmanager to dla mnie najbardziej przydatny artefakt z modułu contextlib. Wystarczy udekorować nim generator i już mamy gotowy kontekst menadżer bez całego nakładu związanego z klasami.

import contextlib

@contextlib.contextmanager
def custom_context():
    try:
        print('Enter')
        yield 'Context value'
    finally:
        print('Exit')

Poniższe wyrażenia są równoważne

Zagnieżdżanie i dynamiczne wejście w kontekst

Kontynuujmy przegląd dobrodziejstw contextlib. Co jeśli chcesz tworzyć konteksty w pętli albo wejście definiować warunkiem? Na pomoc przychodzi ExitStack. Zacznijmy jednak od zilustrowania zagnieżdżonego with w podstawowej formie.

Klasyczne zagnieżdżanie wygląda następująco. Od razu jednak rzucają się w oczy problemy ze skalowaniem tego rozwiązania.


from contextlib import contextmanager

@contextmanager
def get_state(name):
    print("entering:", name)
    yield name
    print("exiting :", name)


with get_state("A") as A:
    with get_state("B") as B:
        with get_state("C") as C:
            print("inside with statement:", A, B, C)

Na ratunek przychodzi sugar syntax który sprawia że można osiągnąć ten sam efekt bez zagnieżdżeń

with \
    get_state("A") as A, \
    get_state("B") as B, 
    get_state("C") as C:
    
    print("inside with statement:", A, B, C)
entering: A
entering: B
entering: C
inside with statement: A B C
exiting : C
exiting : B
exiting : A

Zagnieżdżenia zachowują się jak stos (LIFO – list in, first out). Dlaczego jest to istotne? Ze względu na obsługę błędów. Jeśli pojawi się wyjątek dla n-tego wejścia to kolejne nie zostaną otworzone a poprzednie zamkną się według kolejności, co pokazuje następny blok kodu.

class GetState:
    def __init__(self,name, has_error):
        self.name = name
        self.has_error = has_error
        
    def __enter__(self):
        print('entering: ', self.name)
        if self.has_error:
            raise RuntimeError()
        
        
    def __exit__(self, type_, value, traceback):
        print('exiting: ', self.name, 'error type:', type_)

with \
    GetState('A', False), \
    GetState('B', False), \
    GetState('C', True):
    print('Wont be printed')
entering:  A
entering:  B
entering:  C
exiting:  B error type: <class 'RuntimeError'>
exiting:  A error type: <class 'RuntimeError'>

ExitStack

ExitStack to kontekst menadżer, który jest tak zaprojektowany, aby ułatwić komponowanie kontekstów między sobą, w dowolnym momencie wychodzić z nich oraz wdrożyć własną logikę. Przedstawiam Ci tylko jeden przykład jednak to tylko wycinek jego możliwości.

Dynamiczne otwieranie kontekstów wywołujemy na instancji ExitStack metodą enter_context. Samo wyjście z kontekstu ExitStack zamknie wszystkie pozostałe otwarte konteksty zgodnie z regułą LIFO. Pozwala on również na dynamiczne ich zamykanie metodą pop_all.

Odtwórzmy zatem przykład z zagnieżdżeniem kontekstów korzystając z ExitStack

with ExitStack() as es:
    names = ['A', 'B', 'C']
    ctx_values = []
    for name in names:
        value = es.enter_context(get_state(name))
        ctx_values.append(value)
        
    
    print(f'Inside with statement {ctx_values}')
entering: A
entering: B
entering: C
Inside with statement ['A', 'B', 'C']
exiting : C
exiting : B
exiting : A

Jestem ciekaw czy przekonałem Cię do używania context managera albo czy widzisz zastosowanie w swoim projekcie. Nie jest to święty Graal API Pythona, tylko bardziej wygodna forma zapisu. Zdarzają się sytuacje że korzystając z zewnętrznych narzędzi jesteś skazany na tę składnię, więc mam nadzieje że teraz już możesz swobodniej żonglować kontekstami.

Na koniec zostawiam Cię z przykładami które zebrałem z kilku książek, oficjalnej dokumentacji i społeczności, a w mojej ocenie są godne uwagi.

Przykłady zastosowań

Ta część pełnią formę książki kucharskiej dla wyrażenia with, gdzie każdy przepis obrazuje zmianę zachowania w kontekście lub zapewnienie zachowań na wejściu i wyjściu.

Zmiana koloru w konsoli


@contextmanager
def print_blue():
    print('\033[34m', end='')
    yield
    print('\033[39m', end='')

with print_blue():
    print('Changes color in context')

print('Outside the context with default color')

Zmiana poziomu logowania tylko w kontekście

import logging

logging.basicConfig()

@contextmanager
def debug_logging(logger_name: str, level: int):
    logger = logging.getLogger(logger_name)
    old_level = logger.getEffectiveLevel()
    logger.setLevel(level)
    try:
        yield logger
    finally:
        logger.setLevel(old_level)

with debug_logging('my-logger', logging.DEBUG) as logger:
    logger.debug('This will be printed')

logging\
.getLogger('my-logger')\
.info('This wont be logged because default level is WARNING')

Tłumienie oczekiwanych wyjątków

Celowe zignorowanie wyjątku. Ten przykład ilustruje redukowanie wizualnego szumu związanego z formą try/except

from contextlib import suppress
import os

with suppress(FileNotFoundError):
    os.remove('file.txt')
try:
    os.remove('file.txt')
except FileNotFoundError:
    pass

Tag maker

@contextmanager
def tag(name):
    print(f'<{name}>', end='')
    yield
    print(f'</{name}>', end='')

with tag('header'):
    print('Tag body', end='')
<header>Tag body</header>

Indenter

class Indenter:
    def __init__(self):
        self.level = 0

    def __enter__(self):
        self.level += 1
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.level -= 1

    def print(self, text):
        print('    ' * (self.level-1) + text)

with Indenter() as indenter:
    indenter.print('def mimic_python_syntax():')
    with indenter:
        indenter.print('s = "Hello World"')
        indenter.print('print(s)')

    indenter.print('\nmimic_python_syntax()')
def mimic_python_syntax():
    s = "Hello World"
    print(s)

mimic_python_syntax()

Transakcyjne operacje na listach

Kontekst tworzy listę roboczą na której wykonujemy operacje w kontekście. Jeśli pojawi się wyjątek, to wartości w liście początkowej nie zostaną podmienione.

@contextmanager
def list_transaction(list_: list):
    working = list(list_)
    yield working
    list_[:] = working

items = [1,2,3]

with list_transaction(items) as working:
    working.append(4)
    raise RuntimeError()
print(items)
[1, 2, 3]

Stoper, mierzenie czasu wykonania

import time
@contextmanager
def stopwatch(label: str):
    start = time.time()
    try:
        yield
    finally:
        end = time.time()
        print(f'{label}: {end - start}')


with stopwatch('Sleeping'):
    time.sleep(1)
Sleeping: 1.0004658699035645

Sesja HTTP

Wykorzystanie istniejącego połączenia TCP w celu zredukowaniu czasu zapytań sieciowych

import requests

n = 20

url = "http://httpbin.org/cookies/set/sessioncookie/123456789"
with stopwatch('Using context manager'):
    with requests.Session() as session:
        for _ in range(n):
            session.get(url)

with stopwatch('Establishing HTTP connection for every request'):
    for _ in range(n):
        requests.get(url)

Wyodrębnienie logowania i obsługi błędów

import sys
import logging
from contextlib import contextmanager


logging.getLogger(__name__)

logging.basicConfig(
    level=logging.INFO,
    format="\n(asctime)s [%(levelname)s] %(message)s",
)


class Divider:

    @contextmanager
    def errorhandler(self):
        try:
            yield
        except ZeroDivisionError:
            print(
                f"Custom handling of Zero Division Error! Printing "
                "only 2 levels of traceback.."
            )
            logging.exception("ZeroDivisionError")

    def __call__(self, a, b):
        """Function that we want to save from nasty error handling logic."""

        with self.errorhandler():
            return a / b


divide = Divider()
divide(2, 0)

Te oraz inne przykłady w formie Jupyter Notebook znajdziesz tutaj

Lub bezpośrednio w repozytorium tutaj

Udostępnij ten wpis


Dobrnąłeś do końca. Jeśli ten artykuł był dla Ciebie wartościowy i chcesz otrzymywać informacje o kolejnych, to zapraszam Cię do zapisania się do listy mailingowej. Gwarantuję zero spamu.

Radek.

Źródła

Inne artykuły

Partycjonowanie bazy danych okładka

Partycjonowanie bazy danych

Partycjonowanie pozwala podzielić tabelę na mniejsze części, gdzie każda z nich może się znajdować na innym serwerze. Zobacz, jak to działa oraz dlaczego jest to …