6.2. Context Manager

6.2.1. Rationale

  • Files

  • Buffering data

  • Database connection

  • Database transactions

  • Database cursors

  • Locks

  • Network sockets

  • Network streams

  • HTTP sessions

6.2.2. Protocol

  • __enter__(self) -> self

  • __exit__(self, *args) -> None

class ContextManager:
    def __enter__(self):
        return self

    def __exit__(self, *args):
        return None
with ContextManager as cm:
    ...

6.2.3. Example

class MyClass:
    def __enter__(self):
        print('Entering the block')
        return self

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

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


with MyClass() as my:
    my.do_something()

# Entering the block
# I am inside
# Exiting the block
class Rocket:
    def __enter__(self):
        print('Launching')
        return self

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

    def fly_to_space(self):
        print('I am in space!')


with Rocket() as rocket:
    rocket.fly_to_space()

# Launching
# I am in space!
# Landing

6.2.4. Inheritance

from contextlib import ContextDecorator
from time import time


class Timeit(ContextDecorator):
    def __enter__(self):
        self.start = time()
        return self

    def __exit__(self, *args):
        end = time()
        print(f'Duration {end-self.start:.2f} seconds')


@Timeit()
def myfunction():
    list(range(100_000_000))


myfunction()
# Duration 3.90 seconds

6.2.5. Decorator

  • Split function for before and after yield

  • Code before yield becomes __enter__()

  • Code after yield becomes __exit__()

from contextlib import contextmanager
from time import time


@contextmanager
def timeit():
    start = time()
    yield
    end = time()
    print(f'Duration {end-start:.4f} seconds')


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

# Duration 4.0250 seconds
from contextlib import contextmanager


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


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

# <p>
# foo
# </p>

6.2.6. Use Cases

6.2.6.1. Files

f = open(FILE)

try:
    content = f.read()
finally:
    f.close()
with open(FILE) as f:
    content = f.read()

6.2.6.2. 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'

DATA = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Jan', 'lastname': 'Twardowski'},
    {'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, DATA)

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

6.2.6.3. Lock

from threading import Lock

# Make lock
lock = Lock()

# Use lock
lock.acquire()

try:
    print('Critical section 1')
    print('Critical section 2')
finally:
    lock.release()
from threading import Lock

# Make lock
lock = Lock()

# Use lock
with lock:
    print('Critical section 1')
    print('Critical section 2')

6.2.6.4. String Microbenchmark

from time import time


class Timeit:
    def __init__(self, name):
        self.name = name

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

    def __exit__(self, *arg):
        end = time()
        print(f'Duration of {self.name} is {end-self.start:.2f} second')


a = 1
b = 2
repetitions = int(1e7)


with Timeit('f-string'):
    for _ in range(repetitions):
        f'{a}{b}'

with Timeit('string concat'):
    for _ in range(repetitions):
        a + b

with Timeit('str.format()'):
    for _ in range(repetitions):
        '{0}{1}'.format(a, b)

with Timeit('str.format()'):
    for _ in range(repetitions):
        '{}{}'.format(a, b)

with Timeit('str.format()'):
    for _ in range(repetitions):
        '{a}{b}'.format(a=a, b=b)

with Timeit('%-style'):
    for _ in range(repetitions):
        '%s%s' % (a, b)

with Timeit('%-style'):
    for _ in range(repetitions):
        '%d%d' % (a, b)

with Timeit('%-style'):
    for _ in range(repetitions):
        '%f%f' % (a, b)

# Duration of f-string is 2.70 second
# Duration of string concat is 0.68 second
# Duration of str.format() is 3.46 second
# Duration of str.format() is 3.37 second
# Duration of str.format() is 4.85 second
# Duration of %-style is 2.59 second
# Duration of %-style is 2.59 second
# Duration of %-style is 3.82 second

6.2.7. Assignments

6.2.7.1. Protocol ContextManager File

  • Assignment name: Protocol ContextManager File

  • Last update: 2020-10-02

  • Complexity level: easy

  • Lines of code to write: 13 lines

  • Estimated time of completion: 13 min

  • Solution: solution/protocol_contextmanager_file.py

English
  1. Use data from "Input" section (see below)

  2. Define class File with parameter: filename: str

  3. File must implement Context Manager protocol

  4. File buffers lines added using File.append(text: str) method

  5. On with block exit File class opens file and write buffer

  6. Compare result with "Output" section (see below)

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Stwórz klasę File z parametrem: filename: str

  3. File ma implementować protokół Context Manager

  4. File buforuje linie dodawane za pomocą metody File.append(text: str)

  5. Na wyjściu z bloku with klasa File otwiera plik i zapisuje bufor

  6. Porównaj wyniki z sekcją "Output" (patrz poniżej)

Output
>>> from inspect import isclass, ismethod
>>> assert isclass(File)
>>> assert hasattr(File, 'append')
>>> assert hasattr(File, '__enter__')
>>> assert hasattr(File, '__exit__')
>>> assert ismethod(File(None).append)
>>> assert ismethod(File(None).__enter__)
>>> assert ismethod(File(None).__exit__)

>>> with File('_temporary.txt') as file:
...    file.append('One')
...    file.append('Two')

>>> open('_temporary.txt').read()
'One\\nTwo\\n'
Hint
  • Append newline character (\n) before adding to buffer

6.2.7.2. Protocol ContextManager Buffer

  • Assignment name: Protocol ContextManager Buffer

  • Last update: 2020-10-16

  • Complexity level: medium

  • Lines of code to write: 22 lines

  • Estimated time of completion: 21 min

  • Solution: solution/protocol_contextmanager_buffer.py

English
  1. Use data from "Input" section (see below)

  2. Define class configuration attribute BUFFER_LIMIT: int = 100 bytes

  3. File has to be written to disk every X bytes of buffer

  4. Writing and reading takes time, how to make buffer save data in the background, but it could be still used?

  5. Compare result with "Output" section (see below)

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Zdefiniuj klasowy atrybut konfiguracyjny BUFFER_LIMIT: int = 100 bajtów

  3. Plik na dysku ma być zapisywany co X bajtów bufora

  4. Operacje zapisu i odczytu trwają, jak zrobić, aby do bufora podczas zapisu na dysk, nadal można było pisać?

  5. Porównaj wyniki z sekcją "Output" (patrz poniżej)

Output
>>> from inspect import isclass, ismethod
>>> assert isclass(File)
>>> assert hasattr(File, 'append')
>>> assert hasattr(File, 'BUFFER_LIMIT')
>>> assert hasattr(File, '__enter__')
>>> assert hasattr(File, '__exit__')
>>> assert ismethod(File(None).append)
>>> assert ismethod(File(None).__enter__)
>>> assert ismethod(File(None).__exit__)
>>> assert FILE.BUFFER_LIMIT == 100

>>> with File('_temporary.txt') as file:
...    file.append('One')
...    file.append('Two')
...    file.append('Three')
...    file.append('Four')
...    file.append('Five')
...    file.append('Six')

>>> open('_temporary.txt').read()
'One\\nTwo\\nThree\\nFour\\nFive\\nSix\\n'
Hint
  • sys.getsizeof()

6.2.7.3. Protocol ContextManager AutoSave

  • Assignment name: Protocol Context Manager AutoSave

  • Last update: 2020-10-16

  • Complexity level: hard

  • Lines of code to write: 32 lines

  • Estimated time of completion: 21 min

  • Solution: solution/protocol_contextmanager_autosave.py

English
  1. Use data from "Input" section (see below)

  2. Define class configuration attribute AUTOSAVE_SECONDS: float = 2.0

  3. Save buffer content to file every AUTOSAVE_SECONDS seconds

  4. Writing and reading takes time, how to make buffer save data in the background, but it could be still used?

  5. Compare result with "Output" section (see below)

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Zdefiniuj klasowy atrybut konfiguracyjny AUTOSAVE_SECONDS: float = 2.0

  3. Zapisuj zawartość bufora do pliku co AUTOSAVE_SECONDS sekund

  4. Operacje zapisu i odczytu trwają, jak zrobić, aby do bufora podczas zapisu na dysk, nadal można było pisać?

  5. Porównaj wyniki z sekcją "Output" (patrz poniżej)

Output
>>> from inspect import isclass, ismethod
>>> assert isclass(File)
>>> assert hasattr(File, 'append')
>>> assert hasattr(File, 'AUTOSAVE_SECONDS')
>>> assert hasattr(File, '__enter__')
>>> assert hasattr(File, '__exit__')
>>> assert ismethod(File(None).append)
>>> assert ismethod(File(None).__enter__)
>>> assert ismethod(File(None).__exit__)
>>> assert FILE.AUTOSAVE_SECONDS == 2.0

>>> with File('_temporary.txt') as file:
...    file.append('One')
...    file.append('Two')
...    file.append('Three')
...    file.append('Four')
...    file.append('Five')
...    file.append('Six')

>>> open('_temporary.txt').read()
'One\\nTwo\\nThree\\nFour\\nFive\\nSix\\n'