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

Dlaczego warto pisać? Jak pisać?

Pisania można się nauczyć; pisania, czyli myślenia. Na najbardziej abstrakcyjnych poziomach twój umysł jest zorganizowany w sposób werbalny. Tym samym jeśli dzięki pisaniu nauczysz się …

Transakcje ACID

W trakcie zapisu danych do bazy wiele rzeczy może pójść nie tak. Może zapełnić się dysk, połączenie sieciowe zostanie zerwane pomiędzy bazą a klientem, przewróci się aplikacja, która wysyła …
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 …