Decorators bieten eine neue und bequeme Möglichkeit für alles, vom Caching bis zum Senden von Benachrichtigungen.
Zunächst ist es das Ziel eines jeden Entwicklers, die Dinge zum Laufen zu bringen. Langsam machen wir uns Gedanken über Lesbarkeit und Skalierbarkeit. Das ist der Zeitpunkt, an dem wir anfangen, über Dekoratoren nachzudenken.
Dekoratoren sind eine hervorragende Möglichkeit, einer Funktion zusätzliches Verhalten zu verleihen. Und es gibt kleine Dinge, die wir Datenwissenschaftler oft in eine Funktionsdefinition einbauen müssen.
Sie werden überrascht sein, wie sehr Sie mit Dekoratoren die Wiederholung von Code reduzieren und die Lesbarkeit verbessern können. Ich habe es jedenfalls getan.
Hier sind die fünf häufigsten, die ich bei fast jedem datenintensiven Projekt verwende.
1. Der Wiederholungsdekorator
In Data-Science-Projekten und Softwareentwicklungsprojekten gibt es so viele Fälle, in denen wir von externen Systemen abhängig sind. Wir haben nicht immer die Kontrolle über die Dinge
Wenn ein unerwartetes Ereignis eintritt, möchten wir vielleicht, dass unser Code eine Weile wartet, damit das externe System sich selbst korrigieren und den Vorgang wiederholen kann.
Ich ziehe es vor, diese Wiederholungslogik in einem Python-Dekorator zu implementieren, so dass ich jede Funktion mit Anmerkungen versehen kann, um das Wiederholungsverhalten anzuwenden.
Hier ist der Code für einen Retry-Dekorator.
import time
from functools import wraps
def retry(max_tries=3, delay_seconds=1):
def decorator_retry(func):
@wraps(func)
def wrapper_retry(*args, **kwargs):
tries = 0
while tries < max_tries:
try:
return func(*args, **kwargs)
except Exception as e:
tries += 1
if tries == max_tries:
raise e
time.sleep(delay_seconds)
return wrapper_retry
return decorator_retry
@retry(max_tries=5, delay_seconds=2)
def call_dummy_api():
response = requests.get("https://jsonplaceholder.typicode.com/todos/1")
return responseIn dem obigen Code versuchen wir, eine API-Antwort zu erhalten. Wenn dies fehlschlägt, versuchen wir dieselbe Aufgabe 5 Mal erneut. Zwischen jedem erneuten Versuch warten wir 2 Sekunden lang.
2. Zwischenspeicherung von Funktionsergebnissen
Einige Teile unserer Codebasis ändern ihr Verhalten nur selten. Dennoch kann dies einen großen Teil unserer Rechenleistung in Anspruch nehmen. In solchen Situationen können wir einen Dekorator verwenden, um Funktionsaufrufe zwischenzuspeichern.
Die Funktion wird nur einmal ausgeführt, wenn die Eingaben identisch sind. Bei jedem weiteren Durchlauf werden die Ergebnisse aus dem Zwischenspeicher geholt. Daher müssen wir nicht ständig teure Berechnungen durchführen.
def memoize(func):
cache = {}
def wrapper(*args):
if args in cache:
return cache[args]
else:
result = func(*args)
cache[args] = result
return result
return wrapperDer Decorator verwendet ein Wörterbuch, speichert die Funktions-Args und gibt Werte zurück. Wenn wir diese Funktion ausführen, prüft der Dekorator das Wörterbuch auf vorherige Ergebnisse. Die eigentliche Funktion wird nur aufgerufen, wenn vorher kein Wert gespeichert wurde.
Das folgende Beispiel ist eine Fibonacci-Zahl, die eine Funktion berechnet. Da es sich um eine wiederkehrende Funktion handelt, wird dieselbe Funktion mehrmals aufgerufen. Mit Caching können wir diesen Prozess jedoch beschleunigen.
@memoize
def fibonacci(n):
if n <= 1:
return n
else:
return fibonacci(n-1) + fibonacci(n-2)
Hier sind die Ausführungszeiten für diese Funktion mit und ohne Caching. Beachten Sie, dass die zwischengespeicherte Version nur den Bruchteil einer Millisekunde zur Ausführung benötigt, während die nicht zwischengespeicherte Version fast eine Minute benötigte.
Function slow_fibonacci took 53.05560088157654 seconds to run.
Function fast_fibonacci took 7.772445678710938e-05 seconds to run.Die Verwendung eines Wörterbuchs zur Speicherung von Daten aus früheren Ausführungen ist ein einfacher Ansatz. Es gibt jedoch eine ausgefeiltere Methode zum Speichern von Caching-Daten. Sie können eine In-Memory-Datenbank, wie Redis, verwenden.
3. Timing-Funktionen
Dieser Punkt ist keine Überraschung. Wenn wir mit datenintensiven Funktionen arbeiten, wollen wir unbedingt wissen, wie lange die Ausführung dauert.
Normalerweise werden dazu zwei Zeitstempel gesammelt, einer am Anfang und einer am Ende der Funktion. Wir können dann die Dauer berechnen und sie zusammen mit den Rückgabewerten ausgeben.
Aber dies immer wieder für mehrere Funktionen zu tun, ist sehr mühsam.
Stattdessen können wir das von einem Dekorator erledigen lassen. Wir können jede Funktion, die eine Dauer ausgeben muss, mit einem Kommentar versehen.
Hier ist ein Beispiel für einen Python-Dekorator, der die Laufzeit einer Funktion ausgibt, wenn sie aufgerufen wird:
import time
def timing_decorator(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"Function {func.__name__} took {end_time - start_time} seconds to run.")
return result
return wrapperSie können diesen Dekorator verwenden, um die Ausführung einer Funktion zu verzögern:
@timing_decorator
def my_function():
# some code here
time.sleep(1) # simulate some time-consuming operation
returnDer Aufruf der Funktion würde die Zeit ausgeben, die sie zur Ausführung benötigt.
my_function()
>>> Function my_function took 1.0019128322601318 seconds to run.4. Funktionsaufrufe protokollieren
Dieser Dekorator ist eine Erweiterung des vorherigen Dekorators. Aber er hat einige besondere Anwendungen.
Wenn Sie die Prinzipien des Software-Designs befolgen, werden Sie das Prinzip der einzigen Verantwortung zu schätzen wissen. Dies bedeutet im Wesentlichen, dass jede Funktion eine und nur eine Verantwortung hat.
Wenn Sie Ihren Code auf diese Weise gestalten, möchten Sie auch die Ausführungsinformationen Ihrer Funktionen protokollieren. Hier kommen die Logging-Dekoratoren ins Spiel.
Das folgende Beispiel veranschaulicht dies.
import logging
import functools
logging.basicConfig(level=logging.INFO)
def log_execution(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
logging.info(f"Executing {func.__name__}")
result = func(*args, **kwargs)
logging.info(f"Finished executing {func.__name__}")
return result
return wrapper
@log_execution
def extract_data(source):
# extract data from source
data = ...
return data
@log_execution
def transform_data(data):
# transform data
transformed_data = ...
return transformed_data
@log_execution
def load_data(data, target):
# load data into target
...
def main():
# extract data
data = extract_data(source)
# transform data
transformed_data = transform_data(data)
# load data
load_data(transformed_data, target)Der obige Code ist eine vereinfachte Version einer ETL-Pipeline. Wir haben drei separate Funktionen, um jedes Extrahieren, Transformieren und Laden zu behandeln. Wir haben jede von ihnen mit unserem log_execution Dekorator umhüllt.
Wenn der Code nun ausgeführt wird, sehen Sie eine Ausgabe ähnlich der folgenden:
INFO:root:Executing extract_data
INFO:root:Finished executing extract_data
INFO:root:Executing transform_data
INFO:root:Finished executing transform_data
INFO:root:Executing load_data
INFO:root:Finished executing load_dataWir könnten auch die Ausführungszeit innerhalb dieses Dekorators drucken lassen. Aber ich würde sie gerne beide in separaten Dekoratoren haben. Auf diese Weise kann ich wählen, welchen (oder beide) ich für eine Funktion verwenden möchte.
Hier ist, wie man mehrere Dekoratoren für eine einzige Funktion verwendet.
@log_execution
@timing_decorator
def my_function(x, y):
time.sleep(1)
return x + y
5. Benachrichtigungsdekorator
Schließlich ist ein sehr nützlicher Dekorator in Produktionssystemen der Benachrichtigungsdekorator.
Auch hier gilt, dass selbst bei mehreren Wiederholungsversuchen selbst eine gut getestete Codebasis versagt. Und wenn das passiert, müssen wir jemanden darüber informieren, um schnell handeln zu können.
Das ist nicht neu, wenn Sie jemals eine Datenpipeline gebaut haben und gehofft haben, dass sie für immer gut funktioniert.
Der folgende Decorator sendet eine E‑Mail, wenn die Ausführung der inneren Funktion fehlschlägt. Es muss in Ihrem Fall nicht unbedingt eine E‑Mail-Benachrichtigung sein. Sie können ihn so konfigurieren, dass er eine Teams/Slack-Benachrichtigung sendet.
import smtplib
import traceback
from email.mime.text import MIMEText
def email_on_failure(sender_email, password, recipient_email):
def decorator(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
# format the error message and traceback
err_msg = f"Error: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
# create the email message
message = MIMEText(err_msg)
message['Subject'] = f"{func.__name__} failed"
message['From'] = sender_email
message['To'] = recipient_email
# send the email
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
smtp.login(sender_email, password)
smtp.sendmail(sender_email, recipient_email, message.as_string())
# re-raise the exception
raise
return wrapper
return decorator
@email_on_failure(sender_email='your_email@gmail.com', password='your_password', recipient_email='recipient_email@gmail.com')
def my_function():
# code that might fail
Fazit
Decorators sind eine sehr bequeme Möglichkeit, unseren Funktionen ein neues Verhalten zu verleihen. Ohne sie gibt es eine Menge Code-Wiederholungen.
In diesem Beitrag habe ich meine am häufigsten verwendeten Dekoratoren besprochen. Sie können diese für Ihre speziellen Bedürfnisse erweitern. Zum Beispiel können Sie einen Redis-Server verwenden, um Cache-Antworten anstelle von Wörterbüchern zu speichern. Dadurch erhalten Sie mehr Kontrolle über die Daten, z. B. über die Persistenz. Oder Sie könnten den Code so anpassen, dass die Wartezeit im Retry-Dekorator schrittweise erhöht wird.
In allen meinen Projekten verwende ich eine Version dieser Dekoratoren. Obwohl sich ihr Verhalten leicht unterscheidet, sind dies die gemeinsamen Ziele, für die ich Dekoratoren häufig verwende.
Ich hoffe, dieser Beitrag hilft Ihnen.
Quelle: medium
Erfahren Sie mehr über Lösungen im Bereich Data Management oder besuchen Sie eines unserer kostenlosen Webinare.
