Maximieren Sie Ihre FastAPI-Effizienz: Blitzschnelle Implementierung von Caching und Sperren mit py-cachify

In der schnelllebigen Welt der Webentwicklung ist Leistung von größter Bedeutung. Effiziente Caching-Mechanismen können die Reaktionsfähigkeit Ihrer API erheblich verbessern, indem sie redundante Berechnungen und Datenbankabfragen reduzieren. In diesem Artikel erfahren Sie, wie Sie die py-cachify-Bibliothek mithilfe von SQLModel und Redis in eine FastAPI-Anwendung integrieren, um Caching und Parallelitätskontrolle zu implementieren.

Inhaltsverzeichnis:

• Einführung
• Projekt-Setup
• Datenbankmodelle mit SQLModel erstellen
• FastAPI-Endpunkte erstellen
• Endpunktergebnisse zwischenspeichern
• Sperren der Ausführung von Update-Endpunkten
• Ausführen der Anwendung
• Fazit

Einführung

Caching ist eine leistungsstarke Technik zur Verbesserung der Leistung von Webanwendungen, indem die Ergebnisse kostspieliger Vorgänge gespeichert und aus einem Schnellzugriffsspeicher bereitgestellt werden. Mit py-cachify können wir Caching nahtlos zu unseren FastAPI-Anwendungen hinzufügen und dabei Redis für die Speicherung nutzen. Darüber hinaus bietet py-cachify Tools zur Parallelitätskontrolle, die Race Conditions bei kritischen Vorgängen verhindern.

In diesem Tutorial gehen wir durch die Einrichtung der py-cachify-Bibliothek in einer FastAPI-Anwendung mit SQLModel für ORM und Redis für das Caching.

Projekt-Setup

Beginnen wir mit der Einrichtung unserer Projektumgebung.

Voraussetzungen

• Python 3.12
• Poesie (Sie können einen beliebigen Paketmanager verwenden)
• Redis-Server läuft lokal oder ist remote zugänglich

Abhängigkeiten installieren

Starten Sie ein neues Projekt über Poesie:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

• FastAPI: Das Web-Framework zum Erstellen unserer API.
• SQLModel aiosqlite: Kombiniert SQLAlchemy und Pydantic für ORM und Datenvalidierung.
• Redis: Python-Client für die Interaktion mit Redis.
• py-cachify: Caching- und Sperrdienstprogramme.

Py-Cachify wird initialisiert

Bevor wir py-cachify verwenden können, müssen wir es mit unseren Redis-Clients initialisieren. Wir tun dies mithilfe des Lebensdauerparameters von FastAPI.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

Innerhalb der Lebensspanne:

• Erstellen Sie einen asynchronen Redis-Client.
• Py-Cachify mit diesem Client initialisieren.

Erstellen von Datenbankmodellen mit SQLModel

Wir erstellen ein einfaches Benutzermodell für die Interaktion mit unserer Datenbank.

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

Richten Sie die Datenbank-Engine ein und erstellen Sie die Tabellen in der Lifespan-Funktion:

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

Hinweis: Der Einfachheit halber verwenden wir SQLite, Sie können jedoch jede von SQLAlchemy unterstützte Datenbank verwenden.

Erstellen von FastAPI-Endpunkten

Lassen Sie uns Endpunkte erstellen, um mit unserem Benutzermodell zu interagieren.

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

Endpunktergebnisse zwischenspeichern

Lassen Sie uns nun die Ergebnisse des Endpunkts read_user zwischenspeichern, um unnötige Datenbankabfragen zu vermeiden.

Der Endpunktcode sieht folgendermaßen aus:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

Mit dem @cached-Dekorator:

• Wir geben einen eindeutigen Schlüssel mithilfe der Benutzer-ID an.
• Stellen Sie die TTL (Time-to-Live) auf 5 Minuten (300 Sekunden) ein.
• Aufeinanderfolgende Aufrufe dieses Endpunkts mit derselben Benutzer-ID innerhalb von 5 Minuten geben das zwischengespeicherte Ergebnis zurück.

Cache bei Updates zurücksetzen

Wenn die Daten eines Benutzers aktualisiert werden, müssen wir den Cache zurücksetzen, um sicherzustellen, dass Kunden die neuesten Informationen erhalten. Um dies zu erreichen, ändern wir den Endpunkt update_user.

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

Durch den Aufruf von read_user.reset(user_id=user_id) führen wir Folgendes aus:

• Löschen Sie die zwischengespeicherten Daten für die spezifische Benutzer-ID.
• Stellen Sie sicher, dass nachfolgende GET-Anfragen aktuelle Daten aus der Datenbank abrufen.

Darunter umschließt der zwischengespeicherte Dekorator Ihre Funktion dynamisch und fügt die .reset-Methode hinzu. Diese Methode ahmt die Signatur und den Typ der Funktion nach. Auf diese Weise ist sie je nach Originalfunktion entweder synchron oder asynchron und akzeptiert dieselben Argumente.

Die .reset-Methode verwendet dieselbe Schlüsselgenerierungslogik, die im zwischengespeicherten Dekorator definiert ist, um zu identifizieren, welcher zwischengespeicherte Eintrag ungültig gemacht werden soll. Wenn Ihr Caching-Schlüsselmuster beispielsweise „user-{user_id}“ lautet, zielt der Aufruf von „await read_user.reset(user_id=123)“ speziell auf den Cache-Eintrag für „user_id=123“ ab und löscht ihn.

Sperren der Ausführung von Update-Endpunkten

Um Race Conditions während Updates zu verhindern, verwenden wir den Once Decorator, um die Ausführung des Update-Endpunkts zu sperren.

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

Mit einmal:

• Wir sperren die Funktion basierend auf der Benutzer-ID.
• Wenn eine andere Anfrage gleichzeitig versucht, denselben Benutzer zu aktualisieren, wird die Antwort sofort mit dem Statuscode 226 IM verwendet zurückgegeben.
• Dies verhindert gleichzeitige Aktualisierungen, die zu inkonsistenten Daten führen könnten.

Optional können Sie @once so konfigurieren, dass eine Ausnahme ausgelöst wird oder ein bestimmter Wert zurückgegeben wird, wenn die Sperre bereits erworben wurde.

Ausführen der Anwendung

Jetzt ist es an der Zeit, unsere App auszuführen und zu testen!

1) Starten Sie den Redis-Server:

Stellen Sie sicher, dass Ihr Redis-Server lokal läuft oder remote zugänglich ist. Sie können einen lokalen Redis-Server mit Docker starten:

# app/main.py

# Adjust imports
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

from .db import User, engine, session_maker


# Database session dependency
async def get_session():
    async with session_maker() as session:
        yield session


app = FastAPI(lifespan=lifespan)


@app.post('/users/')
async def create_user(user: User, session: AsyncSession = Depends(get_session)) -> User:
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


@app.get('/users/{user_id}')
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)


@app.put('/users/{user_id}')
async def update_user(user_id: int, new_user: User, session: AsyncSession = Depends(get_session)) -> User | None:
    user = await session.get(User, user_id)
    if not user:
        return None

    user.name = new_user.name
    user.email = new_user.email

    session.add(user)
    await session.commit()
    await session.refresh(user)

    return user

2) Führen Sie die FastAPI-Anwendung aus:

Sobald alles eingerichtet ist, können Sie Ihre FastAPI-Anwendung mit Poetry starten. Navigieren Sie zum Stammverzeichnis Ihres Projekts und führen Sie den folgenden Befehl aus:

# app/main.py

# Add the import
from py_cachify import cached


@app.get('/users/{user_id}')
@cached('read_user-{user_id}', ttl=300)  # New decorator
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)

3) Testen und Spielen mit Caching und Sperren:

Caching: Fügen Sie eine Verzögerung hinzu (z. B. mit asyncio.sleep) in der read_user-Funktion, um eine lang laufende Berechnung zu simulieren. Beobachten Sie, wie sich die Reaktionszeit drastisch verbessert, sobald das Ergebnis zwischengespeichert wird.

Beispiel:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

Gleichzeitigkeit und Sperren: Führen Sie in ähnlicher Weise eine Verzögerung in der update_user-Funktion ein, um das Verhalten von Sperren zu beobachten, wenn gleichzeitige Aktualisierungsversuche durchgeführt werden.

Beispiel:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

Diese Verzögerungen können Ihnen helfen, die Wirksamkeit von Caching- und Sperrmechanismen in Aktion zu sehen, da nachfolgende Lesevorgänge aufgrund des Cachings schneller sein sollten und gleichzeitige Schreibvorgänge auf dieselbe Ressource durch Sperren effektiv verwaltet werden sollten.

Jetzt können Sie Ihre Endpunkte mit einem Tool wie Postman oder unter http://127.0.0.1:8000/docs testen (wenn die App ausgeführt wird!) und die Leistungsverbesserungen und Parallelitätskontrollen in Aktion beobachten.

Viel Spaß beim Experimentieren mit Ihrer erweiterten FastAPI-App!

Abschluss

Durch die Integration von py-cachify in unsere FastAPI-Anwendung haben wir eine Fülle von Vorteilen freigeschaltet, die sowohl die Leistung als auch die Zuverlässigkeit unserer API verbessern.

Lassen Sie uns einige der wichtigsten Stärken noch einmal zusammenfassen:

• Verbesserte Leistung: Durch das Zwischenspeichern sich wiederholender Funktionsaufrufe werden redundante Berechnungen und Datenbanktreffer reduziert, wodurch die Antwortzeiten drastisch verbessert werden.
• Parallelitätskontrolle: Mit integrierten Sperrmechanismen verhindert py-cachify Race Conditions und stellt die Datenkonsistenz sicher – entscheidend für Anwendungen mit hohem gleichzeitigem Zugriff.
• Flexibilität: Unabhängig davon, ob Sie mit synchronen oder asynchronen Vorgängen arbeiten, passt sich py-cachify nahtlos an und ist somit eine vielseitige Wahl für moderne Webanwendungen.
• Benutzerfreundlichkeit: Die Bibliothek lässt sich reibungslos in gängige Python-Frameworks wie FastAPI integrieren, sodass Sie mit minimaler Reibung loslegen können.
• Vollständige Typanmerkungen: py-cachify ist vollständig typannotiert und hilft dabei, mit minimalem Aufwand besseren, wartbareren Code zu schreiben.
• Minimaler Setup: Wie in diesem Tutorial gezeigt, erfordert das Hinzufügen von py-cachify nur ein paar zusätzliche Zeilen zusätzlich zu Ihrem bestehenden Setup, um seine Funktionen voll auszuschöpfen.

Für diejenigen, die mehr erfahren möchten, schauen Sie sich das GitHub-Repository von py-cachify und die offizielle Dokumentation an, um ausführlichere Anleitungen, Tutorials und Beispiele zu erhalten.

Sie können hier auf den vollständigen Code für dieses Tutorial auf GitHub zugreifen. Klonen Sie gerne das Repository und experimentieren Sie mit der Implementierung, um sie an die Anforderungen Ihres Projekts anzupassen.

Wenn Sie py-cachify nützlich finden, sollten Sie das Projekt unterstützen, indem Sie ihm auf GitHub einen Stern geben! Ihre Unterstützung trägt dazu bei, weitere Verbesserungen und neue Funktionen voranzutreiben.

Viel Spaß beim Codieren!

Das obige ist der detaillierte Inhalt vonMaximieren Sie Ihre FastAPI-Effizienz: Blitzschnelle Implementierung von Caching und Sperren mit py-cachify. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!