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!