Context Manager ochroni Twoje plecy!

Menadżer kontekstu (ang. context manager) to obiekt, który odpowiednio zarządza danym zasobem, zapewniając, że zostanie on odpowiednio zamknięty. Przez zamknięty mam tu na myśli czyszczenie, zwalnianie zasobów oraz sprzątanie po wykonaniu bloku kodu. Dobrym przykładem takiego obiektu, jest użycie open w Pythonie. Korzystając z niego w domyślny sposób, otwieramy plik, odczytujemy pewien jego element, a następnie go zamykamy.

f = open('versions.txt')
print(f.readline())
f.close()

Jednak open to też menadżer kontekstu, aby z niego skorzystać, należy skorzystać z konstrukcji with ... as ... :, gdzie samo as nie jest wymagane gdy nie potrzebujemy w danym kontekście korzystać z jego obiektu. Więc po with podajemy nasz menadżer, czyli open(). Następnie, aby mieć dostęp do otwieranego pliku, dodajemy as i zmienną, ja użyję zmiennej f.

with open('base.py') as f:
    print(f.readline())

Dzięki ten konstrukcji nie musimy korzystać z f.close(), plik zostanie zamknięty samoczynnie po wyjściu z kontekstu, czyli po zakończeniu działania na kodzie, który znalazł się w jej ciele poprzez wcięcie.

Baza

Aby pokazać więcej na ten temat, utworzyłem klasę Storage, która będzie symulować magazyn danych. Dla zabawy, przy próbie pobrania czegokolwiek z zasobu możemy dostać wyjątek StorageException. Aby z niego korzystać, musimy uprzednio zablokować dostęp do zasobu, a na koniec zasób zwolnić.

import random


class StorageException(Exception):
    """Raised when there is some issue with storage"""


class Storage():

    def __init__(self):
        print('init')

    def block_resources(self):
        print('block_resources')

    def release_resources(self):
        print("release_resources")

    def get(self, *args, **kwargs):
        if random.randint(0, 1):
            raise StorageException("Something goes wrong")
        print("get", args, kwargs)

Domyślne użycie tej klasy wyglądałoby następująco.

from base import Storage

storage = Storage()
storage.block_resources()
storage.get(filename='secrets.yaml')
storage.release_resources()

init
block_resources
get () {'filename': 'secrets.yaml'}
release_resources

O ile nie wystąpiłby błąd, wszystko byłoby ok. Jednak czasem przy wywołaniu możemy dostać błąd. Spowoduje to nie wykonanie się ostatniej linii kodu, a to oznacza zablokowanie zasobu do momentu, gdy ktoś manualnie go odblokuje lub proces utrzymujący dostęp do zasobu zostanie zabity i system przywróci zasób do swojej puli.

init
block_resources
Traceback (most recent call last):
    storage.get(filename='secrets.yaml')
    raise StorageException("Something goes wrong")
base.StorageException: Something goes wrong

Dispose Pattern

Taki problem rozwiązuje wzorzec projektowy Dispose, który służy właśnie do zamykania zasobów w przypadku gdy pojawi się błąd. Jego najprostsza implementacja w naszym przypadku korzysta z try finally, więc jeżeli cokolwiek wydarzy się w naszym kodzie, to przed ostatecznym wywołaniem wyjątku i zakończeniem działania kodu wykona się kod sekcji finally. Dzięki temu mamy pewność, że nasz zasób zawsze zostanie zamknięty, czy też w tym przypadku zwolniony.

from base import Storage

storage = Storage()
storage.block_resources()
try:
    storage.get(filename='secrets.yaml')
finally:
    storage.release_resources()

Mała uwaga, nie próbuj w bloku finally zwracać żadnej wartości. Spowoduje to nadpisanie tego, co powinno zostać zwrócone. Utracisz w ten sposób wartość zwracaną z kodu, lub zignorujesz wyjątek.

Elementy menadżera kontekstu

Przejdźmy do stworzenia własnego menadżera, bez korzystania z dokumentacji. Będziemy zatem obserwować błędy wyrzucane po uruchomieniu kodu, i dodawać do niego brakujące elementy. Więc w pierwszej kolejności tworzymy pustą klasę, i spróbujemy ją użyć z instrukcją with.

class StorageContext:
    pass


with StorageContext() as storage:
    print(storage)

    with StorageContext() as storage:
AttributeError: __enter__

Wyjątek, który się pojawił, mówi jasno, iż brakuje nam atrybutu __enter__. Podejrzewam, że chodzi o metodę, dlatego dodaje do naszej klasy taką metodę. Następnie uruchamiam kod ponownie.

class StorageContext:

    def __enter__(self):
        pass


with StorageContext() as storage:
    print(storage)

    with StorageContext() as storage:
AttributeError: __exit__

To ma sens. Mamy metodę __enter__ do otwierania pewnego kontekstu, ale potrzebujemy także __exit__ aby ten kontekst zakończyć. Dodajemy metodę i uruchamiamy kod ponownie.

class StorageContext:

    def __enter__(self):
        pass

    def __exit__(self):
        pass


with StorageContext() as storage:
    print(storage)

TypeError: __exit__() takes 1 positional argument but 4 were given

Dostajemy informacje, że metoda __exit__() przyjmuje aż 4 argumenty pozycyjne. Jeden to self, resztę opakujemy w *args, sprawdzimy potem, co się w nich kryje. Uruchamiamy ponownie.

class StorageContext:

    def __enter__(self):
        pass

    def __exit__(self, *args):
        pass


with StorageContext() as storage:
    print(storage)

None

Kod uruchomił się bez błędów. Udało się zdefiniować szkielet, jakiego wymaga Python. Popracujmy dalej nad tym, aby funkcja print zwróciła None. Podejrzewam, że metoda __enter__ powinna zwrócić jakiś obiekt, na którym będziemy pracować, tak więc dodaje return self.

class StorageContext:

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass


with StorageContext() as storage:
    print(storage)

<__main__.StorageContext object at 0x7fc769596ee0>

Tak jak podejrzewałem, aby móc pracować na obiekcie ograniczonym naszym kontekstem, w metodzie __enter__ należy ten obiekt zwrócić. Została ostatnia magiczna rzecz, a jest nią zbiór argumentów pozycyjnych w metodzie __exit__. Dodajmy tam print aby się przekonać.

class StorageContext:

    def __enter__(self):
        return self

    def __exit__(self, *args):
        print(args)


with StorageContext() as storage:
    print(storage)

<__main__.StorageContext object at 0x7f1f9aaf2ee0>
(None, None, None)

Jest pusto, to dziwne. Jednak pamiętajmy, że wzorzec projektowy korzystał z obsługi błędów, więc może jego wywołanie coś zmieni.

class StorageContext:

    def __enter__(self):
        return self

    def __exit__(self, *args):
        print(args)


with StorageContext() as storage:
    raise ValueError

(<class 'ValueError'>, ValueError(), <traceback object at 0x10ab4ae88>)
Traceback (most recent call last):
    raise ValueError
ValueError

Mamy wyjaśnienie. Metoda __exit__ dostaje informacje o wyjątkach, które się pojawiły. Dzięki temu możemy te wyjątki odpowiednio obsłużyć lub ich obsługę pominąć. Jest jeszcze jedna rzecz, którą ciężko by mi było wypracować. W momencie gdy __exit__ zwraca wartość domyślną None, to wyjątek i tak zostanie wyrzucony. Aby do tego nie dopuścić, należy zwrócić z metody __exit__ wartość True.

Implementacja za pomocą klasy

Ostateczna implementacja naszego kodu do obsługi zasobu za pomocą klasy wygląda zatem następująco:

from base import Storage


class StorageContext:
    def __init__(self):
        self.storage = Storage()

    def __enter__(self):
        self.storage.block_resources()
        return self.storage

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.storage.release_resources()


with StorageContext() as s:
    s.get(filename='secrets.yaml')

Implementacja za pomocą funkcji

Ktoś mądry pomyślał, że dobrym pomysłem będzie tworzenie prostych menadżerów kontekstu opartych o generator. W taki sposób powstał dekorator contextmanager z biblioteki wbudowanej contextlib, dzięki któremu możemy przekształcić generator w menadżera kontekstu. Sposób implementacji naszego problemu za pomocą tego narzędzia poniżej. Rozwiązanie to jest bardzo często używane, gdyż ma prostszą konstrukcję oraz dużo łatwiej go zaimplementować.

from contextlib import contextmanager
from base import Storage

@contextmanager
def storage_access():
    storage = Storage()
    storage.block_resources()
    try:
        yield storage
    finally:
        storage.release_resources()

with storage_access() as s:
    s.get(filename='secrets.yaml')

Przykłady

Skoro wiemy już, jak tworzyć context manager, to co można z nim zrobić? Jeden z najważniejszych przypadków pokazałem na powyższych przykładach, czyli dostęp do pewnych zasobów, które wymagają konkretnej operacji przed i po.

Code Timer

Inny ciekawy przykład, to stworzenie klasy do mierzenia czasu działania danego kawałka kodu. Z reguły jest to przykład na dekorator, ale wymaga on wtedy zamknięcia naszego kodu w funkcji. Licznik oparty o menadżera kontekstu nie ma tego ograniczenia.

Dodatkowo pojawia się tutaj prosta konfiguracja logera, dzięki której możemy logować informacje z naszego kodu. Korzystanie z logera jest dużo lepszym wzorcem niż korzystanie z print w naszym kodzie, ponieważ loger możemy odpowiednio skonfigurować, aby dla przykładu wysyłał nasze logi do platformy , gdzie te logi możemy następnie przejrzeć.

import time
import logging

logging.basicConfig(level=logging.DEBUG)


class LogElapsedTime:
    def __init__(self, name=""):
        self.start = None
        self.name = name

    def __enter__(self):
        self.start = time.time()

    def __exit__(self, *args):
        end = time.time()
        delta = end - self.start
        logging.info("time of execution '%s':%s", self.name, delta)


with LogElapsedTime("sleep"):
    time.sleep(1)

INFO:root:time of execution 'sleep':1.0010440349578857

Ignore Exceptions

Inną propozycją ode mnie, jest context manager, którego zadaniem będzie ignorowanie wszelkich błędów, które wystąpią. Nie chcemy jednak stracić informacji o tym, iż wystąpiły, dlatego odpowiednia informacja zostanie przekazana do logera.

import logging

logging.basicConfig(level=logging.DEBUG)


class IgnoreExceptions:
    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            logging.error("exception ignored : %s", exc_type.__name__)
        return True


with IgnoreExceptions():
    raise ArithmeticError

ERROR:root:exception ignored : ArithmeticError