Tips and tricks from my Telegram-channel @pythonetc, July 2019

It is a new selection of tips and tricks about Python and programming from my Telegram-channel @pythonetc.

← Previous publications


You can’t mutate closure variables by simply assigning them. Python treats assignment as a definition inside a function body and doesn’t make closure at all.

Works fine, prints 2:

def make_closure(x):
    def closure():
        print(x)

    return closure

make_closure(2)

Throws UnboundLocalError: local variable 'x' referenced before assignment:

def make_closure(x):
    def closure():
        print(x)
        x *= 2
        print(x)

    return closure

make_closure(2)()

To make it work you should use nonlocal. It explicitly tells the interpreter not to treat assignment as a definition:

def make_closure(x):
    def closure():
        nonlocal x
        print(x)
        x *= 2
        print(x)

    return closure

make_closure(2)()

Sometimes during iteration you may want to know whether it’s the first or the last element step of the iteration. Simple way to handle this is to use explicit flag:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    first = True
    for x in iterable:
        if not first:
            result += zeros
        result.append(x)

        first = False

    return result

assert sparse_list([1, 2, 3], 2) == [
    1,
    0, 0,
    2,
    0, 0,
    3,
]

You also could process the first element outside of the loop, that may seem more clear but leads to code duplication to the certain extent. It is also not a simple thing to do while working with abstract iterables:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    iterator = iter(iterable)
    try:
        result.append(next(iterator))
    except StopIteration:
        return []

    for x in iterator:
       result += zeros
       result.append(x)

    return result

You also could use enumerate and check for the i == 0 (works only for the detection of the first element, not the last one), but the ultimate solution might be a generator that returns first and last flags along with the element of an iterable:

def first_last_iter(iterable):
    iterator = iter(iterable)

    first = True
    last = False
    while not last:
    if first:
        try:
            current = next(iterator)
            except StopIteration:
                return
    else:
        current = next_one

    try:
        next_one = next(iterator)
    except StopIteration:
        last = True

    yield (first, last, current)

    first = False

The initial function now may look like this:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    for first, last, x in first_last_iter(iterable):
        if not first:
            result += zeros
        result.append(x)

    return result

If you want to measure time between two events you should use time.monotonic() instead of time.time(). time.monotonic() never goes backwards even if system clock is updated:

from contextlib import contextmanager
import time


@contextmanager
def timeit():
    start = time.monotonic()
    yield
    print(time.monotonic() - start)

def main():
    with timeit():
           time.sleep(2)

main()

Nested context managers normally don’t know that they are nested. You can make them know by spawning inner context managers by the outer one:

from contextlib import AbstractContextManager
import time


class TimeItContextManager(AbstractContextManager):
    def __init__(self, name, parent=None):
        super().__init__()

        self._name = name
        self._parent = parent
        self._start = None
        self._substracted = 0

    def __enter__(self):
        self._start = time.monotonic()
        return self
        
    def __exit__(self, exc_type, exc_value, traceback):
        delta = time.monotonic() - self._start
        if self._parent is not None:
            self._parent.substract(delta)

    print(self._name, 'total', delta)
    print(self._name, 'outer', delta - self._substracted)

    return False

    def child(self, name):
        return type(self)(name, parent=self)

    def substract(self, n):
        self._substracted += n


timeit = TimeItContextManager


def main():
    with timeit('large') as large_t:
        with large_t.child('medium') as medium_t:
            with medium_t.child('small-1'):
                time.sleep(1)
            with medium_t.child('small-2'):
                time.sleep(1)
        time.sleep(1)
    time.sleep(1)


main()

If you want to pass some information down the call chain, you usually use the most straightforward way possible: you pass it as functions arguments.

However, in some cases, it may be highly inconvenient to modify all functions in the chain to propagate some new piece of data. Instead, you may want to set up some kind of context to be used by all functions down the chain. How can this context be technically done?

The simplest solution is a global variable. In Python, you also may use modules and classes as context holders since they are, strictly speaking, global variables too. You probably do it on a daily basis for things like loggers.

If your application is multi-threaded, a bare global variable won't work for you since they are not thread-safe. You may have more than one call chain running at the same time, and each of them needs its own context. The threading module gets you covered, it provides the threading.local() object that is thread-safe. Store there any data by simply accessing attributes: threading.local().symbol = '@'.

Still, both of that approaches are concurrency-unsafe meaning they won't work for coroutine call-chain where functions are not only called but can be awaited too. Once a coroutine does await, an event loop may run a completely different coroutine from a completely different chain. That won't work:

import asyncio
import sys

global_symbol = '.'

async def indication(timeout):
    while True:
        print(global_symbol, end='')
        sys.stdout.flush()
        await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
    loop = asyncio.get_event_loop()

    global global_symbol
    global_symbol = symbol
    task = loop.create_task(
            indication(indication_t)
    )
    await asyncio.sleep(t)
    task.cancel()

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    sleep(1, 0.1, '0'),
    sleep(1, 0.1, 'a'),
    sleep(1, 0.1, 'b'),
    sleep(1, 0.1, 'c'),
))

You can fix that by having the loop set and restore the context every time it switches between coroutines. You can do it with the contextvars module since Python 3.7.

import asyncio
import sys
import contextvars

global_symbol = contextvars.ContextVar('symbol')

async def indication(timeout):
    while True:
        print(global_symbol.get(), end='')
        sys.stdout.flush()
        await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
    loop = asyncio.get_event_loop()

    global_symbol.set(symbol)
    task = loop.create_task(indication(indication_t))
    await asyncio.sleep(t)
    task.cancel()

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    sleep(1, 0.1, '0'),
    sleep(1, 0.1, 'a'),
    sleep(1, 0.1, 'b'),
    sleep(1, 0.1, 'c'),
))