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.
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.
Commençons par configurer notre environnement de projet.
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
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 :
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é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
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 :
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 :
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.
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 :
En option, vous pouvez configurer @once pour déclencher une exception ou renvoyer une valeur spécifique si le verrou est déjà acquis.
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 !
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 :
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!
La solution à Google Chrome ne fonctionne pas
La différence entre les fichiers d'en-tête et les fichiers source
mesures de protection de la sécurité du serveur cdn
Introduction à l'emplacement du panneau de configuration Win10
monnaie numérique virtuelle
utilisation de la fonction de tri
Résumé des connaissances de base de Java
Comment supprimer un répertoire sous LINUX
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
