它是我的Telegram频道@pythonetc中有关Python和编程的一些新技巧和窍门。
←
以前的出版物您不能仅通过分配闭包变量来对其进行变异。 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, ]
您还可以在循环外处理第一个元素,这看起来似乎更清晰,但在一定程度上导致代码重复。 在处理抽象可迭代对象时,这也不是一件容易的事:
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
(仅用于检测第一个元素,而不是最后一个元素),但是最终的解决方案可能是生成器,该生成器返回
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()
对象。 只需访问以下属性即可在其中存储任何数据:
threading.local().symbol = '@'
。
尽管如此,这两种方法都是并发不安全的,这意味着它们不能用于协程调用链,在协程调用链中,不仅要调用函数,还可以等待函数。 一旦协程确实
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'), ))
您可以通过设置循环并在每次在协程之间切换时恢复上下文来解决此问题。 从Python 3.7开始,您可以使用
contextvars
模块来实现。
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'), ))