It is a new selection of tips and tricks about Python and programming from my Telegram-channel @pythonetc.
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 = FalseThe 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'),
))
Автор: pushtaev
