Maximisez l'efficacité de votre FastAPI : mise en œuvre incroyablement rapide de la mise en cache et du verrouillage avec py-cachify

Dans le monde trépidant du développement Web, la performance est primordiale. Des mécanismes de mise en cache efficaces peuvent améliorer considérablement la réactivité de votre API en réduisant les calculs redondants et les requêtes de base de données. Dans cet article, nous explorerons comment intégrer la bibliothèque py-cachify dans une application FastAPI à l'aide de SQLModel et Redis pour implémenter la mise en cache et le contrôle de concurrence.

Table des matières:

• Présentation
• Configuration du projet
• Création de modèles de base de données avec SQLModel
• Création de points de terminaison FastAPI
• Mise en cache des résultats des points de terminaison
• Verrouillage de l'exécution des points de terminaison de mise à jour
• Exécuter l'application
• Conclusion

Introduction

La mise en cache est une technique puissante pour améliorer les performances des applications Web en stockant les résultats d'opérations coûteuses et en les servant à partir d'un stockage à accès rapide. Avec py-cachify, nous pouvons ajouter de manière transparente une mise en cache à nos applications FastAPI, en utilisant Redis pour le stockage. De plus, py-cachify fournit des outils de contrôle de concurrence, évitant ainsi les conditions de concurrence lors d'opérations critiques.

Dans ce didacticiel, nous expliquerons la configuration de la bibliothèque py-cachify dans une application FastAPI avec SQLModel pour ORM et Redis pour la mise en cache.

Configuration du projet

Commençons par configurer notre environnement de projet.

Conditions préalables

• Python 3.12
• Poésie (vous pouvez utiliser n'importe quel gestionnaire de paquets de votre choix)
• Serveur Redis exécuté localement ou accessible à distance

Installer les dépendances

Démarrer un nouveau projet via la poésie :

# 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 : Le framework Web pour créer notre API.
• SQLModel aiosqlite : combine SQLAlchemy et Pydantic pour l'ORM et la validation des données.
• Redis : client Python pour interagir avec Redis.
• py-cachify : utilitaires de mise en cache et de verrouillage.

Initialisation de py-cachify

Avant de pouvoir utiliser py-cachify, nous devons l'initialiser avec nos clients Redis. Nous ferons cela en utilisant le paramètre de durée de vie de 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)

À l'intérieur de la durée de vie, nous :

• Créez un client Redis asynchrone.
• Initialisez py-cachify avec ce client.

Création de modèles de base de données avec SQLModel

Nous allons créer un modèle utilisateur simple pour interagir avec notre base de données.

# 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

Configurer le moteur de base de données et créer les tables dans la fonction durée de vie :

# 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

Remarque : nous utilisons SQLite pour plus de simplicité, mais vous pouvez utiliser n'importe quelle base de données prise en charge par SQLAlchemy.

Création de points de terminaison FastAPI

Créons des points de terminaison pour interagir avec notre modèle utilisateur.

# 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

Mise en cache des résultats des points de terminaison

Maintenant, mettons en cache les résultats du point de terminaison read_user pour éviter les requêtes inutiles dans la base de données.

Le code du point de terminaison ressemblera à ceci :

# 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)

Avec le décorateur @cached :

• Nous spécifions une clé unique en utilisant le user_id.
• Réglez le TTL (durée de vie) sur 5 minutes (300 secondes).
• Les appels ultérieurs vers ce point de terminaison avec le même user_id dans les 5 minutes renverront le résultat mis en cache.

Réinitialisation du cache lors des mises à jour

Lorsque les données d'un utilisateur sont mises à jour, nous devons réinitialiser le cache pour garantir que les clients reçoivent les dernières informations. Pour y parvenir, modifions le point de terminaison 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

En appelant read_user.reset(user_id=user_id), nous :

• Effacez les données mises en cache pour l'ID utilisateur spécifique.
• Assurez-vous que les requêtes GET suivantes récupèrent de nouvelles données de la base de données.

En dessous, le décorateur mis en cache enveloppe dynamiquement votre fonction, en ajoutant la méthode .reset. Cette méthode imite la signature et le type de la fonction, de cette façon elle sera soit synchronisée, soit asynchrone selon la fonction d'origine et acceptera les mêmes arguments.

La méthode .reset utilise la même logique de génération de clé définie dans le décorateur mis en cache pour identifier l'entrée mise en cache à invalider. Par exemple, si votre modèle de clé de mise en cache est user-{user_id}, l'appel de wait_user.reset(user_id=123) ciblera et supprimera spécifiquement l'entrée de cache pour user_id=123.

Verrouillage de l'exécution des points de terminaison de mise à jour

Pour éviter les conditions de concurrence lors des mises à jour, nous utiliserons le décorateur once pour verrouiller l'exécution du point de terminaison de la mise à jour.

# 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

Avec une fois :

• Nous verrouillons la fonction en fonction de l'ID utilisateur.
• Si une autre requête tente de mettre à jour le même utilisateur simultanément, elle renverra immédiatement la réponse avec un code d'état 226 IM utilisé.
• Cela évite les mises à jour simultanées qui pourraient entraîner des données incohérentes.

En option, vous pouvez configurer @once pour déclencher une exception ou renvoyer une valeur spécifique si le verrou est déjà acquis.

Exécution de l'application

Il est maintenant temps d'exécuter et de tester notre application !

1) Démarrez le serveur Redis :

Assurez-vous que votre serveur Redis s'exécute localement ou est accessible à distance. Vous pouvez démarrer un serveur Redis local à l'aide de Docker :

# 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) Exécutez l'application FastAPI :

Une fois tout configuré, vous pouvez lancer votre application FastAPI à l'aide de Poetry. Accédez au répertoire racine de votre projet et exécutez la commande suivante :

# 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) Tester et jouer avec la mise en cache et le verrouillage :

Mise en cache : Ajoutez un délai (par exemple, en utilisant asyncio.sleep) dans la fonction read_user pour simuler un calcul de longue durée. Observez comment le temps de réponse s'améliore considérablement une fois le résultat mis en cache.

Exemple :

# 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

Concurrence et verrouillage : De même, introduisez un délai dans la fonction update_user pour observer le comportement des verrous lorsque des tentatives de mise à jour simultanées sont effectuées.

Exemple :

# 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)

Ces délais peuvent vous aider à voir l'efficacité des mécanismes de mise en cache et de verrouillage en action, car les lectures ultérieures devraient être plus rapides grâce à la mise en cache et les écritures simultanées sur la même ressource doivent être gérées efficacement grâce au verrouillage.

Maintenant, vous pouvez tester vos points de terminaison à l'aide d'un outil comme Postman ou en accédant à http://127.0.0.1:8000/docs (lorsque l'application est en cours d'exécution !), et observer les améliorations de performances et les contrôles de concurrence en action.

Amusez-vous à expérimenter votre application FastAPI améliorée !

Conclusion

En intégrant py-cachify dans notre application FastAPI, nous avons débloqué une multitude d'avantages qui améliorent à la fois les performances et la fiabilité de notre API.

Récapitulons quelques-uns des principaux points forts :

• Performances améliorées : La mise en cache des appels de fonctions répétitifs réduit les calculs redondants et les accès à la base de données, améliorant ainsi considérablement les temps de réponse.
• Contrôle de la concurrence : Grâce aux mécanismes de verrouillage intégrés, py-cachify empêche les conditions de concurrence et garantit la cohérence des données, ce qui est crucial pour les applications avec un accès simultané élevé.
• Flexibilité : Que vous travailliez avec des opérations synchrones ou asynchrones, py-cachify s'adapte de manière transparente, ce qui en fait un choix polyvalent pour les applications Web modernes.
• Facilité d'utilisation : La bibliothèque s'intègre facilement aux frameworks Python populaires comme FastAPI, vous permettant de démarrer avec un minimum de friction.
• Annotations de type complètes : py-cachify est entièrement annoté par type, ce qui facilite l'écriture d'un code meilleur et plus maintenable avec un minimum d'effort.
• Configuration minimale : Comme démontré dans ce didacticiel, l'ajout de py-cachify ne nécessite que quelques lignes supplémentaires en plus de votre configuration existante pour exploiter pleinement ses capacités.

Pour ceux qui souhaitent explorer davantage, consultez le le référentiel GitHub de py-cachify et la la documentation officielle pour des conseils, des didacticiels et des exemples plus approfondis.

Vous pouvez accéder au code complet de ce tutoriel sur GitHub ici. N'hésitez pas à cloner le référentiel et à jouer avec l'implémentation en fonction des besoins de votre projet.

Si vous trouvez py-cachify bénéfique, pensez à soutenir le projet en lui donnant une étoile sur GitHub ! Votre soutien contribue à générer de nouvelles améliorations et de nouvelles fonctionnalités.

Bon codage !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!