data / redis

Redis avec Python : redis-py et pooling

Leçon 131 exercice

Explication

Ce que vous allez apprendre

Installer et utiliser redis-py, le client Python officiel de Redis
Configurer un ConnectionPool adapté à une application en production
Regrouper des commandes avec un pipeline pour réduire la latence réseau
Se connecter à un déploiement Sentinel pour la haute disponibilité
Intégrer un client Redis asynchrone dans une application FastAPI

Dans quel contexte ?

Un développeur backend a jusqu'ici testé Redis uniquement dans redis-cli. Il doit maintenant brancher Redis à une API FastAPI qui reçoit plusieurs centaines de requêtes concurrentes par seconde. Créer une nouvelle connexion TCP à chaque requête serait catastrophique en performance : c'est exactement le problème que le pooling de connexions résout.

D'abord, la connexion la plus simple

redis.Redis(host=..., port=..., decode_responses=True) crée un client fonctionnel en une ligne. Le paramètre decode_responses=True mérite d'être noté : sans lui, Redis renvoie des bytes bruts plutôt que des chaînes Python, ce qui oblige à décoder manuellement chaque réponse.

Prérequis

Cette leçon suppose une bonne connaissance de Python (fonctions, async/await) et des commandes Redis de base déjà vues dans les leçons précédentes.

Une fois cette connexion simple comprise, il faut passer à l'échelle de la production

Une instance Redis() créée à chaque requête ouvrirait une nouvelle connexion TCP à chaque fois — coûteux et rapidement limité par le nombre de sockets disponibles. Un ConnectionPool résout ce problème en maintenant un nombre limité de connexions réutilisables, empruntées puis rendues automatiquement à chaque appel.

Paramètre du pool	Rôle
`max_connections`	Nombre maximal de sockets ouverts simultanément
`socket_timeout`	Délai maximal pour une opération avant erreur
`socket_connect_timeout`	Délai maximal pour établir la connexion initiale
`health_check_interval`	Fréquence des pings pour détecter une connexion morte

Piège courant

Créer un nouveau redis.Redis() (donc un nouveau pool implicite) à chaque requête HTTP annule tout le bénéfice du pooling. Le pool doit être créé une seule fois, au démarrage de l'application, puis partagé et réutilisé pour toute sa durée de vie.

Il reste un gain de performance facile à obtenir : le pipeline

Quand plusieurs commandes indépendantes doivent être envoyées (par exemple 1000 SET d'affilée), les envoyer une par une implique un aller-retour réseau complet pour chacune. Un pipeline les regroupe et les envoie en un seul paquet, ramenant la latence totale à celle d'un unique aller-retour au lieu de mille.

Enfin, deux besoins avancés méritent d'être connus

Pour la haute disponibilité, redis.sentinel.Sentinel permet au client de retrouver automatiquement le maître actuel même après un failover, sans coder l'adresse en dur. Et pour une application asynchrone comme FastAPI, redis.asyncio propose une API identique mais non bloquante, à utiliser avec await — idéale pour ne jamais bloquer la boucle d'événements pendant un appel réseau vers Redis.

Bonne pratique

Dans une application FastAPI, crée le pool Redis dans le gestionnaire de démarrage (startup) et ferme-le proprement à l'arrêt (shutdown). Stocker le client dans app.state évite de le recréer à chaque requête tout en le rendant accessible partout dans l'application.

Maintenant que Redis est correctement branché à une application Python en production, la prochaine leçon explore des structures de données plus spécialisées — HyperLogLog, bitmaps et Streams — qui vont bien au-delà des simples chaînes et hash déjà vus.

Commandes & code

Redis avec Python : redis-py et pooling

pip install redis

import redis

# Connexion simple (une connexion socket par instance Redis())
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
r.set("cle", "valeur")
print(r.get("cle"))

# --- Pool de connexions : à utiliser en production ---
# Un pool partage un nombre limité de connexions TCP entre threads/requêtes
pool = redis.ConnectionPool(
    host="localhost",
    port=6379,
    db=0,
    max_connections=50,          # limite le nombre de sockets ouverts simultanément
    decode_responses=True,
    socket_timeout=5,             # timeout sur les opérations
    socket_connect_timeout=2,      # timeout à la connexion initiale
    health_check_interval=30,       # ping périodique pour détecter les connexions mortes
)
r = redis.Redis(connection_pool=pool)

# Le pool est thread-safe : chaque appel emprunte/rend une connexion automatiquement
def handler_requete_http():
    r.incr("compteur:requetes")     # utilise le pool en coulisses, pas de connexion dédiée à gérer

# --- Pipelines : grouper plusieurs commandes en un seul aller-retour réseau ---
pipe = r.pipeline()
for i in range(1000):
    pipe.set(f"item:{i}", i)
pipe.execute()                      # envoie les 1000 commandes en une fois, résultats en liste

# --- Connexion via Redis Sentinel (haute dispo, voir leçon clustering) ---
from redis.sentinel import Sentinel

sentinel = Sentinel([("sentinel1", 26379), ("sentinel2", 26379)], socket_timeout=0.5)
maitre = sentinel.master_for("mymaster", socket_timeout=0.5, decode_responses=True)
repliqua = sentinel.slave_for("mymaster", socket_timeout=0.5, decode_responses=True)

# --- Client asynchrone (asyncio), pour une app FastAPI par exemple ---
import asyncio
import redis.asyncio as aredis

async def main():
    r_async = aredis.Redis(host="localhost", port=6379, decode_responses=True)
    await r_async.set("cle_async", "valeur")
    valeur = await r_async.get("cle_async")
    print(valeur)
    await r_async.aclose()

asyncio.run(main())

# Intégration FastAPI typique : un pool global partagé, créé au démarrage de l'app
from fastapi import FastAPI
import redis.asyncio as aredis

app = FastAPI()

@app.on_event("startup")
async def startup():
    app.state.redis = aredis.Redis(
        host="localhost", port=6379, decode_responses=True, max_connections=50
    )

@app.on_event("shutdown")
async def shutdown():
    await app.state.redis.aclose()

@app.get("/compteur")
async def compteur():
    valeur = await app.state.redis.incr("visites")
    return {"visites": valeur}

Résumé

Toujours utiliser un ConnectionPool en production plutôt qu'une connexion Redis() par appel.
Les pipelines regroupent des commandes indépendantes en un seul aller-retour réseau : gain de latence important.
redis.asyncio s'intègre naturellement dans une app FastAPI via un pool créé au démarrage et fermé à l'arrêt.

Exercices pratiques

1 disponible

Mission : stabiliser une API FastAPI qui sature les connexions Redis

Objectif : Corriger une mauvaise gestion de connexion Redis en production, exploiter un pipeline pour un traitement en masse, et sécuriser un pool contre les connexions mortes.

Contexte

Un développeur a écrit une fonction qui crée une nouvelle connexion Redis à chaque requête HTTP au lieu de réutiliser un pool. Sous forte charge, l'API ralentit et Redis journalise un nombre anormal de connexions ouvertes. Tu dois diagnostiquer ce problème, optimiser une écriture en masse, et anticiper un risque silencieux de connexion morte.

Résoudre l’exercice →