Хабрахабр

Подборка @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.local().symbol = '@'. Вам поможет модуль threading, он предоставляет объект threading.local(), который потокобезопасен.

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

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

Теги
Показать больше

Похожие статьи

Добавить комментарий

Ваш e-mail не будет опубликован. Обязательные поля помечены *

Кнопка «Наверх»
Закрыть