In den Python-Kursen, die ich gebe, spreche ich immer über Generatoren und Dekoratoren. Die Studenten, die oft aus der (nicht angewandten) Datenwissenschaft oder Datenanalyse kommen, haben Schwierigkeiten zu verstehen, warum diese beiden Konzepte für ihre tägliche Arbeit wichtig sind.
Ich spreche natürlich darüber, wie Generatoren Ihren Geist befreien, indem sie Sie von Annahmen über die Verwendung und den Rückgabetyp befreien, d.h.:
- Wenn Ihr Ziel darin besteht, eine längere Sequenz zurückzugeben, können Sie mit Generatoren entscheiden, wie viel von dieser Sequenz produziert wird (Vermutung über die Verwendung);
- Der Aufrufer kann entscheiden, ob er die Ergebnisse in einer Liste, einer Menge oder einem Tupel speichern möchte (Vermutung über den Rückgabetyp).
Die Beispiele, die ich zeigte, waren jedoch immer auf mathematische Berechnungen ausgerichtet und weckten nie wirklich die Aufmerksamkeit der Schüler... und zwangsläufig griffen die Schüler zu Numpy, um eine schnellere Version zu erstellen. Schlechte Lehrerin!
Datenwissenschaftler leiden ein wenig unter dem gleichen Schicksal: Sie wissen (hoffentlich) dank Bibliotheken wie Flask, wie man sie einsetzt, aber es fällt ihnen schwer, ihren Nutzen für die Vereinfachung ihres eigenen Codes zu erkennen.
Letzte Woche stieß ich jedoch auf ein Problem, das sich elegant lösen ließ1 mit beidem gelöst werden konnte.
API, der Segen und Fluch des modernen Datenwissenschaftlers
Immer wenn ich über angewandte Datenwissenschaft spreche, nenne ich APIs als eine neue Datenquelle (oder -senke), die normalerweise nicht von Datenwissenschaftlern/Datenanalysten bearbeitet wird. Der Grund, warum sie so anders sind, ist, dass sie ein neues Paradigma einführen und mit diesem Paradigma eine neue Denkweise.
Diese API, mit der ich zu tun hatte, ist keine Ausnahme. Um Zugang zur API zu erhalten, müssen Sie einen Schlüssel/Geheimnis verwenden, aber nicht direkt: Mit dem Geheimnis können Sie ein Token von einem Token-Dienst erhalten. Der Token ist 30 Minuten lang gültig.
Wenn Sie immer davon ausgehen könnten, dass der Token 30 Minuten lang gültig ist, wäre es einfach, Anfragen an den Token-Dienst zwischenzuspeichern. Wenn jedoch jemand anderes eine Anfrage an den Token-Dienst mit Ihrem Schlüssel/Geheimnis stellt (es könnte ein anderer Benutzer innerhalb Ihrer Organisation sein, wenn der Schlüssel/das Geheimnis gemeinsam genutzt wird, oder es könnte ein anderer Prozess Ihres Programms sein), dann wird die Zeit, die der Token gültig ist, zu etwas zwischen 0 und 30 Minuten.
Glücklicherweise lässt uns dieser spezielle Token-Dienst die Gültigkeit wissen, so dass wir diese Gültigkeit irgendwo speichern können.
Um diesen Zustand irgendwo zu speichern, gibt es zwei Möglichkeiten: Entweder Sie instanziieren eine Klasse für diesen speziellen Zweck und lassen die Klasse die Arbeit machen, oder Sie verwenden einen - Trommelwirbel - Generator!
Der betreffende Code lautet wie folgt:
import requests as r
Authenticator = Iterator[Dict[str, str]]
def get_authenticator(server: str=None, key: str=None, secret: str=None) -> Authenticator:
endpoint = server + "/auth"
header_auth = {"key": key} # 1
body_auth = {"secret": secret} # 1
expire = 0 # 2
while True: # 3
if expire < time.time(): # 4
response_auth = r.post(endpoint,
json=body_auth,
headers=header_auth).json()
access_token = response_auth.get('accessToken') # 1
expire = response_auth.get('expiration')
yield {**header_auth, 'authorization': f"Bearer {access_token}"} # 5, # 1
authenticator = get_authenticator(server, key, secret)
auth_header = next(authenticator) # 3
# do stuff
new_auth_header = next(authenticator) # 6Der Code bedarf einiger Erklärungen:
- Dies ist spezifisch für den Token-Dienst, mit dem ich interagiere;
- Da dieser Generator beim ersten Start die Anfrage stellen muss, muss die
ifin Punkt # 4 passieren.expire = 0ist eine praktische Möglichkeit, dies zu erreichen; - Ich beschließe, dass die Leute mit meinem Generator unbegrenzt neue Token erhalten können. Es macht also Sinn, ihn auf unbestimmte Zeit laufen zu lassen;
- Im Prinzip könnte ich hier eine gewisse Zeitspanne einfügen, z.B.
expire < time.time() + 5, aber ich i) mag keine magischen Zahlen und ii) können immer noch Dinge schief gehen und es könnte sein, dass 5 (oder welche magische Zahl Sie auch immer wählen) nicht ausreicht. Dazu später mehr; - In diesem Fall wird
yieldden Autorisierungs-Header liefern, der hoffentlich immer gültig ist. Übrigens eine schöne neue Python 3.6-Syntax, um neue Wörterbücher zu erstellen. Beachten Sie, dass ich hier das Modultypingverwende, so dass einige Dinge magisch erscheinen könnten, wenn Sie nicht damit vertraut sind. new_auth_headerkönnte identisch mitauth_headersein. Als Anrufer muss und will ich mich nicht darum kümmern.
Das ist schön. Ich habe mit einem leichtgewichtigen Generator gelöst, was sonst mit einer schwerfälligen Klasse gelöst worden wäre.
Wie würde ich also meinen Generator verwenden?
Nun, der erste Schritt wäre, eine Funktion wie diese zu schreiben:
def get_endpoints(server: str, auth: Authenticator):
"Get the available API endpoints"
endpoint = server + "/apis"
response = r.get(endpoint, headers=next(auth)).json()
return response.get('apis')Dies scheint eine nette Funktion zu sein. Sie geht auch von einer perfekten Welt aus (Datenwissenschaftler wie ich tun das oft), in der alle Netzwerkanfragen sofort erfolgen und in der nie Fehler auftreten.
Eines der Dinge, die schief gehen können, ist die Netzwerkanfrage an den Endpunkt apis. Sie könnte zum Beispiel zu lange dauern. In diesem Fall könnte das Token ablaufen oder die gesamte Anfrage fehlschlagen. Lassen Sie uns den ersten Fall hier behandeln und den zweiten auf einen späteren Zeitpunkt verschieben. Eine Möglichkeit, dies zu tun, ist die folgende:
def get_endpoints(server: str, auth: Authenticator, num_retries: int=3):
"Get the available API endpoints"
endpoint = server + "/apis"
for _ in range(num_retries):
response = r.get(endpoint, headers=next(auth)).json()
return response.get('apis')Ok, das ist schon mal gut. Aber was ist, wenn unser Authentifikator den falschen Schlüssel verwendet? Dann erhalten wir nie die richtige Antwort und wissen nicht, womit wir es zu tun haben. Der spezielle Endpunkt, den ich verwende, hat eine Möglichkeit, uns dies mitzuteilen, und zwar in
def get_endpoints(server: str, auth: Authenticator, num_retries: int=3):
"Get the available API endpoints"
endpoint = server + "/apis"
for _ in range(num_retries):
response = r.get(endpoint, headers=next(auth)).json()
error = response.get('error')
if not error:
return response.get('apis')
# this code will only execute if we always get an 'error' after retrying num_retries times code = error.get('code') message = error.get('message') raise AuthenticationFailed(f"http code {code} with message {message}")Übrigens: Was hat es mit AuthenticationFailed auf sich?
Die Klasse AuthenticationFailed ist nicht in reinem Python enthalten. Es handelt sich um eine benutzerdefinierte Klasse, die ich erstellt habe, um meine Fehler in den try/except Anweisungen zu isolieren. Ich ermutige Studenten immer dazu, für ihr Modul/Paket eigene Ausnahmen zu erstellen. Diese benutzerdefinierten Ausnahmen ermöglichen eine viel feinere Kontrolle über den Ablauf der Ausnahmebehandlung. Im Prinzip ist es eine gute Idee, eine Root-Ausnahme für Ihren Code zu erstellen (die von Exception erbt.
Dabei ist die Definition von AuthenticationFailed eigentlich ganz einfach:
class AuthenticationFailed(Exception):
"Class to indicated that an authentication request failed"
def __init__(self, message, **kwargs):
self.message = message
super().__init__(**kwargs)Mehr Ausnahmebehandlung in unseren Ablauf einbauen
Aber eine fehlgeschlagene Authentifizierung ist nicht das einzige, was schief gehen kann. Eine Anfrage, die die Bibliothek requests verwendet, kann viele verschiedene Ausnahmen auslösen, darunter Zeitüberschreitungen und DNS-Probleme. Wir wollen irgendwie zwischen diesen beiden unterscheiden. Der Hauptgrund dafür ist, dass der Benutzer nicht viel dagegen tun kann, wenn das Netzwerk ausgefallen ist, während eine fehlgeschlagene Authentifizierung bedeutet, dass Sie wahrscheinlich irgendwo falsche Parameter verwenden.
Der Code dafür lautet:
def get_endpoints(server: str, auth: Authenticator, num_retries: int=3):
"Get the available API endpoints"
endpoint = server + "/apis"
exception = error = None
for _ in range(num_retries):
auth = next(authenticator)
try:
response = r.get(endpoint, headers=next(auth)).json()
error = response.get('error')
if not error:
return response.get('api')
except r.exceptions.RequestException as e:
exception = e
if not error:
raise exception
code = error.get('code')
message = error.get('message')
raise AuthenticationFailed(f"http code {code} with message {message}")Viel besser! Der Code sagt sehr genau, was schief läuft.
Aber warten Sie einen Moment! Dieser Code ist jetzt ziemlich komplex für eine sehr begrenzte Funktionalität. Alles, was wir letztendlich tun wollten, war, r.get(endpoint, headers=next(auth)).json().get('api') zu erhalten!
Wenn wir verschiedene Funktionen für den Aufruf verschiedener APIs benötigen, vielleicht jeweils POSTing oder PUTing, werden wir eine Menge doppelten Code haben.
Dekorateure zur Rettung
Hier kommen Dekoratoren ins Spiel. Wir können viel Logik in einer "Helferfunktion" kapseln und unseren API-Aufruf dekorieren. Die Dekorfunktion könnte wie folgt aussehen:
from functools import wraps
def retry(f):
"""
Decorator to retry functions that fails because of authentication/network issues
"""
@wraps(f) # 1
def wrap(*args, **kwargs):
"""Decorator to retry network calls"""
n_times = 3
authenticator = get_authenticator(SERVER, KEY, SECRET) # three globals
exception = error = None # 2
for _ in range(n_times):
auth = next(authenticator) # 3
try:
ret, error = f(*args, **kwargs, auth=auth) # 4, #5
if not error:
return ret
except r.exceptions.RequestException as e:
exception = e
if not error: # *
raise exception
code = error.get('code')
message = error.get('message')
raise AuthenticationFailed(f"http code {code} with message {message}")
return wrapHier gibt es eine Menge zu verdauen:
- Hier sorgt
wrapsdafür, dass die dekorierte Funktion ihren eigenen Docstring behält, anstatt den Docstring vonwrapzu erben; - Wir initialisieren alle Fehler und Ausnahmen auf
None, so dass, wenn in der for-Schleife eine Ausnahme ausgelöst wird, die mit*markierte Zeile nicht fehlschlägt; - Hier verschieben wir die Erzeugung eines neuen Authentifikators auf den Dekorator. Der Grund dafür ist, dass sich der API-Aufruf nicht wirklich um diese Details kümmern sollte;
- Und hier ergänzen wir den Funktionsaufruf mit
auth. Mit anderen Worten, die Funktion, die wir dekorieren werden, akzeptiertauthals Parameter, aber wenn wir sie mitretrydekorieren, müssen wir sie nie übergeben (siehe unten, wie man sie aufruft); - Sie sehen, dass die Rückgabewerte der dekorierten Funktion etwas anders sind als in der vorherigen Version. Wir müssen
get_endpointsaktualisieren, um diese Änderung zu berücksichtigen.
Wie bereits in Punkt 5 gesagt, wie ändern wir unsere get_endpoints Funktion, damit sie mit dem Dekorator funktioniert? Die Antwort ist einfach:
@retry
def get_endpoints(server: str, *, auth: Dict[str, str]):
"Get the available API endpoints"
endpoint = server + "/apis"
response = r.get(endpoint, headers=auth).json()
return response.get('apis'), response.get('error')Das ist viel besser. Wir geben jetzt immer die error zurück (die auch keine sein kann) und verschieben die Aufgabe, die apis Antwort zu extrahieren, auf den Dekorateur. Schön!
Aber sind wir glücklich über unseren Code? Vielleicht. Ich störe mich daran, dass jede Funktion ihre eigene authenticator erhält, was der Tatsache widerspricht, dass wir einen generischen Authentifikator wollen. Zweitens ist die Anzahl der Wiederholungsversuche für jede Funktion gleich. Wir hätten gerne mehr Flexibilität. Dekoratoren können in Dekoratoren zweiter Ordnung umgewandelt werden, so dass sie Parameter akzeptieren können. Die Syntax kann am Anfang etwas beängstigend sein, aber so sieht sie aus:
from functools import wraps
def retry(n_times: int, authenticator: Authenticator):
"""
Decorator to retry functions that fails because of authentication/network issues
"""
def fun_wrapper(f):
"Wrapper"
@wraps(f) # 1
def wrap(*args, **kwargs):
"""Decorator to retry network calls"""
exception = error = None # 2
for _ in range(n_times):
auth = next(authenticator) # 3
try:
ret, error = f(*args, **kwargs, auth=auth) # 4, #5
if not error:
return ret
except r.exceptions.RequestException as e:
exception = e
if not error: # *
raise exception
code = error.get('code')
message = error.get('message')
raise AuthenticationFailed(f"http code {code} with message {message}")
return wrap
return fun_wrapperJetzt können wir get_endpoints schreiben als:
authenticator = get_authenticator(SERVER, KEY, SECRET)
[...]
@retry(3, authenticator)
def get_endpoints(server: str, *, auth: Dict[str, str]):
"Get the available API endpoints"
endpoint = server + "/apis"
response = r.get(endpoint, headers=auth).json()
return response.get('apis'), response.get('error')An dieser Stelle könnten wir noch ein wenig weiter gehen und unsere Funktion wie folgt umschreiben:
@retry(3, authenticator)
def get(endpoint: str, *, response_part: str=None, auth: Dict[str, str]):
"Get the available API endpoints"
response = r.get(endpoint, headers=auth).json()
if response_part:
return response.get(response_part), response.get('error')
else:
return response, response.get('error')
get_endpoints = lambda endpoint: get(endpoint, response_part='api')Das macht es einfacher, eine generische get Funktion zu haben, aber Vorsicht: Ich gehe davon aus, dass jeder Endpunkt den Fehler in response.get('error') platziert. Das ist vielleicht nicht immer der Fall!
Es ist ein (functools.)Wraps. Ich meine, ein Wrap!
Ok, damit ist die lange Tirade irgendwie beendet. Ich hoffe, Sie haben eine Vorstellung davon bekommen, wie Generatoren und Dekoratoren über die klassischen Lehrbuchbeispiele hinaus nützlich sein können!
Wenn ich darf, sind hier meine letzten Bemerkungen:
- Echter Code sollte eine Art Backoff eingebaut haben. Das kann im Generator geschehen. Ich habe es nicht aufgenommen, um Sie nicht von Generatoren und Dekoratoren abzulenken. Aber Sie müssen nur die Funktion
backoff(_, t)in der for-Schleife aufrufen, die wie folgt definiert werden kann2:
import time
def backoff(n: int, t: float):
time.sleep(t * 2 ** (n - 1))- Die anfängliche POST zum Abrufen des Authentifikators sollte die Klasse
Retryvon urllib3 verwenden. Der Einfachheit halber haben wir auch dies vermieden. Aber sehen Sie hier, um eine Idee zu bekommen, wie man es implementieren kann3. - Da mein Token-Dienst die Token nicht erneuert, bevor sie ablaufen, gibt es keinen Grund, ihn im Hintergrund vor dem Ablauf der Gültigkeit aufzurufen. Das ist der Grund, warum ich die for-Schleife im Dekorator benötige und nicht einfach das
RetryObjekt verwende: Der Token-Dienst muss nach Ablauf der Gültigkeit aufgerufen werden, um ein neues Token zu erhalten.
Ich weiß, dass Sie es wissen, aber nur für den Fall, dass Sie es nicht wissen: Wir stellen ein. Wenn Sie also gerne Datenwissenschaft betreiben und gute Software schreiben, melden Sie sich bei uns!
Verbessern Sie Ihre Python-Kenntnisse, lernen Sie von den Experten!
Bei GoDataDriven bieten wir eine Vielzahl von Python-Kursen für Anfänger und Experten an, die von den besten Fachleuten auf diesem Gebiet unterrichtet werden. Kommen Sie zu uns und verbessern Sie Ihr Python-Spiel:
- Python für Datenanalysten - Ideal, wenn Sie gerade erst mit Python anfangen.
- Certified Data Science with Python Foundation - Möchten Sie den Schritt von der Datenanalyse und -visualisierung zu echter Datenwissenschaft machen? Dies ist der richtige Kurs.
- Advanced Data Science with Python - Lernen Sie, Ihre Modelle wie ein Profi zu produzieren und Python für maschinelles Lernen zu verwenden.
Verfasst von
Giovanni Lanzani
Unsere Ideen
Weitere Blogs
Contact