Die besten Tipps & Tricks für robuste API Calls mit Python Skripten

Um sicherzustellen, dass dein Python-Skript robust bleibt, wenn es regelmäßig API-Überprüfungen durchführt und JSON-Daten in eine lokale Datenbank schreibt, insbesondere angesichts von Verbindungsproblemen oder gelegentlichen Fehlern, kannst du eine Vielzahl von Strategien anwenden.

Hier ist eine Aufschlüsselung einiger effektiver Techniken:

1. Exception Handling nutzen: Setze try-except-Blöcke ein, um Ausnahmen abzufangen und zu behandeln, die während der API-Aufrufe auftreten können. Dies hilft, bekannte Fehler wie Timeouts, Verbindungsfehler usw. zu verwalten.

2. Wiederholungen mit Backoff implementieren: Im Falle eines Fehlers (wie ein Timeout oder ein 500er Statuscode vom Server) solltest du den Request wiederholen. Es wird empfohlen, exponentielles Backoff und Jitter für Wiederholungen zu verwenden, um den Server nicht mit wiederholten Anfragen in schneller Folge zu überfluten.

3. Timeouts setzen: Definiere Timeouts für deine Anfragen, um sicherzustellen, dass dein Skript nicht unendlich lange bei einem einzigen Aufruf hängen bleibt.

4. Eine robuste HTTP-Bibliothek verwenden: Bibliotheken wie requests sind gut, aber für ein widerstandsfähigeres Verhalten könnten httpx oder urllib3, die umfassendere Funktionen für Verbindungs-Pooling und Wiederholungsstrategien bieten, eine bessere Wahl sein.

5. Logging: Implementiere Logging, um zu verfolgen, wann Probleme auftreten. Protokolliere nützliche Informationen, die beim Debuggen oder Verstehen der Natur der Fehler helfen können.

6. Die Gesundheit der API überwachen: Wenn möglich, überwache die Gesundheit und den Status der API mit deren Status-Endpunkt oder durch externe Überwachungswerkzeuge. Dies kann helfen, Probleme zu erkennen, bevor Anfragen gemacht werden.

7. Datenbanktransaktionen: Stelle sicher, dass Datenbankschreibvorgänge innerhalb von Transaktionen gehandhabt werden, damit im Falle eines Fehlers die Änderungen zurückgerollt werden können, um die Datenintegrität zu wahren.

8. Benachrichtigungssystem: Richte Benachrichtigungen ein (z. B. E-Mails, SMS, Slack-Nachrichten) für kritische Fehler, sodass sofortige Maßnahmen ergriffen werden können, wenn etwas schiefgeht.

9. Circuit Breaker-Muster: Implementiere ein Circuit Breaker-Muster, das die Ausführung von API-Aufrufen vorübergehend stoppen kann, wenn eine bestimmte Schwelle von Fehlern erreicht wird, um zu verhindern, dass dein System überwältigt wird.

Wie sieht das Ganze anhand eines Beispiel-Python-Skripts aus?

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import sqlite3
import logging
import time
import smtplib  # Für das Benachrichtigungssystem

# Logging einrichten (Punkt 5)
logging.basicConfig(level=logging.INFO, filename='api_calls.log')

# Benachrichtigungssystem einrichten (Punkt 8)
def send_notification(message):
    try:
        server = smtplib.SMTP('smtp.example.com', 587)
        server.starttls()
        server.login("your_email@example.com", "password")
        server.sendmail("from@example.com", "to@example.com", message)
        server.quit()
        logging.info("Notification sent.")
    except Exception as e:
        logging.error(f"Failed to send notification: {e}")

# Robuste HTTP-Session aufbauen (Punkt 4)
def setup_http_session():
    session = requests.Session()
    retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])  # Wiederholungen mit Backoff (Punkt 2)
    session.mount('http://', HTTPAdapter(max_retries=retries))
    return session

# Exception Handling für API-Aufrufe (Punkt 1)
def fetch_data(url, session):
    try:
        response = session.get(url, timeout=(2, 5))  # Timeouts setzen (Punkt 3)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logging.error(f"Request failed: {e}")
        return None

# Datenbanktransaktionen (Punkt 7)
def save_to_database(data, db_path):
    conn = sqlite3.connect(db_path)
    c = conn.cursor()
    try:
        c.execute("INSERT INTO api_data (json_data) VALUES (?)", (str(data),))
        conn.commit()
    except sqlite3.DatabaseError as e:
        logging.error(f"Database error: {e}")
        conn.rollback()
    finally:
        conn.close()

# API-Gesundheitsüberwachung (Punkt 6)
def check_api_health(url):
    try:
        response = requests.get(url)
        response.raise_for_status()
        return True
    except requests.exceptions.RequestException as e:
        logging.error(f"API Health Check failed: {e}")
        return False

# Circuit Breaker-Muster (Punkt 9)
def is_circuit_open(failures, threshold=3):
    return failures >= threshold

# Hauptlogik des Skripts
def main(url, db_path):
    session = setup_http_session()
    health_check_url = url + "/health"
    failures = 0

    while True:
        # Überprüfen der API-Gesundheit (Punkt 6)
        if not check_api_health(health_check_url):
            send_notification("API health check failed, API calls suspended.")  # Benachrichtigung senden (Punkt 8)
            time.sleep(60)
            continue

        # Circuit Breaker überprüfen (Punkt 9)
        if is_circuit_open(failures):
            logging.error("Circuit is open. API calls suspended.")
            send_notification("Circuit breaker activated, API calls suspended.")  # Benachrichtigung senden (Punkt 8)
            time.sleep(60)
            continue

        data = fetch_data(url, session)
        if data:
            save_to_database(data, db_path)  # Daten in die Datenbank schreiben (Punkt 7)
            failures = 0  # Fehlerzähler zurücksetzen
        else:
            failures += 1  # Fehlerzähler erhöhen
            logging.error(f"Incremented failure count: {failures}")

        time.sleep(10)  # Pause zwischen API-Aufrufen

if __name__ == "__main__":
    api_url = "http://example.com/api/data"
    database_path = "example.db"
    main(api_url, database_path)

In diesem Skript findest Du alle o.g. 9 Punkte wieder:

• Punkt 1 bis 3, 4, 5: Exception Handling, Retries, Timeouts und Logging sind in den relevanten Funktionsblöcken eingerichtet.

• Punkt 6 und 8: Die Gesundheitsüberwachung der API und das Benachrichtigungssystem sind implementiert, um auf Störungen reagieren zu können.

• Punkt 7: Die Transaktionssicherheit der Datenbankoperationen ist durch Try-Except-Blöcke garantiert, die sicherstellen, dass im Fehlerfall keine inkonsistenten Daten gespeichert werden.

• Punkt 9: Der Circuit Breaker schützt das System vor Überlastung bei wiederholten Fehlern und unterbricht temporär die Ausführung der API-Aufrufe.

Dieses Beispiel ist umfassend und integriert alle erforderlichen Punkte zur Sicherstellung einer robusten Anwendung, die API-Daten abruft und verarbeitet. Es kann als solide Basis dienen, die je nach Bedarf weiter angepasst und erweitert werden kann, um spezifische Anforderungen zu erfüllen.

Die in diesem Artikel vorgestellten Strategien sollten Dir helfen, dein Python Skript für API Calls gegen eine Vielzahl von Fehlern abzusichern und die Zuverlässigkeit und Stabilität der Datenverarbeitung zu verbessern.