seminars.fb

Seminar (Fri Oct 15): “Seeing Things in Context with Context Managers”

Title	Seeing Things in Context with Context Managers
Topic	Resource Management & Context Managers
Date	Fri Oct 15
Keywords	context managers, with-statement, asynchronous context managers, contextlib, __del__, __weakref__, PEP-343, PEP-567, PEP-492

Audience

These sessions are designed for a broad audience of non-software engineers and software programmers of all backgrounds and skill-levels.

Our expected audience should comprise attendees with a…

• working familiarty with basic Python syntax.
• a desire to use more advanced features in Python like PEP-343 Context Managers.

During this session, we will endeavour to guide our audience to developing…

• complete familiarity with the metaphors and motivation behind the use of PEP-343 Context Managers
• complete familiarity with common use-cases and common built-in or standard library Context Managers
• proficiency in writing PEP-343 Context Managers (using contextlib.contextmanager)
• familiarity with topics related to Context Managers, including context-local state (PEP-567 Context Variables) and interaction with asyncio code (PEP-492 Asynchronous Context Managers)

Abstract

In previous seminars, we have discussed the motivation, mechanisms, and metaphors provided by advanced Python features such as generators, coroutines, and aspects of the OO model.

In this seminar, we will tackle PEP-343 Context Managers. We’ll discuss the motivation of deterministic management of resources, prior approaches and potential missteps and misapprehensions when coming from other programming languages. We’ll discuss the “metaphor” that context manager provide, and how this sequencing metaphor leads to a direct relationship between context managers and generators (typically via contextlib.contextmanager.) We’ll discuss the mechanism behind context managers, including the OO model API, as well as details related to context manager in asynchronous code (and the motivation and appropriate use of async with syntax.) Finally, we’ll discuss common problems related to composition of context managers, as well as the need for context-local state (and the subsequent development of PEP-567 context variables.)

Agenda:

• PEP-343: the with statement
• contextlib
• PEP-492: Coroutines with async and await syntax & asynchronous context managers
• PEP-567: context variables
• __del__ and weakref

What’s Next?

Did you enjoy this seminar? Did you learn something new that will help you as you use advanced features in Python more and more in your work?

In a future seminar, we can discuss other advanced Python syntax. We can discuss:

• the metaphors, mechanisms, and motivations behind decorators, including class decorators and higher-order decorators
• how to use context managers to design elegant, extensible APIs, as well as common compositional problems that users run into when designing around context managers.
• how the various asyncio-syntaxes interact (i.e., async def, async for, async with) including why there is no async while and why it is conceptually reasonable by mechanically impossible to yield from an asynchronous context manager.

If you’re interested in any of these topics, please let us know! Send us an e-mail at learning@dutc.io or contact us over Workplace with your feedback!

Notes

print("Let's go!")
print("Let's go!")

Why Context Managers?

f = open('/tmp/temp.dat', 'w')
f.write('…')
f.close()

f = open('/tmp/temp.dat', 'w')
f.write('…')
0 / 0
f.close()

try:
    f = open('/tmp/temp.dat', 'w')
    f.write('…')
finally:
    f.close()

with open('/tmp/temp.dat', 'w') as f:
    f.write('…')

with open('/tmp/temp.dat', 'w') as f:
    print(f'{not f.closed = }')
print(f'{not f.closed = }')
print(f'{f = }')

with open('/tmp/temp.dat', 'w') as f:
    for x in range(10):
        f.write(f'{x}\n')

with open('/tmp/temp.dat') as f:
    pass
print(f'{f.readlines() = }')

with open('/tmp/temp.dat', 'w') as f:
    for x in range(10):
        f.write(f'{x}\n')

with open('/tmp/temp.dat') as f:
    numbers = (int(x) for x in f)
    print(f'{[*numbers] = }')

from tempfile import TemporaryDirectory
from pathlib import Path

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    print(f'{d              = }')
    print(f'{d.exists()     = }')
    print(f'{[*d.iterdir()] = }')

print(f'{d.exists()     = }')

from tempfile import TemporaryDirectory, NamedTemporaryFile
from pathlib import Path

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    with NamedTemporaryFile(mode='w', dir=d, delete=False) as f:
        p = Path(f.name)
        print(f'{d          = }')
        print(f'{p          = }')
        print(f'{d.exists() = }')
        print(f'{p.exists() = }')
        f.write('\n')
print(f'{p.exists() = }')
print(f'{d.exists() = }')

from tempfile import TemporaryDirectory, NamedTemporaryFile
from pathlib import Path

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    with NamedTemporaryFile(mode='w', dir=d, delete=False) as f:
        p = Path(f.name)
        print(f'{d             = }')
        print(f'{p             = }')
        print(f'{d.exists()    = }')
        print(f'{p.exists()    = }')
        for x in range(10):
            f.write(f'{x}\n')
    #  with open(p) as f:
    #      print(f'{f.readlines() = }')
    print(f'{p.exists()    = }')
print(f'{d.exists()    = }')

from tempfile import TemporaryDirectory, NamedTemporaryFile
from sqlite3 import connect
from pathlib import Path
from collections import namedtuple
from random import randint

Row = namedtuple('Row', 'a b')

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    with NamedTemporaryFile(mode='w', dir=d) as f:
        p = Path(f.name).with_suffix('.db')
        with connect(p) as con:
            cur = con.cursor()

            cur.execute('create table test (a int, b int)')
            cur.executemany('insert into test values (?, ?)', [
                Row(randint(-10, +10), randint(-10, +10)) for _ in range(10)
            ])

        with connect(p) as con:
            cur = con.cursor()

            res = cur.execute('select count(*) from test').fetchone()
            print(f'{res = }')

            for res in cur.execute('select a, sum(a + b), avg(b) from test group by a'):
                print(f'{res = }')

def f():
    return g()

def g():
    return h()

def main():
    return f()

def f(): return g()
def g(): return h()
def h(): pass

def step0():
    return f()

def step1(): pass
def step2(): pass

def main():
    step0()
    step1()
    step2()

def f(): pass
def g(): pass
def h(): pass

def step0():
    return f()

def step1():
    pass

def step2():
    pass

def main():
    step0()
    with open('sqlite.db') as f:
        step1()
        step2()

How Context Managers?

for x in xs:
    pass

with open(__file__) as f:
    for line in f:
        pass

from tempfile import TemporaryDirectory, NamedTemporaryFile
from sqlite3 import connect
from pathlib import Path
from collections import namedtuple
from random import randint

class Row(namedtuple('RowBase', 'a b')):
    @classmethod
    def from_random(cls):
        return cls(a=randint(-10, +10), b=randint(-10, +10))

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    with NamedTemporaryFile(mode='w', dir=d) as f:
        p = Path(f.name).with_suffix('.db')
        with connect(p) as con:
            cur = con.cursor()

            cur.execute('create table test (a int, b int)')
            cur.executemany('insert into test values (?, ?)', [
                Row.from_random() for _ in range(10)
            ])

        with connect(p) as con:
            for res in cur.execute('select a, avg(b) from test group by a'):
                print(f'{res = }')

from enum import Enum
from random import choice

Status = Enum('Status', 'Success Failure Unknown')

for x in Status:
    print(f'{x = }')
print(f'{choice([*Status]) = }')

xs = 'abc'
for x in xs:
    print(f'{x = }')

xi = iter(xs)
while True:
    try:
        x = next(xi)
        print(f'{x = }')
    except StopIteration:
        break

class T:
    def __init__(self, data):
        self.data = [*data]
        self.index = 0
    def __iter__(self):
        return self
    def __next__(self):
        try:
            rv = self.data[self.index]
            self.index += 1
        except IndexError as e:
            raise StopIteration() from e
        else:
            return rv

for x in T(range(3)):
    print(f'{x = }')

from collections.abc import Iterator, Iterable

class T:
    def __iter__(self): pass
    def __next__(self): pass

print(f'{isinstance(T(), Iterator) = }')
print(f'{isinstance(T(), Iterable) = }')

from collections.abc import Iterator, Iterable

class T:
    def __iter__(self):
        return TIter()

class TIter:
    def __iter__(self): return self
    def __next__(self): pass

x = T()
xi = iter(x)

print(f'{isinstance(x, Iterable)  = }')
print(f'{isinstance(xi, Iterator) = }')

from collections.abc import Iterator, Iterable
from dataclasses import dataclass

@dataclass
class Data:
    data : list
    def __iter__(self):
        return DataWithState(self)

@dataclass
class DataWithState:
    data  : Data
    state : int = 0
    def __iter__(self): return self
    def __next__(self):
        try:
            rv = self.data[self.index]
            self.index += 1
        except IndexError as e:
            raise StopIteration() from e
        else:
            return rv

x = Data([*range(3)])
xi = iter(x)

print(f'{isinstance(x, Iterable)  = }')
print(f'{isinstance(xi, Iterator) = }')

from contextlib import nullcontext

with nullcontext() as ctxvar:
    print(f'{ctxvar = }')

mgr = nullcontext()
try:
    ctxvar = mgr.__enter__()
    print(f'{ctxvar = }')
except BaseException as e:
    mgr.__exit__(type(e), e, e.__traceback__)
else:
    mgr.__exit__(None, None, None)

class T:
    def __enter__(self):
        pass
    def __exit__(self, exc_type, exc_value, traceback):
        pass

with T() as _:
    pass

class T:
    @staticmethod
    def __enter__():
        pass
    @staticmethod
    def __exit__(exc_type, exc_value, traceback):
        pass

with T() as _:
    pass

class T:
    def __call__(self):
        return self
    @staticmethod
    def __enter__():
        pass
    @staticmethod
    def __exit__(exc_type, exc_value, traceback):
        pass

inst = T()
with inst() as _:
    pass

class T:
    __enter__ = staticmethod(lambda: print('__enter__'))
    __exit__ = staticmethod(lambda *_: print('__exit__'))

with T():
    print('inside')

def g():
    print('before')
    yield
    print('after')

gi = g()
next(gi)
print('…')
next(gi, None)

from dataclasses import dataclass
from collections.abc import Generator

def g():
    print('before')
    yield
    print('after')

@dataclass
class contextmanager:
    g : Generator
    def __call__(self, *args, **kwargs):
        self.gi = self.g(*args, **kwargs)
        return self
    def __enter__(self):
        next(self.gi)
    def __exit__(self, exc_type, exc_value, traceback):
        if exc_value is not None:
            self.gi.throw(exc_value)
        else:
            next(self.gi, None)

with contextmanager(g)():
    print('inside')

from tempfile import TemporaryDirectory, NamedTemporaryFile
from sqlite3 import connect
from pathlib import Path
from collections import namedtuple
from random import randint
from contextlib import contextmanager

class Row(namedtuple('RowBase', 'a b')):
    @classmethod
    def from_random(cls):
        return cls(a=randint(-10, +10), b=randint(-10, +10))

@contextmanager
def temp_table(con, *, prefix=''):
    cur = con.cursor()
    table = f'{prefix}test'
    cur.execute(f'create table {table} (a int, b int)')
    cur.executemany('insert into test values (?, ?)', [
        Row.from_random() for _ in range(10)
    ])
    try:
        yield table
    finally:
        cur.execute(f'drop table {table}')

with TemporaryDirectory(prefix='test.') as d:
    d = Path(d)
    with NamedTemporaryFile(mode='w', dir=d) as f:
        p = Path(f.name).with_suffix('.db')
        with connect(p) as con:
            with temp_table(con) as table:
                cur = con.cursor()
                for res in cur.execute(f'select a, avg(b) from {table} group by a'):
                    print(f'{res = }')

from contextlib import contextmanager

@contextmanager
def context0():
    print('context0: before')
    yield
    print('context0: after')

@contextmanager
def context1(context):
    with context() as ctx:
        print('context1: before')
        yield ctx
        print('context1: after')

with context1(context0) as ctx:
    print(f'{ctx = }')

from contextlib import contextmanager

@contextmanager
def context0():
    print('context0: before')
    yield
    print('context0: after')

@contextmanager
def context1():
    print('context1: before')
    yield
    print('context1: after')

with context0() as ctx1:
    with context1() as ctx0:
        print(f'{ctx0, ctx1 = }')

What Context Managers?

PEP 492: Coroutines with async and await syntax

from contextlib import contextmanager, asynccontextmanager
from asyncio import run

@contextmanager
def context(): yield

@asynccontextmanager
async def asynccontext(): yield

async def main():
    with context() as ctx:
        async with asynccontext() as actx:
            print(f'{ctx, actx = }')

run(main())

from time import sleep as time_sleep
from asyncio import sleep as asyncio_sleep

from contextlib import contextmanager
from time import perf_counter
from asyncio import run, gather

@contextmanager
def debug(msg):
    print(f'{msg} {perf_counter():.0f}')
    yield
    print(f'{msg} {perf_counter():.0f}')

def f():
    time_sleep(1)

async def t():
    await asyncio_sleep(1)

async def task(name):
    with debug(name):
        f()
        await t()

async def main():
    await gather(task('task0'), task('task1'))

run(main())

from asyncio import run

def f(): g()
def g(): h()
def h(): pass

async def task():
    f()

async def main():
    await task()

run(main())

def blocking(): pass
async def nonblocking(): pass

def sync():
    blocking()
    #  await nonblocking()

async def async_():
    blocking()
    await nonblocking()

# `async def` → `def`       ✓
# `async def` → `async def` ✓
# `def`       → `def`       ✓
# `def`       → `async def` ❌

from contextlib import contextmanager
from asyncio import run, sleep

@contextmanager
def context():
    print('before')
    yield
    print('after')

async def task():
    await sleep(1)

async def main():
    with context():
        await task()
    #  with context():
    #      t = task()
    #  await t

run(main())

from asyncio import run, sleep
from contextlib import asynccontextmanager

@asynccontextmanager
async def context():
    print('before')
    await sleep(.1)
    yield
    await sleep(.1)
    print('after')

async def task():
    async with context():
        await sleep(1)

async def main():
    await task()

run(main())

from contextlib import asynccontextmanager
from asyncio import run

@asynccontextmanager
async def context0():
    print('context0: before')
    yield
    print('context0: after')

@asynccontextmanager
async def context1(context):
    async with context() as ctx:
        print('context1: before')
        yield ctx
        print('context1: after')

async def main():
    async with context1(context0) as ctx:
        print(f'{ctx = }')

run(main())

from contextlib import asynccontextmanager, contextmanager
from asyncio import run

@contextmanager
def context0():
    print('context0: before')
    yield
    print('context0: after')

@asynccontextmanager
async def context1(context):
    with context() as ctx:
        print('context1: before')
        yield ctx
        print('context1: after')

async def main():
    async with context1(context0) as ctx:
        print(f'{ctx = }')

run(main())

from asyncio import run

class T:
    async def __aenter__(self):
        print('__aenter__')
    async def __aexit__(self, *_):
        print('__aexit__')

async def main():
    async with T():
        pass

run(main())

PEP 567: Context Variables

from locale import setlocale, LC_ALL
from datetime import datetime

setlocale(LC_ALL, 'de_DE.UTF-8')
print(
    f'{datetime.now():%A %B %d, %Y}',
    f'{123_456.789:,.2f}',
    sep='\n'
)

from locale import getlocale, setlocale, LC_ALL
from datetime import datetime

oldlocale = getlocale()
setlocale(LC_ALL, 'de_DE.UTF-8')
print(
    f'{datetime.now():%A %B %d, %Y}',
    f'{123_456.789:,.2f}',
    sep='\n'
)
setlocale(LC_ALL, oldlocale)

from locale import getlocale, setlocale, LC_ALL
from datetime import datetime
from contextlib import contextmanager

@contextmanager
def localecontext(locale):
    oldlocale = getlocale()
    setlocale(LC_ALL, locale)
    try:
        yield
    finally:
        setlocale(LC_ALL, oldlocale)
        # NOTE: what if oldlocale has
        #       different values
        #       for different categories?

with localecontext('de_DE.UTF-8'):
    print(
        f'{datetime.now():%A %B %d, %Y}',
        f'{123_456.789:,.2f}',
        sep='\n'
    )

from decimal import localcontext, Decimal

print(f'{Decimal("1") / Decimal("7") = }')

with localcontext() as ctx:
    ctx.prec = 3
    print(f'{Decimal("1") / Decimal("7") = }')

print(f'{Decimal("1") / Decimal("7") = }')

from contextlib import contextmanager
from threading import Thread
from random import random
from time import sleep

state = None
@contextmanager
def context(new_state):
    global state
    old_state = state
    state = new_state
    yield
    state = old_state

def target(name):
    print(f'before {name = } {state = }')
    with context(random()):
        sleep(random())
        print(f'inside {name = } {state = }')
    print(f'after  {name = } {state = }')

pool = [
    Thread(target=target, kwargs={'name': 'thread#1'}),
    Thread(target=target, kwargs={'name': 'thread#2'}),
]
print(f'initial {state = }')
for x in pool: x.start()
for x in pool: x.join()
print(f'final {state = }')

from contextlib import contextmanager
from threading import Thread, local
from random import random
from time import sleep

var = local()
@contextmanager
def context(state):
    old_state = getattr(var, 'state', None)
    var.state = state
    yield
    var.state = old_state

def target(name):
    print(f'before {name = } {getattr(var, "state", None) = }')
    with context(random()):
        sleep(random())
        print(f'inside {name = } {var.state = }')
    print(f'after  {name = } {var.state = }')

pool = [
    Thread(target=target, kwargs={'name': 'thread#1'}),
    Thread(target=target, kwargs={'name': 'thread#2'}),
]
for x in pool: x.start()
for x in pool: x.join()

from contextlib import asynccontextmanager
from random import random
from asyncio import run, gather, sleep

@asynccontextmanager
async def context(name):
    await sleep(random())
    print(f'before: context({name = })')
    yield
    print(f'after:  context({name = })')

async def task(name):
    print(f'task({name = })')
    async with context(name):
        await sleep(random())

async def main():
    await gather(task('task0'), task('task1'))

run(main())

from contextlib import asynccontextmanager
from contextvars import ContextVar
from random import random
from asyncio import run, gather, sleep

var = ContextVar('var', default=None)
@asynccontextmanager
async def context(name):
    await sleep(random())
    print(f'before: context({name = }) {var.get() = }')
    old = var.get()
    var.set(random())
    yield
    var.set(old)
    print(f'after:  context({name = }) {var.get() = }')

async def task(name):
    print(f'task({name = })')
    async with context(name):
        print(f'inside: {var.get() = }')
        await sleep(random())

async def main():
    await gather(task('task0'), task('task1'))

run(main())

What else?

class T:
    def __init__(self):
        print('__init__')
    def __del__(self):
        print('__del__')

def f():
    obj = T()
    print('middle')

print('before')
f()
print('after')

from dataclasses import dataclass

@dataclass
class GlobalState:
    obj : object = None

class T:
    def __init__(self):
        print('__init__')
    def __del__(self):
        print('__del__')

gs = GlobalState()
def f():
    obj = T()
    gs.obj = obj

print('before')
f()
print('after')

from dataclasses import dataclass
from gc import collect

@dataclass
class T:
    obj : object = None
    def __del__(self):
        print('__del__')

def f():
    obj0, obj1 = T(), T()
    obj0.obj, obj1.obj = obj1, obj0
    del obj0, obj1

f()
#  collect()
for x in range(10):
    print(f'{x = }')

x = 123
del x

d = {'key': 'value'}
del d['key']

class T: pass
obj = T()
obj.attr = ...
del obj.attr

class T:
    def __delitem__(self, key): pass
    def __delattr__(self, key): pass

from pandas import DataFrame

df = DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
print(
    df,
)
#  del df['a']
#  df = df.drop('a', axis='columns')
print(
    df,
)

from weakref import ref

class T: pass

def callback(ref):
    print(f'callback({ref})')

obj = T()
r = ref(obj, callback)
print(f'{r()            = }')
print(f'{r.__callback__ = }')
del obj
print(f'{r()            = }')

from weakref import finalize

class T: pass

def func(*a, **kw):
    params = *(f'{x}' for x in a), *(f'{k}={v}' for k, v in kw.items())
    print(f'func({", ".join(params)})')

print('before')
obj = T()
fin = finalize(obj, func, 1, 2, 3, a='aaa', b='bbb', c='ccc')
#  fin()
print('after')

class T:
    def __enter__(self):
        pass

    def __exit__(self, *_):
        0 / 0
        return True

with T():
    None.upper()

def f():
    x = 123

class T:
    x = 123

try:
    ...
except Exception as e:
    e

for x in range(10):
    ...
x

from contextlib import contextmanager

@contextmanager
def ctx():
    yield [1, 2, 3]


with ctx() as x:
    pass
    del x
#  print(f'{x = }')