company_banner

Подборка @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'),
    ))
    
    • +37
    • 6,2k
    • 9
    Mail.ru Group
    1 016,66
    Строим Интернет
    Поделиться публикацией

    Похожие публикации

    Комментарии 9

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

      if not first:

      не особо оптимально каждый раз в цикле спрашивать не первый ли элемент. Лучше уже выделить действие цикла в функцию. Тут надо подумать, завтра если что то получше придумаю напишу
        –1
        так вроде все проще можно сделать
        список без первого и последнего
         for x in list[1:-1]

        первый
         list[:(len(list)-1)*-1]

        последний
         list[-1:]
          0
          Это работает только для списков, а не для любых итерируемых объектов.
            –2
            Любой итерируемый объект можно превратить в список)
              0
              ага, особенно генератор на много значений, вперёд ))
        0
        Подскажите пожалуйста, что означает конструкция: «make_closure(2)()». Конкретно, что делают вторые пустые скобки?
          0
          Вызывают результат выполнения make_closure(2). Т. е. это примерно то же самое, что и:

          closure = make_closure(2)
          closure()
            0
            Спасибо.
          0
          Если ваше приложение многопоточное, то обычные глобальные переменные вам не подойдут, поскольку они не потокобезопасны.
          Откуда инфа?

          Только полноправные пользователи могут оставлять комментарии. Войдите, пожалуйста.

          Самое читаемое