@ Pythonetc-Zusammenstellung, Juli 2019


Dies ist die zwölfte Sammlung von Python-Tipps und -Programmierungen aus meinem @ pythonetc-Feed.

Frühere Sammlungen


Sie können Abschlussvariablen nicht mit einer einfachen Zuordnung ändern. Python betrachtet die Zuweisung als Definition innerhalb eines Funktionskörpers und schließt überhaupt nicht.

Funktioniert gut, zeigt 2 :

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

Und dieser Code UnboundLocalError: local variable 'x' referenced before assignment einen UnboundLocalError: local variable 'x' referenced before assignment :

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


Verwenden Sie nonlocal damit der Code funktioniert. Dies weist den Interpreter ausdrücklich an, die Zuweisung nicht als Definition zu betrachten:

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


Manchmal müssen Sie während einer Iteration zuerst oder zuletzt herausfinden, welches Element verarbeitet wird. Dies kann leicht mit einem expliziten Flag ermittelt werden:

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

Natürlich können Sie das erste Element außerhalb der Schleife behandeln. Dies sieht sauberer aus, führt jedoch zu einer teilweisen Codeduplizierung. Darüber hinaus wird es nicht so einfach sein, dies zu tun, wenn Sie mit abstrakt 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 

Sie können auch enumerate und eine i == 0 Prüfung durchführen (dies dient nur zur Bestimmung des ersten und nicht des letzten Elements). Die beste Lösung wäre jedoch ein Generator, der das first und das last Flag mit dem iterable Element zurückgibt:

 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 

Jetzt könnte die ursprüngliche Funktion folgendermaßen aussehen:

 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 


Wenn Sie die zwischen zwei Ereignissen verstrichene Zeit messen müssen, verwenden Sie time.monotonic() anstelle von time.time() . time.monotonic() ändert sich niemals in die kleinere Richtung, auch wenn die Systemuhr aktualisiert wird:

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


Verschachtelte Kontextmanager wissen normalerweise nicht, dass sie verschachtelt sind. Sie können ihnen dies mitteilen, indem Sie interne Manager mithilfe eines externen Managers erstellen:

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


Wenn Sie Informationen zu einer Kette von Aufrufen übergeben müssen, müssen Sie zunächst Daten in Form von Funktionsargumenten übergeben.

In einigen Fällen kann es viel bequemer sein, alle Funktionen in der Kette zu ändern, um neue Daten zu übertragen. Stattdessen können Sie einen Kontext angeben, der von allen Funktionen in der Kette verwendet wird. Wie kann man das machen?

Die einfachste Lösung ist die Verwendung einer globalen Variablen. In Python können Sie auch Module und Klassen als Kontextbewahrer verwenden, da sie streng genommen auch globale Variablen sind. Sie tun dies wahrscheinlich bereits regelmäßig, zum Beispiel für das Journaling.

Wenn Ihre Anwendung Multithread-fähig ist, funktionieren normale globale Variablen für Sie nicht, da sie nicht threadsicher sind. Zu jedem Zeitpunkt können Sie mehrere Anrufketten haben, von denen jede einen eigenen Kontext benötigt. Das threading Modul hilft Ihnen dabei. Es stellt ein threading.local() -Objekt bereit, das threadsicher ist. Sie können Daten darin speichern, indem Sie einfach auf die Attribute zugreifen: threading.local().symbol = '@' .

Beide beschriebenen Ansätze sind jedoch nicht parallelitätssicher, dh sie sind nicht für die Coroutine-Aufrufkette geeignet, in der das System Funktionen nicht nur aufruft, sondern auch erwartet, dass sie ausgeführt werden. Wenn eine Coroutine auf Warten await , kann ein Ereignisfluss eine andere Coroutine aus einer anderen Kette auslösen. Dies wird nicht funktionieren:

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

Sie können dies beheben, indem Sie den Zyklus zwingen, den Kontext bei jedem Wechsel zwischen Coroutinen festzulegen und wiederherzustellen. Sie können dieses Verhalten mit dem contextvars Modul contextvars , das seit Python 3.7 verfügbar ist.

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

Source: https://habr.com/ru/post/de462311/


All Articles