Caching avec FastAPI et Redis

Pourquoi Caching avec FastAPI et Redis ?

Le caching est un outil essentiel pour augmenter la performance des applications web en réduisant la charge sur les serveurs et en accélérant l'accès aux données fréquemment utilisées. En combinant FastAPI, une moderniser framework d'API à haut niveau, avec Redis, une base de données in-memory populaire, on peut créer des systèmes qui répondent rapidement et efficacement même sous un fort débit.

Un cas concret est l’application de recommandations personnalisées. Lorsque les utilisateurs interagissent avec un site e-commerce, leurs préférences et comportements sont stockés temporairement dans le cache pour permettre une expérience utilisateur plus fluide et des recommandations pertinentes à temps réel.

Prerequis

• Connaissance de base de Python
• Familiarité avec les bases de FastAPI
• Installation d'un environnement virtuel (Python 3.6+)
• Installation de Redis (version >= 6)

Pour installer ces outils, utilisez les commandes suivantes :

## Créer un environnement virtuel
python -m venv venv

## Activer l'environnement virtuel
source venv/bin/activate  # Sur Windows : `venv\Scripts\activate`

## Installer FastAPI et Uvicorn (serveur ASGI)
pip install fastapi uvicorn

## Installer Redis en utilisant Homebrew sur macOS
brew install redis

## Lancer Redis en local
redis-server

Concepts fondamentaux

1. FastAPI : Un framework asynchrone pour créer des API REST avec Python 3.7+.

2. Redis : Une base de données in-memory utilisée comme un cache, une base de données NoSQL et même comme une banque de messages.

Exemple de code : Installation et configuration d'une application FastAPI

## Importer les bibliothèques nécessaires
from fastapi import FastAPI
from pydantic import BaseModel
import redis

app = FastAPI()

## Configuration de Redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

## Définition d'un modèle Pydantic pour les données
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
async def create_item(item: Item):
    # Enregistrer l'item dans Redis en utilisant le nom comme clé
    redis_client.set(item.name, item.json())
    return {"message": "Item created", "name": item.name}

@app.get("/items/{item_name}")
async def read_item(item_name: str):
    # Lire l'item de Redis en utilisant le nom comme clé
    item = redis_client.get(item_name)
    if item:
        return {"name": item_name, "data": item.decode()}
    else:
        return {"message": "Item not found"}

Explications

• FastAPI : Le framework est utilisé pour définir les routes de l'API (/items/ et /items/{item_name}).
• RedisClient: Un objet redis.StrictRedis est créé pour interagir avec Redis.
• Item Model: Un modèle Pydantic est défini pour la structure des données qui seront stockées dans le cache.

Mise en pratique : projet fil rouge

Mini-projet : Gestionnaire de tâches

Étape 1: Création du projet

Créez un nouveau répertoire et initialisez un environnement virtuel :

mkdir task_manager
cd task_manager
python -m venv venv
source venv/bin/activate  # Sur Windows : `venv\Scripts\activate`

Étape 2: Installation des dépendances

Installez FastAPI et Redis avec pip :

pip install fastapi uvicorn redis

Étape 3: Création de la structure du projet

Créez les fichiers suivants :

• main.py : Contiendra le code principal de l'application.
• task_manager.py : Contiendra des fonctions pour gérer les tâches.

Structure du répertoire :

task_manager/
├── main.py
└── task_manager.py

Étape 4: Configuration de Redis

Ajoutez le code suivant dans main.py :

## Importer les bibliothèques nécessaires
from fastapi import FastAPI, HTTPException
import redis
from task_manager import create_task, get_task, delete_task

app = FastAPI()

## Configuration de Redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

@app.post("/tasks/", response_model=dict)
async def add_task(task_data: dict):
    # Créer une tâche dans le cache
    task_id = create_task(redis_client, task_data)
    return {"message": "Task created", "task_id": task_id}

@app.get("/tasks/{task_id}", response_model=dict)
async def retrieve_task(task_id: str):
    # Récupérer une tâche du cache
    task = get_task(redis_client, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"task": task}

@app.delete("/tasks/{task_id}", response_model=dict)
async def delete_specific_task(task_id: str):
    # Supprimer une tâche du cache
    success = delete_task(redis_client, task_id)
    if not success:
        raise HTTPException(status_code=404, detail="Task not found")
    return {"message": "Task deleted"}

Étape 5: Implémentation des fonctions de gestion des tâches

Ajoutez le code suivant dans task_manager.py :

import uuid

def create_task(redis_client, task_data):
    # Générer un ID unique pour la tâche
    task_id = str(uuid.uuid4())
    # Enregistrer la tâche dans Redis en utilisant l'ID comme clé
    redis_client.set(task_id, task_data)
    return task_id

def get_task(redis_client, task_id):
    # Récupérer une tâche de Redis en utilisant l'ID comme clé
    task = redis_client.get(task_id)
    if task:
        return task.decode()
    return None

def delete_task(redis_client, task_id):
    # Supprimer une tâche de Redis en utilisant l'ID comme clé
    if redis_client.delete(task_id):
        return True
    return False

Étape 6: Lancer l'application

Exécutez le serveur FastAPI :

uvicorn main:app --reload

Accédez à http://127.0.0.1:8000/docs pour voir la documentation de votre API et tester les routes.

Erreurs frequentes et debugging

Erreur 1 : Connexion échouée à Redis

Code incorrect :

redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

Code correct :

## Assurez-vous que Redis est en cours d'exécution
redis_client = redis.StrictRedis(host='127.0.0.1', port=6379, db=0)

Erreur 2 : Tâche non trouvée

Code incorrect :

def get_task(redis_client, task_id):
    return None

Code correct :

def get_task(redis_client, task_id):
    task = redis_client.get(task_id)
    if task:
        return task.decode()
    return None

Erreur 3 : Tâche non supprimée

Code incorrect :

def delete_task(redis_client, task_id):
    return False

Code correct :

def delete_task(redis_client, task_id):
    if redis_client.delete(task_id):
        return True
    return False

Pour aller plus loin

1. Sérialisation avec ORJSON

Utiliser ORJSON pour une sérialisation rapide et efficace des données.

• Documentation ORJSON

2. Gestion de la session utilisateur avec FastAPI

Intégrer les sessions utilisateurs avec FastAPI en utilisant un middleware.

• Guide FastAPI sur les sessions

3. Utilisation de Redis pour des tâches asynchrones avec Celery

Intégrer Redis et Celery pour exécuter des tâches asynchrones dans votre application.

• Documentation Celery

Défi pratique : Création d'un scraper avec caching

Créez un scraper qui utilise Redis pour stocker les données récupérées afin de ne pas avoir à relire les mêmes pages Web chaque fois.

Bon codage!