Ernährungsprotokoll-API mit FastAPI, Python und Redis-Caching

Im Zeitalter von Gesundheits- und Wellness-Apps ist die Fähigkeit, Nährwertinformationen schnell zu protokollieren und abzurufen, eine Kernfunktion. Für Entwickler bedeutet dies den Aufbau von APIs, die nicht nur genau, sondern auch unglaublich schnell und skalierbar sind. Eine langsame API kann zu einer frustrierenden Benutzererfahrung führen, die dazu führt, dass Benutzer die Anwendung verlassen.

In diesem Tutorial werden wir eine Hochleistungs-API für Ernährungsprotokolle mit FastAPI erstellen, einem modernen, schnellen (Hochleistungs-)Web-Framework zum Erstellen von APIs mit Python. Wir werden Pydantic für eine robuste Datenvalidierung und Redis für blitzschnelles Caching verwenden, um Datenbankabfragen für häufig abgerufene Lebensmittel zu minimieren. Am Ende haben Sie eine produktionsreife Vorlage zum Erstellen Ihrer eigenen Hochleistungs-APIs.

Dies ist für Entwickler relevant, da die Beherrschung dieser Tools es Ihnen ermöglicht, skalierbare, effiziente und reaktionsschnelle Anwendungen zu erstellen, die eine große Anzahl von Anfragen bewältigen können, während die Infrastrukturkosten niedrig gehalten werden.

Das Problem verstehen

Eine Ernährungsprotokoll-Anwendung umfasst typischerweise zwei Hauptfunktionen: die Suche nach Lebensmitteln und ihren Nährwertinformationen sowie das Protokollieren dieser Lebensmittel als Teil einer Mahlzeit. Ein häufiger Leistungsengpass entsteht, wenn Tausende von Benutzern wiederholt die Nährwertdaten für dieselben beliebten Lebensmittel anfordern (z.B. „Apfel", „Banane", „Hähnchenbrust"). Ständiges Abfragen der Datenbank nach diesen statischen Daten ist ineffizient und kann zu langsamen Antwortzeiten und hoher Datenbankbelastung führen.

Unser Ansatz wird dies durch die Implementierung einer Caching-Schicht mit Redis angehen. Wenn ein Benutzer Informationen zu einem Lebensmittel anfordert, prüfen wir zunächst, ob es sich in unserem Redis-Cache befindet. Falls ja, geben wir die Daten direkt aus Redis zurück, einem In-Memory-Datenspeicher, der für seine Geschwindigkeit bekannt ist. Falls nicht, rufen wir sie aus der Datenbank ab, speichern sie im Cache für zukünftige Anfragen und geben sie dann an den Benutzer zurück. Dies verbessert die Leistung für nachfolgende Anfragen desselben Elements erheblich.

Voraussetzungen

Bevor wir beginnen, stellen Sie sicher, dass Folgendes installiert ist:

• Python 3.7+: Sie können es von der offiziellen Python-Website herunterladen.
• Docker: Wir werden Docker verwenden, um einfach eine Redis-Instanz auszuführen. Installationsanweisungen finden Sie auf der Docker-Website.
• Eine IDE Ihrer Wahl: VS Code mit der Python-Erweiterung ist eine großartige Option.
• Grundlegende Kenntnisse in Python, APIs und Datentypen.

Richten wir zunächst unser Projektverzeichnis und unsere virtuelle Umgebung ein:

mkdir nutrition-api
cd nutrition-api
python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`

Jetzt installieren wir die erforderlichen Python-Bibliotheken:

pip install "fastapi[all]" redis

Dieser Befehl installiert FastAPI mit allen Standardabhängigkeiten, einschließlich uvicorn (unserem ASGI-Server) und pydantic, zusammen mit der redis-py-Bibliothek zur Interaktion mit Redis.

Starten Sie als Nächstes eine Redis-Instanz mit Docker:

docker run -d -p 6379:6379 --name redis-cache redis/redis-stack-server:latest

Dieser Befehl lädt das Redis Stack-Image herunter, das Redis Insight zur Visualisierung Ihrer Daten enthält, und führt es im Hintergrund aus.

Schritt 1: Definieren unserer Datenmodelle mit Pydantic

Pydantic ist eine Bibliothek zur Datenvalidierung und Einstellungsverwaltung unter Verwendung von Python-Typannotationen. FastAPI verwendet Pydantic-Modelle, um die Struktur von Anfrage- und Antwortdaten zu definieren.

Erstellen wir eine Datei namens schemas.py und definieren unsere Datenmodelle für ein Lebensmittel und einen Ernährungsprotokolleintrag.

# schemas.py
from pydantic import BaseModel
from typing import Optional
from datetime import date

class FoodItem(BaseModel):
    id: int
    name: str
    calories: float
    protein: float
    fat: float
    carbs: float

class NutritionLog(BaseModel):
    id: int
    food_item: FoodItem
    serving_size: float
    log_date: date

class Config:
        orm_mode = True

Wie es funktioniert

• FoodItem: Stellt ein Lebensmittel mit seinen Nährwertinformationen dar.
• NutritionLog: Stellt das Protokoll eines Benutzers eines bestimmten Lebensmittels an einem bestimmten Datum dar. Es enthält das verzehrte FoodItem und die serving_size.
• Config: orm_mode = True ermöglicht es Pydantic, mit ORM-Objekten zu arbeiten, was wir in diesem Tutorial simulieren werden.

Schritt 2: Einrichten der FastAPI-Anwendung

Jetzt erstellen wir unsere Hauptanwendungsdatei main.py.

# main.py
from fastapi import FastAPI, HTTPException
import redis
import json
from . import schemas

app = FastAPI()

# In-memory "database" for demonstration
fake_db = {
    1: {"id": 1, "name": "Apple", "calories": 95, "protein": 0.5, "fat": 0.3, "carbs": 25},
    2: {"id": 2, "name": "Chicken Breast", "calories": 165, "protein": 31, "fat": 3.6, "carbs": 0},
    3: {"id": 3, "name": "Brown Rice", "calories": 215, "protein": 5, "fat": 1.8, "carbs": 45},
}

# Connect to Redis
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

@app.get("/")
def read_root():
    return {"message": "Welcome to the Nutrition Log API"}

Wie es funktioniert

• Wir importieren FastAPI und erstellen eine app-Instanz.
• Wir erstellen ein fake_db-Wörterbuch, um eine Datenbank für dieses Tutorial zu simulieren. In einer realen Anwendung wäre dies eine Verbindung zu einer Datenbank wie PostgreSQL oder MySQL.
• Wir stellen eine Verbindung zu unserem Redis-Server her. decode_responses=True stellt sicher, dass Antworten von Redis von Bytes in Strings dekodiert werden.

Um die Anwendung auszuführen, verwenden Sie den folgenden Befehl in Ihrem Terminal:

uvicorn main:app --reload

Sie sollten eine Ausgabe sehen, die anzeigt, dass der Server läuft. Sie können jetzt auf die API-Dokumentation unter http://127.0.0.1:8000/docs zugreifen.

Schritt 3: Implementieren der Caching-Logik

Jetzt zum Kern unserer Leistungsverbesserung: der Redis-Caching-Schicht. Wir erstellen einen Endpunkt, um ein Lebensmittel anhand seiner ID abzurufen. Dieser Endpunkt prüft zunächst den Redis-Cache.

# main.py (continued)

@app.get("/foods/{food_id}", response_model=schemas.FoodItem)
async def get_food_item(food_id: int):
    # Check cache first
    cached_food = redis_client.get(f"food:{food_id}")
    if cached_food:
        return json.loads(cached_food)

    # If not in cache, get from "database"
    if food_id not in fake_db:
        raise HTTPException(status_code=404, detail="Food item not found")

    food = fake_db[food_id]

    # Store in cache for future requests
    redis_client.setex(f"food:{food_id}", 3600, json.dumps(food)) # Cache for 1 hour

    return food

Wie es funktioniert

• Async-Endpunkt: Wir definieren die Funktion mit async def. Während unsere aktuelle Datenbankinteraktion synchron ist, bereitet die Verwendung von async unsere API auf asynchrone Datenbanktreiber vor, die die Leistung weiter verbessern würden.
• Cache-Prüfung: Wir versuchen zunächst, das Lebensmittel mit einem eindeutigen Schlüssel (f"food:{food_id}") aus Redis abzurufen.
• Cache-Treffer: Wenn cached_food nicht None ist, bedeutet dies, dass wir einen Cache-Treffer haben. Wir deserialisieren den JSON-String aus Redis und geben ihn zurück.
• Cache-Fehler: Wenn das Element nicht im Cache ist, rufen wir es aus unserem fake_db ab.
• Cache-Befüllung: Wir serialisieren dann das Lebensmittel-Wörterbuch in einen JSON-String und speichern es mit setex in Redis. Der ex-Parameter setzt eine Ablaufzeit in Sekunden (hier 1 Stunde oder 3600 Sekunden), was eine gute Praxis ist, um veraltete Daten zu verhindern.

Testen des Cache

1. Führen Sie die Anwendung aus.
2. Senden Sie eine GET-Anfrage an http://127.0.0.1:8000/foods/1.
3. Beim ersten Mal wird die Antwort aus dem fake_db abgerufen.
4. Senden Sie dieselbe Anfrage erneut. Dieses Mal wird die Antwort aus dem Redis-Cache bedient, was in einem realen Szenario deutlich schneller sein wird.

Zusammenführung aller Komponenten

Hier ist die vollständige main.py-Datei mit einem zusätzlichen Endpunkt zum Protokollieren eines neuen Ernährungseintrags.

# main.py
from fastapi import FastAPI, HTTPException
import redis
import json
from datetime import date
from . import schemas

app = FastAPI()

# In-memory "database" for demonstration
fake_db = {
    "foods": {
        1: {"id": 1, "name": "Apple", "calories": 95, "protein": 0.5, "fat": 0.3, "carbs": 25},
        2: {"id": 2, "name": "Chicken Breast", "calories": 165, "protein": 31, "fat": 3.6, "carbs": 0},
        3: {"id": 3, "name": "Brown Rice", "calories": 215, "protein": 5, "fat": 1.8, "carbs": 45},
    },
    "logs": []
}

# Connect to Redis
try:
    redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    redis_client.ping()
    print("Connected to Redis successfully!")
except redis.exceptions.ConnectionError as e:
    print(f"Could not connect to Redis: {e}")
    redis_client = None

@app.get("/")
def read_root():
    return {"message": "Welcome to the Nutrition Log API"}

@app.get("/foods/{food_id}", response_model=schemas.FoodItem)
async def get_food_item(food_id: int):
    if not redis_client:
        raise HTTPException(status_code=500, detail="Redis connection not available")

    # Check cache first
    cached_food = redis_client.get(f"food:{food_id}")
    if cached_food:
        print("Cache HIT")
        return json.loads(cached_food)
    
    print("Cache MISS")
    # If not in cache, get from "database"
    if food_id not in fake_db["foods"]:
        raise HTTPException(status_code=404, detail="Food item not found")

    food = fake_db["foods"][food_id]

    # Store in cache for future requests
    redis_client.setex(f"food:{food_id}", 3600, json.dumps(food)) # Cache for 1 hour

    return food

@app.post("/logs", response_model=schemas.NutritionLog, status_code=201)
async def create_nutrition_log(food_id: int, serving_size: float):
    if food_id not in fake_db["foods"]:
        raise HTTPException(status_code=404, detail="Food item not found")

    food_item = fake_db["foods"][food_id]
    new_log_id = len(fake_db["logs"]) + 1

    new_log = {
        "id": new_log_id,
        "food_item": food_item,
        "serving_size": serving_size,
        "log_date": date.today().isoformat()
    }
    
    fake_db["logs"].append(new_log)
    
    # In a real app, you might want to invalidate related caches here.
    
    return new_log

Leistungsüberlegungen

• Async vs. Sync: Für I/O-gebundene Operationen wie Datenbankabfragen und externe API-Aufrufe ermöglichen async-Endpunkte FastAPI, andere Anfragen zu bearbeiten, während auf den Abschluss der Operation gewartet wird, was die Parallelität erheblich verbessert. Für CPU-gebundene Aufgaben ist es besser, reguläre def-Endpunkte zu verwenden, da FastAPI diese in einem separaten Thread-Pool ausführt.
• Cache-Invalidierung: Wenn die Nährwertinformationen eines Lebensmittels aktualisiert werden, muss der entsprechende Cache-Eintrag invalidiert (gelöscht) werden, um die Bereitstellung veralteter Daten zu verhindern.
• Cache-Räumungsrichtlinie: Redis hat verschiedene Richtlinien zum Entfernen von Schlüsseln, wenn der Speicherplatz ausgeht. Für Caching sind Richtlinien wie allkeys-lru (Least Recently Used) oft eine gute Wahl.

Best Practices für die Sicherheit

• Datenvalidierung: Pydantic bietet eine starke erste Verteidigungslinie durch die Validierung eingehender Daten gegen Ihre definierten Schemata.
• Umgebungsvariablen: Hartcodieren Sie keine sensiblen Informationen wie Datenbankanmeldedaten oder Redis-Verbindungsdetails. Verwenden Sie Umgebungsvariablen.
• Ratenbegrenzung: Um Missbrauch zu verhindern, sollten Sie eine Ratenbegrenzung für Ihre API-Endpunkte implementieren.

Fazit

Wir haben erfolgreich eine Hochleistungs-API für Ernährungsprotokolle mit FastAPI, Pydantic und Redis erstellt. Wir haben gesehen, wie eine einfache Caching-Schicht die Leistung bei read-lastigen Operationen drastisch verbessern kann. Durch die Nutzung der modernen Funktionen von FastAPI und der Geschwindigkeit von Redis sind Sie nun in der Lage, schnelle, skalierbare und robuste APIs zu erstellen.

Als nächste Schritte könnten Sie:

• Die In-Memory-Datenbank durch eine echte asynchrone Datenbankverbindung ersetzen (z.B. mit databases oder asyncpg).
• Eine Benutzerauthentifizierung implementieren, um Ernährungsprotokolle bestimmten Benutzern zuzuordnen.
• Die Pydantic-Modelle mit detaillierteren Nährwertinformationen erweitern.

Ressourcen

• FastAPI-Dokumentation
• Pydantic-Dokumentation
• Redis-Dokumentation
• redis-py GitHub-Repository

Kurzübersicht

Häufig gestellte Fragen

F: Wie weiß ich, ob dies auf mich zutrifft? A: Besprechen Sie Ihre persönliche und familiäre Krankengeschichte mit Ihrem Arzt, um Ihre individuellen Risikofaktoren einzuschätzen und geeignete nächste Schritte zu bestimmen.

F: Was sollte ich meinen Arzt fragen? A: Gute Fragen sind: „Was bedeuten meine Ergebnisse?", „Was sind meine Optionen?", „Welche Lebensstiländerungen würden helfen?" und „Wann sollte ich einen Folgetermin vereinbaren?"

F: Gibt es Lebensstiländerungen, die helfen können? A: Die meisten Erkrankungen profitieren von grundlegenden Maßnahmen: ausgewogener Ernährung, regelmäßiger körperlicher Aktivität (150+ Minuten/Woche), qualitätsvollem Schlaf (7-9 Stunden), Stressbewältigung und Vermeidung von Tabak.

Haftungsausschluss

Dieser Artikel dient ausschließlich zu Informationszwecken und stellt keine medizinische Beratung dar. Konsultieren Sie stets einen qualifizierten Arzt, bevor Sie Entscheidungen über Ihre Gesundheit, Ihre Behandlung oder Ihre Pflege treffen. Individuelle Umstände variieren, und was für eine Person angemessen ist, kann für eine andere ungeeignet sein.

Wenn Sie einen medizinischen Notfall haben, rufen Sie bitte sofort 112 (oder Ihre örtliche Notrufnummer) an.