Подборка @pythonetc, июль 2019

Это двенадцатая подборка советов про Python и программирование из моего авторского канала @pythonetc.

← Предыдущие подборки


Нельзя изменять переменные замыканий с помощью простого присваивания. Python расценивает присваивание как определение внутри тела функции и вообще не делает замыкание. 

Работает отлично, выводит на экран 2:

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

    return closure

make_closure(2)()

А этот код бросает UnboundLocalError: local variable 'x' referenced before assignment:

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

    return closure

make_closure(2)()

Чтобы код работал, используйте nonlocal. Это явным образом говорит интерпретатору не рассматривать присвоение как определение:

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

    return closure

make_closure(2)()

Иногда в ходе итерации вам нужно узнать, какой это элемент обрабатывается, первый или последний. Это можно легко выяснить с помощью явного флага:

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,
]

Конечно, вы могли бы обрабатывать первый элемент за пределами цикла. Это выглядит чище, но приводит к частичному дублированию кода. Кроме того, сделать это будет не так просто при работе с абстрактным iterable:

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

Ещё вы можете использовать enumerate и выполнять проверку i == 0 (работает только для определения первого элемента, а не последнего), однако наилучшим решением будет генератор, возвращающий вместе с элементом iterable флаги first и last:

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

Теперь исходная функция может выглядеть так:

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

Если вам нужно измерить время, прошедшее между двумя событиями, то используйте time.monotonic() вместо time.time(). time.monotonic() никогда не изменяется в меньшую сторону, даже при обновлении системных часов:

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()

Вложенные менеджеры контекста обычно не знают, что они вложены. Вы можете сообщить им об этом, создавая внутренние менеджеры с помощью внешнего:

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()

Когда вам нужно передать информацию по цепочке вызовов, то первое, что приходит в голову, это передавать данные в виде аргументов функций.

В некоторых случаях может быть гораздо удобнее модифицировать все функции в цепочке для передачи новой порции данных. Вместо этого вы можете задать некий контекст, который будет использоваться всеми функциями в цепочке. Как это сделать?

Самое простое решение — использовать глобальную переменную. В Python вы также можете использовать модули и классы в качестве хранителей контекста, потому что они, строго говоря, тоже являются глобальными переменными. Вероятно, вы и так уже это регулярно делаете, например, для журналирования.

Если ваше приложение многопоточное, то обычные глобальные переменные вам не подойдут, поскольку они не потокобезопасны. В каждый момент времени у вас может выполняться несколько цепочек вызовов, и каждой из них нужен собственный контекст. Вам поможет модуль threading, он предоставляет объект threading.local(), который потокобезопасен. Хранить в нём данные можно с помощью простого обращения к атрибутам: threading.local().symbol = '@'.

Тем не менее, оба описанных подхода не concurrency-safe, то есть они не подходят для цепочки вызовов корутин, в которой система не только вызывает функции, но и ожидает их исполнения. Когда корутина выполняет await, поток событий может запустить другую корутину из другой цепочки. Это не будет работать:

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'),
))

Исправить это можно, заставив цикл задавать и восстанавливать контекст при каждом переключении между корутинами. Реализовать такое поведение можно с помощью модуля contextvars, который доступен начиная с 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'),
))