2. Context Managers

2.1. Protocol

  • __enter__(self, *args, **kwargs) -> self

  • __exit__(self, *args, **kwargs) -> None

2.2. Application

  • Files

  • Buffering data

  • Database connection

  • Database transactions

  • Database cursors

  • Locks

  • Network sockets

  • Network streams

  • HTTP sessions

2.3. Implementation

Listing 356. How to create Context Managers
class MyClass:
    def __enter__(self):
        print('Entering the block')
        return self

    def __exit__(self, *args):
        print('Exiting the block')
        return None

    def hello(self):
        print('I am inside')


with MyClass() as cls:
    cls.hello()

# Entering the block
# I am inside
# Exiting the block

2.4. Examples

2.4.1. Files

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

2.4.2. Database

import sqlite3


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


astronauts = [
    {'pesel': '61041212345', 'first_name': 'José', 'last_name': 'Jiménez'},
    {'pesel': '61041212346', 'first_name': 'Jan', 'last_name': 'Twardowski'},
    {'pesel': '61041212347', 'first_name': 'Melissa', 'last_name': 'Lewis'},
    {'pesel': '61041212348', 'first_name': 'Alex', 'last_name': 'Vogel'},
    {'pesel': '61041212349', 'first_name': 'Ryan', 'last_name': '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)

2.4.3. Lock

from threading import Lock

my_list = [1, 2, 3]


with Lock() as lock:
    my_list.append(4)
from threading import Lock

my_list = [1, 2, 3]
lock = Lock()


with lock:
    my_list.append(4)

replaces the more verbose:

from threading import Lock

lock = Lock()
lock.acquire()

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

2.5. Contextmanager decorator

  • Split function for before and after yield

  • Code before yield becomes __enter__()

  • Code after yield becomes __exit__()

2.5.1. contextmanager decorator

from contextlib import contextmanager
import time


@contextmanager
def MicroBenchmark():
    start_time = time.time()
    yield
    end_time = time.time()
    duration = end_time - start_time
    print(f'Duration {duration:.4f} seconds')


with MicroBenchmark():
    list(range(100_000_000))

# Duration 3.3795 seconds

2.5.2. ContextDecorator class

from contextlib import ContextDecorator
import time


class MicroBenchmark(ContextDecorator):
    def __enter__(self):
        self.start_time = time.time()
        return self

    def __exit__(self, *args):
        end_time = time.time()
        duration = end_time - self.start_time
        print(f'Duration {duration:.4f} seconds')


@MicroBenchmark()
def my_function():
    list(range(100_000_000))


my_function()
# Duration 3.4697 seconds
class Timeit:
    def __init__(self, name):
        self.name = name

    def __enter__(self):
        self.start_time = datetime.now().timestamp()

    def __exit__(self, *arg, **kwargs):
        self.end_time = datetime.now().timestamp()
        duration = self.end_time - self.start_time
        print(f'Duration of {self.name} is {duration:f} seconds')


with Timeit('function'):
    list(get_for_species_function(data, 'setosa'))

with Timeit('comprehension'):
    list([row for row in data if row[4] == 'setosa'])

with Timeit('generator short'):
    list((row for row in data if row[4] == 'setosa'))

with Timeit('generator'):
    list(get_for_species_generator(data, 'setosa'))

2.5.3. Use Case

from contextlib import contextmanager


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


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

# <p>
# foo
# </p>

2.6. Assignments

2.6.1. Buffered file

English
  1. Take input code from listing below

  2. Create Context manager for file which buffers data before save

  3. When block closes, then open file and write data

  4. How to make buffer save data every X bytes?

  5. How to make buffer save data every X seconds?

  6. How to make buffer save data in the background, but it could be still used?

Polish
  1. Weź kod wejściowy z listingu poniżej

  2. Stwórz Context Manager dla plików, który buforuje dane przed zapisem

  3. Gdy nastąpi wyjście z bloku context managera, to otwórz plik i zapisz dane

  4. Jak zrobić, aby bufor zapisywał dane na dysku co X bajtów?

  5. Jak zrobić, aby bufor zapisywał dane na dysku co X sekund?

  6. Jak zrobić, aby do bufora podczas zapisu na dysk, nadal można było pisać?

Input
FILENAME = '/tmp/context-manager.txt'

class File:
    pass


with File(FILENAME) as file:
    file.append_line(...)
    file.append_line(...)
    file.append_line(...)

# after block with exits, save to file