Fehlerbehandlung stellt häufig einen Schwachpunkt in ansonsten soliden Code dar. Probleme wie fehlende Schlüssel, fehlgeschlagene Anfragen und lang laufende Funktionen treten in realen Projekten häufig auf. Die integrierten try-except-Blöcke von Python sind zwar nützlich, decken jedoch viele praktische Fälle nicht allein ab.
Um häufige Fehlerszenarien zu bewältigen, ist es notwendig, kleine, wiederverwendbare Funktionen zu erstellen, die helfen, Wiederholungen mit Grenzen, Eingangsvalidierungen und Schutzmaßnahmen zu handhaben, die verhindern, dass Code länger als nötig ausgeführt wird. Dieser Artikel erläutert fünf Funktionen zur Fehlerbehandlung, die Sie in Aufgaben wie Web-Scraping, beim Erstellen von Programmierschnittstellen (APIs), bei der Verarbeitung von Benutzerdaten und mehr verwenden können. Weitere Informationen finden Sie in unserem Artikel über Fünf effektive Python-Dekoratoren zur Optimierung von KI-Agenten.
Der Code ist auf GitHub verfügbar.
Wiederholte fehlgeschlagene Operationen mit exponentiellem Backoff
In vielen Projekten schlagen API-Aufrufe und Netzwerkrequests häufig fehl. Ein Anfängeransatz besteht darin, es einmal zu versuchen, Ausnahmen zu erfassen, sie zu protokollieren und dann zu stoppen. Ein besserer Ansatz ist es, es erneut zu versuchen.
Hier kommt das exponentielle Backoff ins Spiel. Anstatt einen fehlerhaften Dienst sofort mit weiteren Versuchen zu belasten – was die Situation nur verschlechtert – wartet man etwas länger zwischen den einzelnen Versuchen: 1 Sekunde, dann 2 Sekunden, dann 4 Sekunden und so weiter.
Hier ist ein Dekorator, der dies umsetzt:
import time
import functools
from typing import Callable, Type, Tuple
def retry_with_backoff(
max_attempts: int = 3,
base_delay: float = 1.0,
exponential_base: float = 2.0,
exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
\"\"\"
Wiederhole eine Funktion mit exponentiellem Backoff.
Args:
max_attempts: Maximale Anzahl der Wiederholungsversuche
base_delay: Anfangsverzögerung in Sekunden
exponential_base: Multiplikator für die Verzögerung (2.0 = Verdopplung jedes Mal)
exceptions: Tuple von Ausnahmetypen, die erfasst und wiederholt werden sollen
\"\"\"
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
time.sleep(base_delay * (exponential_base ** attempt))
raise last_exception
return wrapper
return decorator
Der Dekorator umschließt Ihre Funktion und erfasst die angegebenen Ausnahmen. Die entscheidende Berechnung ist delay = base_delay * (exponential_base ** attempt). Bei base_delay=1 und exponential_base=2 betragen Ihre Verzögerungen 1s, 2s, 4s, 8s. Dies gibt überlasteten Systemen Zeit, sich zu erholen.
Das Parameter exceptions ermöglicht es Ihnen, festzulegen, welche Fehler wiederholt werden sollen. Sie könnten ConnectionError wiederholen, jedoch nicht ValueError, da Verbindungsprobleme vorübergehend sind, Validierungsfehler jedoch nicht.
Jetzt sehen wir es in Aktion:
import random
@retry_with_backoff(max_attempts=4, base_delay=0.5, exceptions=(ConnectionError,))
def fetch_user_data(user_id):
\"\"\"Simuliere eine unzuverlässige API.\"\"\"
if random.random() < 0.5:
raise ConnectionError(\"Fehler beim Abrufen der Benutzerdaten\")
return {'id': user_id, 'name': 'Sara', 'status': 'active'}
Ausgabe:
Success: {'id': 12345, 'name': 'Sara', 'status': 'active'}
Eingaben mit komposierbaren Regeln validieren
Die Validierung von Benutzereingaben ist mühsam und wiederholend. Man überprüft, ob Strings leer sind, ob Zahlen im Bereich liegen und ob E-Mails gültig aussehen. Bevor man sich versieht, hat man überall verschachtelte if-Anweisungen und der Code sieht chaotisch aus.
Lasst uns ein einfach zu verwendendes Validierungssystem erstellen. Zunächst benötigen wir eine benutzerdefinierte Ausnahme:
from typing import Any, Callable, Dict, List, Optional
class ValidationError(Exception):
\"\"\"Wird ausgelöst, wenn die Validierung fehlschlägt.\"\"\"
def __init__(self, field: str, errors: List[str]):
self.field = field
self.errors = errors
super().__init__(f\"{field}: {', '.join(errors)}\")
Diese Ausnahme hält mehrere Fehlermeldungen. Wenn die Validierung fehlschlägt, möchten wir dem Benutzer alles zeigen, was falsch ist, nicht nur den ersten Fehler.
Hier ist der Validator:
def validate_input(
value: Any,
field_name: str,
rules: Dict[str, Callable[[Any], bool]],
messages: Optional[Dict[str, str]] = None
) -> Any:
\"\"\"Validiert die Eingabe gegen mehrere Regeln.
Gibt den Wert zurück, wenn er gültig ist, löst andernfalls ValidationError aus.
\"\"\"
if messages is None:
messages = {}
errors = []
for rule_name, rule_func in rules.items():
try:
if not rule_func(value):
error_msg = messages.get(
rule_name,
f\"Fehlgeschlagene Validierungsregel: {rule_name}\"
)
errors.append(error_msg)
except Exception as e:
errors.append(f\"Validierungsfehler in {rule_name}: {str(e)}\")
if errors:
raise ValidationError(field_name, errors)
return value
Im rules-Dictionary ist jede Regel einfach eine Funktion, die True oder False zurückgibt. Dies macht Regeln komposierbar und wiederverwendbar.
Lasst uns einige gängige Validierungsregeln erstellen:
# Wiederverwendbare Validierungsregeln
def not_empty(value: str) -> bool:
return bool(value and value.strip())
def min_length(min_len: int) -> Callable:
return lambda value: len(str(value)) >= min_len
def max_length(max_len: int) -> Callable:
return lambda value: len(str(value)) <= max_len
def in_range(min_val: int, max_val: int) -> Callable:
return lambda value: min_val <= value <= max_val
Beachten Sie, wie min_length, max_length und in_range Fabrikfunktionen sind. Sie geben Validierungsfunktionen zurück, die mit spezifischen Parametern konfiguriert sind. Dies ermöglicht es Ihnen, min_length(3) zu schreiben, anstatt für jede Längenanforderung eine neue Funktion zu erstellen.
Lasst uns einen Benutzernamen validieren:
try:
username = validate_input(
\"ab\",
\"username\",
{
\"not_empty\": not_empty,
\"min_length\": min_length(3),
\"max_length\": max_length(20),
},
messages={
\"not_empty\": \"Benutzername darf nicht leer sein\",
\"min_length\": \"Benutzername muss mindestens 3 Zeichen lang sein\",
\"max_length\": \"Benutzername darf maximal 20 Zeichen lang sein\",
}
)
print(f\"Gültiger Benutzername: {username}\")
except ValidationError as e:
print(f\"Ungültig: {e}\")
Ausgabe:
Ungültig: username: Benutzername muss mindestens 3 Zeichen lang sein
Dieser Ansatz skaliert gut. Definieren Sie Ihre Regeln einmal, kombinieren Sie sie nach Bedarf und erhalten Sie klare Fehlermeldungen.
Sicheres Navigieren in verschachtelten Dictionaries
Der Zugriff auf verschachtelte Dictionaries ist oft herausfordernd. Sie erhalten KeyError, wenn ein Schlüssel nicht existiert, TypeError, wenn Sie versuchen, einen String zu indizieren, und Ihr Code wird mit Ketten von .get()-Aufrufen oder defensiven try-except-Blöcken überladen. Die Arbeit mit JavaScript Object Notation (JSON) von APIs macht dies noch herausfordernder.
Lasst uns eine Funktion erstellen, die sicher durch verschachtelte Strukturen navigiert:
from typing import Any, Optional, List, Union
def safe_get(
data: dict,
path: Union[str, List[str]],
default: Any = None,
separator: str = \".\"
) -> Any:
\"\"\"Sicherer Zugriff auf einen Wert aus einem verschachtelten Dictionary.
Args:
data: Das Dictionary, auf das zugegriffen werden soll
path: Punktgetrennte Pfad (z.B. \"user.address.city\") oder Liste von Schlüsseln
default: Wert, der zurückgegeben wird, wenn der Pfad nicht existiert
separator: Zeichen zum Trennen des Pfadstrings (Standard: \".\")
Returns:
Der Wert am Pfad oder default, wenn nicht gefunden
\"\"\"
# Konvertiere den String-Pfad in eine Liste
if isinstance(path, str):
keys = path.split(separator)
else:
keys = path
current = data
for key in keys:
try:
# Behandle Listenindizes (konvertiere String in int, wenn numerisch)
if isinstance(current, list):
try:
key = int(key)
except (ValueError, TypeError):
return default
current = current[key]
except (KeyError, IndexError, TypeError):
return default
return current
Die Funktion splittet den Pfad in einzelne Schlüssel und navigiert schrittweise durch die verschachtelte Struktur. Wenn ein Schlüssel nicht existiert oder wenn Sie versuchen, etwas zu indizieren, das nicht indizierbar ist, gibt sie den Standardwert zurück, anstatt abzustürzen.
Sie behandelt auch Listenindizes automatisch. Wenn der aktuelle Wert eine Liste ist und der Schlüssel numerisch ist, wird der Schlüssel in eine Ganzzahl umgewandelt.
Hier ist die Begleitfunktion zum Setzen von Werten:
def safe_set(
data: dict,
path: Union[str, List[str]],
value: Any,
separator: str = \".\",
create_missing: bool = True
) -> bool:
\"\"\"Sicheres Setzen eines Wertes in einem verschachtelten Dictionary.
Args:
data: Das Dictionary, das geändert werden soll
path: Punktgetrennte Pfad oder Liste von Schlüsseln
value: Zu setzender Wert
separator: Zeichen zum Trennen des Pfadstrings
create_missing: Ob fehlende Zwischen-Dicts erstellt werden sollen
Returns:
True, wenn erfolgreich, sonst False
\"\"\"
if isinstance(path, str):
keys = path.split(separator)
else:
keys = path
if not keys:
return False
current = data
# Navigiere zum Elternteil des letzten Schlüssels
for key in keys[:-1]:
if key not in current:
if create_missing:
current[key] = {}
else:
return False
current = current[key]
if not isinstance(current, dict):
return False
# Setze den endgültigen Wert
current[keys[-1]] = value
return True
Die Funktion safe_set erstellt die benötigte verschachtelte Struktur und setzt den Wert. Dies ist nützlich, um Dictionaries dynamisch zu erstellen.
Lasst uns beide testen:
# Beispiel für verschachtelte Daten
user_data = {
\"user\": {
\"name\": \"Anna\",
\"address\": {
\"city\": \"San Francisco\",
\"zip\": \"94105\"
},
\"orders\": [
{\"id\": 1, \"total\": 99.99},
{\"id\": 2, \"total\": 149.50}
]
}
}
# Sichere Get-Beispiele
city = safe_get(user_data, \"user.address.city\")
print(f\"Stadt: {city}\")
country = safe_get(user_data, \"user.address.country\", default=\"Unbekannt\")
print(f\"Land: {country}\")
first_order = safe_get(user_data, \"user.orders.0.total\")
print(f\"Erste Bestellung: ${first_order}\")
# Sichere Set-Beispiel
new_data = {}
safe_set(new_data, \"user.settings.theme\", \"dark\")
print(f\"Erstellt: {new_data}\")
Ausgabe:
Stadt: San Francisco
Land: Unbekannt
Erste Bestellung: $99.99
Erstellt: {'user': {'settings': {'theme': 'dark'}}}
Dieses Muster beseitigt die Unordnung durch defensives Programmieren und macht Ihren Code sauberer, wenn Sie mit JSON, Konfigurationsdateien oder beliebigen tief verschachtelten Daten arbeiten.
Timeouts bei langen Operationen durchsetzen
Einige Operationen dauern zu lange. Eine Datenbankabfrage könnte hängen bleiben, eine Web-Scraping-Operation könnte sich auf einem langsamen Server festsetzen oder eine Berechnung könnte ewig laufen. Sie benötigen eine Möglichkeit, ein Zeitlimit festzulegen und abzubrechen.
Hier ist ein Timeout-Dekorator, der threading verwendet:
import threading
import functools
from typing import Callable, Optional
class TimeoutError(Exception):
\"\"\"Wird ausgelöst, wenn eine Operation ihr Timeout überschreitet.\"\"\"
pass
def timeout(seconds: int, error_message: Optional[str] = None):
\"\"\"Dekorator zur Durchsetzung eines Timeouts bei der Funktionsausführung.
Args:
seconds: Maximale Ausführungszeit in Sekunden
error_message: Benutzerdefinierte Fehlermeldung für Timeout
\"\"\"
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
result = [TimeoutError(
error_message or f\"Operation hat nach {seconds} Sekunden zeitlich überschritten\"
)]
def target():
try:
result[0] = func(*args, **kwargs)
except Exception as e:
result[0] = e
thread = threading.Thread(target=target)
thread.daemon = True
thread.start()
thread.join(timeout=seconds)
if thread.is_alive():
raise TimeoutError(
error_message or f\"Operation hat nach {seconds} Sekunden zeitlich überschritten\"
)
if isinstance(result[0], Exception):
raise result[0]
return result[0]
return wrapper
return decorator
Dieser Dekorator führt Ihre Funktion in einem separaten Thread aus und verwendet thread.join(timeout=seconds), um zu warten. Wenn der Thread nach dem Timeout noch aktiv ist, wissen wir, dass er zu lange gedauert hat, und lösen TimeoutError aus.
Das Ergebnis der Funktion wird in einer Liste (veränderbare Container) gespeichert, damit der innere Thread es ändern kann. Wenn eine Ausnahme im Thread aufgetreten ist, wird sie im Hauptthread erneut ausgelöst.
Eine Einschränkung: Der Thread läuft im Hintergrund weiter, selbst nach dem Timeout. Für die meisten Anwendungsfälle ist dies in Ordnung, aber bei Operationen mit Nebenwirkungen sollten Sie vorsichtig sein.
Lasst uns es testen:
import time
@timeout(2, error_message=\"Abfrage dauerte zu lange\")
def slow_database_query():
\"\"\"Simuliere
Fazit
Die Implementierung effektiver Fehlerbehandlungsstrategien ist entscheidend für die Robustheit Ihrer Anwendungen. Durch die Verwendung von Dekoratoren und komposierbaren Validierungsregeln können Sie Ihren Code nicht nur lesbarer, sondern auch wartbarer gestalten. Weitere Informationen zu leistungsstarken Python-Dekoratoren finden Sie in unseren Artikeln über 5 Powerful Python Decor
Bildquelle: Bildquelle: litoon dev auf Unsplash
