17. Context Managers

17.1. About Context Managers

17.1.1. How to create

  • __enter__()
  • __exit__()
Code Listing 17.1. How to create Context Managers
class MyClass:
    def __enter__(self):
        return self

    def __exit__(self, *args):
        return None

17.1.2. Zastosowanie

  • Pliki
  • Połączenia do bazy danych
  • Lock
  • Stream siecowe

17.1.3. Przykład

f = open(FILENAME)
# ...
f.close()
with open(FILENAME) as file:
    # ...

17.1.4. Lock

import threading


lock = threading.Lock()

with lock:
    my_list.append(item)

replaces the more verbose:

import threading


lock = threading.Lock()
lock.acquire()

try:
    my_list.append(item)
finally:
    lock.release()

17.1.5. Database

import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""
SQL_INSERT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'
SQL_SELECT = 'SELECT * from astronauts'


astronauts = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Matt', 'lastname': 'Kowalski'},
    {'pesel': '61041212347', 'firstname': 'Melissa', 'lastname': 'Lewis'},
    {'pesel': '61041212348', 'firstname': 'Alex', 'lastname': 'Vogel'},
    {'pesel': '61041212349', 'firstname': 'Ryan', 'lastname': 'Stone'},
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)
    db.executemany(SQL_INSERT, astronauts)

    for row in db.execute(SQL_SELECT):
        print(row)

17.2. Contextmanager decorator

from contextlib import contextmanager

@contextmanager
def tag(name):
    print(f"<{name}>")
    yield
    print(f"</{name}>")

with tag("p"):
    print("foo")

# <p>
# foo
# </p>

Dzieli naszą funkcję na bloki przed i po yield.

  • Bloki przed traktuje jako __enter__()
  • Bloki za traktuje jako __exit__()
from contextlib import ContextDecorator

class makeparagraph(ContextDecorator):
    def __enter__(self):
        print('<p>')
        return self

    def __exit__(self, *exc):
        print('</p>')
        return False

@makeparagraph()
def generate_html():
    print('Here is some non-HTML')

generate_html()
<p>
Here is some non-HTML
</p>

17.2.1. timing

from contextlib import contextmanager
import time


@contextmanager
def time_print(task_name):
    start = time.time()

    try:
        yield
    finally:
        duration = time.time() - start
        print(f'{task_name} took {duration} seconds')


with time_print("processes"):
    [doproc() for _ in range(500)]
# processes took 15.236166954 seconds


with time_print("threads"):
    [dothread() for _ in range(500)]
# threads took 0.11357998848 seconds

17.3. Assignments

17.3.1. Buffered file

  1. Stwórz Context Manager dla zapisu do plików
  2. Context Manager buforuje dane (nie zapisuje ich na bieżąco
  3. Gdy program wyjdzie z bloku context managera, to nastąpi zapisanie do pliku
FILENAME = '/tmp/context-manager.txt'

class File:
    pass


with File(FILENAME, encoding='utf-8') as file:
    file.append_line(...)
    file.append_line(...)
    file.append_line(...)

# after block with exits, save to file