Summary
[!note] 📚 Série Smart Assets Manager
- Pourquoi l’abstraction du stockage est essentielle
- Quatre backends de stockage, une interface ← vous êtes ici
- L’API unifiée : réservation de crédits et limitation de débit — 1er avril
- Stratégie de tests : tests unitaires vs E2E — 3 avril
- Les 5 cas limites qui cassent les APIs d’images — 8 avril
- Documentation API : Swagger + Postman comme livrables — 10 avril
Dans le précédent article, j’ai exposé les arguments en faveur de l’abstraction du stockage dans les APIs de génération d’images. Voici l’implémentation.
La conception centrale : une classe de base abstraite avec deux méthodes requises, quatre implémentations concrètes et une fonction factory qui dispatche selon le paramètre de requête. Chaque appelant utilise la même interface — seule la factory sait quel backend est actif.
La classe de base abstraite
# backend/app/integrations/storage_manager.py
from abc import ABC, abstractmethod
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class StorageType(str, Enum):
"""Backends de stockage supportés — passés comme paramètre de requête."""
CLOUDINARY = "cloudinary"
LOCAL = "local"
S3 = "s3"
DIRECT = "direct"
class StorageBackend(ABC):
"""Classe de base abstraite pour tous les backends de stockage d'images.
Chaque backend doit implémenter upload() et generate_signed_url().
Ce contrat est la seule chose dont dépendent les appelants — changer
de backend nécessite de modifier uniquement la factory, pas le code appelant.
"""
@abstractmethod
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
"""Stocker les octets d'image et retourner une URL ou data URI.
Args:
image_bytes: Données brutes de l'image (PNG, JPEG ou WebP)
filename: Nom de fichier suggéré (les backends peuvent l'utiliser ou l'ignorer)
visibility: 'public' pour les URLs ouvertes, 'private' pour l'accès signé uniquement
Returns:
str: URL, URL signée ou data URI base64 selon le backend
"""
...
@abstractmethod
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
"""Générer une URL signée à durée limitée pour un asset privé."""
...
def cleanup_failed_assets(self, asset_ids: list[str]) -> None:
"""Nettoyage au mieux des uploads partiels après un échec de batch.
Appelé quand une génération multi-taille échoue partiellement — certaines images
ont été uploadées avant l'erreur, et ces assets orphelins devraient être supprimés.
Ne lève pas d'exceptions : l'échec du nettoyage est enregistré en avertissement
mais n'est jamais repropagé.
"""
for asset_id in asset_ids:
try:
self._delete(asset_id)
except Exception as e:
logger.warning("Cleanup failed for asset %s: %s", asset_id, e)
@abstractmethod
def _delete(self, asset_id: str) -> None:
"""Supprimer un asset stocké par son identifiant (usage interne uniquement)."""
...
Le backend Cloudinary
import time
import cloudinary.uploader
import cloudinary.utils
from pathlib import Path
class CloudinaryStorage(StorageBackend):
"""Backend de stockage Cloudinary.
Supporte les uploads publics (URL CDN ouverte) et privés
(accessibles uniquement via des URLs signées avec expiration configurable).
"""
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Utiliser le type 'authenticated' pour les assets privés afin que l'URL
# nécessite une signature — empêchant le hotlinking.
upload_type = "authenticated" if visibility == "private" else "upload"
result = cloudinary.uploader.upload(
image_bytes,
public_id=Path(filename).stem,
resource_type="image",
type=upload_type,
)
return result["secure_url"]
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
# expiry_seconds par défaut à 24 heures — assez long pour la plupart des workflows,
# assez court pour expirer avant que des liens périmés circulent.
expires_at = int(time.time()) + expiry_seconds
signed_url, _ = cloudinary.utils.cloudinary_url(
asset_id,
type="authenticated",
sign_url=True,
expires_at=expires_at,
)
return signed_url
def _delete(self, asset_id: str) -> None:
cloudinary.uploader.destroy(asset_id, resource_type="image")
Le backend système de fichiers local
from pathlib import Path
class LocalStorage(StorageBackend):
"""Backend de stockage système de fichiers local.
Enregistre les images dans un répertoire local et retourne des URLs localhost.
Destiné au développement et aux déploiements on-premise où
la dépendance à un CDN externe est indésirable ou indisponible.
"""
BASE_PATH = Path("generated_assets")
BASE_URL = "http://localhost:8000/assets"
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# S'assurer que le répertoire cible existe avant d'écrire
self.BASE_PATH.mkdir(parents=True, exist_ok=True)
target = self.BASE_PATH / filename
target.write_bytes(image_bytes)
# Retourner une URL localhost — l'appelant est responsable de servir ce chemin
return f"{self.BASE_URL}/{filename}"
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
# Le stockage local ne supporte pas les URLs signées — retourner tel quel
# C'est acceptable car le stockage local est utilisé dans des environnements de confiance
return asset_id
def _delete(self, asset_id: str) -> None:
path = self.BASE_PATH / Path(asset_id).name
if path.exists():
path.unlink()
Le backend téléchargement direct
import base64
class DirectDownloadStorage(StorageBackend):
"""Retourne les octets d'image comme data URI encodée en base64.
Pas de stockage externe. Pas d'identifiants requis. Pas de coûts de stockage.
L'image est retournée inline dans la réponse API.
Idéal pour :
- Les pipelines CI/CD (l'image est traitée en aval, pas stockée)
- Le développement et les tests (pas de configuration cloud nécessaire)
- Les consommateurs d'API qui gèrent leur propre stockage
"""
# Correspondance des signatures d'octets communes aux suffixes MIME
_FORMAT_SIGNATURES = {
b"\x89PNG": "png",
b"\xff\xd8\xff": "jpeg",
b"RIFF": "webp",
}
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Détecter le format à partir de la signature d'octets plutôt que de l'extension
# (les appelants ne fournissent pas toujours des extensions correctes)
image_format = "png" # Valeur par défaut sûre
for signature, fmt in self._FORMAT_SIGNATURES.items():
if image_bytes[: len(signature)] == signature:
image_format = fmt
break
encoded = base64.b64encode(image_bytes).decode("utf-8")
return f"data:image/{image_format};base64,{encoded}"
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
# Les data URIs n'expirent pas — ils sont éphémères par nature
return asset_id
def _delete(self, asset_id: str) -> None:
pass # Rien à supprimer — les data URIs n'existent que dans la réponse
La fonction factory
def get_storage_backend(storage_type: str) -> StorageBackend:
"""Retourner le bon backend de stockage pour le type donné.
C'est le seul endroit dans le codebase qui sait quelle classe
correspond à quel type de stockage. Tous les appelants dépendent
de l'interface, pas de l'implémentation.
"""
backends = {
StorageType.CLOUDINARY: CloudinaryStorage,
StorageType.LOCAL: LocalStorage,
StorageType.S3: S3Storage, # L'implémentation suit le même schéma
StorageType.DIRECT: DirectDownloadStorage,
}
backend_class = backends.get(StorageType(storage_type))
if not backend_class:
raise ValueError(f"Unknown storage type: {storage_type}")
return backend_class()
Le schéma nettoyage-sur-échec
Un détail qui mérite d’être mis en avant : cleanup_failed_assets() sur la classe de base. Quand une génération batch (par exemple, 16 tailles à partir d’un preset) échoue partiellement — 7 réussissent et sont uploadées, puis la 8ème lève une exception — l’appelant devrait nettoyer les 7 uploads orphelins avant de retourner l’erreur.
La classe de base fournit cela comme méthode « au mieux » : elle essaie de supprimer chaque asset, mais si une suppression échoue, elle enregistre un avertissement et continue. Elle ne lève jamais d’exceptions, car un nettoyage raté est un problème secondaire — l’erreur principale s’est déjà produite.
Ce schéma est délibérément indulgent car le nettoyage s’exécute dans un contexte d’erreur. Lever une nouvelle exception pendant la gestion d’erreurs masque l’erreur originale et complique le débogage.
Points clés à retenir
- La classe de base abstraite (2 méthodes abstraites + 1 méthode de nettoyage concrète) est toute l’interface. Les appelants n’en dépendent que de ça.
- Le téléchargement direct est le backend le plus simple et ne nécessite aucune dépendance externe — faites-en le défaut de développement.
- Détecter le format d’image à partir des signatures d’octets, pas des extensions de fichiers. Les appelants mentent ; les octets non.
- Le nettoyage-sur-échec devrait être au mieux et sans lever d’exceptions. Les exceptions dans les gestionnaires d’erreurs masquent les erreurs originales.
Et ensuite
Avec le stockage géré, la prochaine pièce est la couche API : un endpoint unifié qui accepte tous les types de génération, valide les requêtes avec Pydantic et coordonne la réservation de crédits pour s’assurer que les utilisateurs ne sont jamais facturés sans livraison.
→ Suivant : L’API unifiée avec réservation de crédits
← Précédent : Pourquoi l’abstraction du stockage est essentielle
— Kékéli
