Summary
[!note] 📚 Série Smart Assets Manager
- Pourquoi l’abstraction du stockage est essentielle — 11 mai
- Quatre Backends, Une Interface ← vous êtes ici
- L’API unifiée : crédits et rate limiting — 27 avril
- Stratégie de test : tests unitaires vs tests E2E — 20 avril
- Les 5 cas limites qui cassent les APIs d’images — 8 juin
- Documentation API : Swagger + Postman — 30 mars
Dans le billet précédent, j’ai expliqué pourquoi coupler un générateur d’images à un seul fournisseur de stockage crée une contrainte qui s’accumule au fil de la croissance du système. Voici l’implémentation.
Le design suit le pattern Strategy classique : une classe de base abstraite qui définit le contrat, quatre classes concrètes qui l’implémentent, et une fonction factory qui dispatche selon la préférence de l’appelant. Chaque appelant utilise la même interface. Seule la factory sait quel backend est actif.
Le contrat : classe de base abstraite
La classe de base est l’endroit où le design est encodé. Chaque backend doit implémenter upload() et generate_signed_url(). Tout le reste n’est que détail.
# 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 en paramètre 'storage' de la 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.
Les deux méthodes abstraites constituent l'intégralité de l'interface dont dépendent les appelants.
Changer de backend ne requiert de modifier que la factory, pas le moindre code appelant.
"""
@abstractmethod
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
"""Stocke les octets d'image et retourne une URL ou un data URI.
Le paramètre visibility contrôle l'accès :
- 'public' : URL ouverte, aucune authentification requise
- 'private' : nécessite une URL signée pour accéder (supporté par Cloudinary et S3)
"""
...
@abstractmethod
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
"""Génère une URL signée à durée limitée pour un asset privé.
L'expiration par défaut de 86400 secondes (24 heures) est suffisamment longue
pour la plupart des workflows aval, mais suffisamment courte pour expirer
avant que des liens périmés ne circulent.
"""
...
def cleanup_failed_assets(self, asset_ids: list[str]) -> None:
"""Nettoyage best-effort des uploads partiels après un échec de lot.
Quand une génération multi-taille échoue partiellement — certaines images
uploadées avant l'erreur, les autres non — ces uploads orphelins doivent
être supprimés. Cette méthode ne lève pas d'exception par conception :
l'échec du nettoyage est loggé, jamais re-levé.
"""
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:
"""Supprime un asset stocké par identifiant (usage interne uniquement)."""
...
Trois décisions dans cette classe de base méritent examen.
Le paramètre visibility dans upload() encode une distinction significative : les images publiques obtiennent des URLs CDN ouvertes ; les images privées nécessitent un accès signé. Tous les backends ne supportent pas cela — le stockage local n’a pas ce concept — mais l’interface l’expose afin que les appelants puissent exprimer leur intention, même quand certains backends le traitent comme une opération no-op.
La méthode cleanup_failed_assets() est concrète, non abstraite. C’est intentionnel : la logique de nettoyage (itérer sur les IDs, appeler _delete(), avaler les exceptions) est identique sur tous les backends. Seul _delete() varie. Rendre le nettoyage concret dans la classe de base signifie que les backends l’obtiennent gratuitement — ils n’ont qu’à implémenter _delete().
Le comportement non-levant de cleanup_failed_assets() est délibéré. Le nettoyage s’exécute dans un contexte d’erreur — un lot a échoué, et on tente de supprimer les uploads partiels avant de retourner l’erreur. Si le nettoyage lui-même lève une exception, cela masque l’erreur originale et complique le débogage. Avertir en cas d’échec du nettoyage ; ne jamais re-lever.
Le backend Cloudinary
Cloudinary est la valeur par défaut en production : les images uploadées ici obtiennent des URLs CDN, une optimisation automatique et le support des transformations de format.
import time
import cloudinary.uploader
import cloudinary.utils
from pathlib import Path
class CloudinaryStorage(StorageBackend):
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Le type d'upload 'authenticated' nécessite une signature pour accéder à l'URL.
# Cela empêche le hotlinking et garantit que les images privées restent privées.
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:
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")
La ligne upload_type = "authenticated" if visibility == "private" est un détail qui compte : le type d’upload “authenticated” de Cloudinary verrouille l’asset derrière une signature. Sans cela, marquer une image “privée” dans l’appel API n’aurait aucun effet — l’URL resterait accessible publiquement.
Le backend système de fichiers local
Le stockage local existe pour deux raisons : le développement (pas de credentials, pas de coûts) et les déploiements on-premise (les données restent sur le serveur interne).
from pathlib import Path
class LocalStorage(StorageBackend):
BASE_PATH = Path("generated_assets")
BASE_URL = "http://localhost:8000/assets"
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Créer le répertoire s'il n'existe pas — premier lancement dans un nouvel environnement
self.BASE_PATH.mkdir(parents=True, exist_ok=True)
target = self.BASE_PATH / filename
target.write_bytes(image_bytes)
# L'appelant est responsable de servir ce chemin.
# En développement, l'application FastAPI sert /assets depuis BASE_PATH.
return f"{self.BASE_URL}/{filename}"
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
# Les URLs signées ne s'appliquent pas au stockage local — environnements de confiance uniquement.
return asset_id
def _delete(self, asset_id: str) -> None:
path = self.BASE_PATH / Path(asset_id).name
if path.exists():
path.unlink()
Le generate_signed_url retournant asset_id tel quel est intentionnel. Le stockage local est utilisé en développement et dans des environnements on-premise de confiance où le concept de “signer” une URL ne s’applique pas. Retourner l’ID de l’asset tel quel maintient la cohérence de l’interface sans prétendre que le backend fait quelque chose qu’il ne fait pas.
Le backend téléchargement direct
C’est le backend le plus utilisé par les développeurs, et que personne n’avait prévu.
Au lieu de stocker quoi que ce soit, il encode en base64 les octets de l’image et retourne un data URI directement dans la réponse API. Pas de credentials. Pas de coûts de stockage. Pas d’aller-retour.
import base64
class DirectDownloadStorage(StorageBackend):
"""Retourne les octets d'image sous forme de data URI encodé en base64.
Pas de stockage externe. Pas de credentials nécessaires.
Idéal pour : pipelines CI/CD, développement et tests, consommateurs d'API
qui gèrent leur propre stockage en aval.
"""
# Détection du format par signature d'octets plutôt que par extension de fichier.
# Les appelants ne fournissent pas toujours les bonnes extensions ; les octets font autorité.
_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:
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.
La détection par signature d’octets dans _FORMAT_SIGNATURES mérite attention. Les appelants passent un paramètre filename, mais les noms de fichiers mentent — un appelant peut uploader un PNG avec une extension .jpg, et si le type MIME dans le data URI est image/jpeg, certains navigateurs refuseront de l’afficher. Lire les premiers octets des données d’image réelles et inférer le format à partir des magic bytes est plus fiable que de faire confiance au nom de fichier.
À retenir : Détecter le format d’image par signatures d’octets, pas par extensions de fichier. Les magic bytes
b"\x89PNG"au début d’un PNG sont toujours présents ; l’extension du nom de fichier est ce que l’appelant a fourni.
La fonction factory
Une seule fonction. C’est le seul endroit dans la codebase qui sait quel classe correspond à quel StorageType. Tous les appelants dépendent de l’interface, pas de l’implémentation.
def get_storage_backend(storage_type: str) -> StorageBackend:
"""Retourne le backend de stockage correct pour le type de chaîne donné.
Lève ValueError si le type n'est pas reconnu — les appelants doivent valider
le paramètre storage contre StorageType avant d'appeler cette fonction.
"""
backends = {
StorageType.CLOUDINARY: CloudinaryStorage,
StorageType.LOCAL: LocalStorage,
StorageType.S3: S3Storage,
StorageType.DIRECT: DirectDownloadStorage,
}
backend_class = backends.get(StorageType(storage_type))
if not backend_class:
raise ValueError(f"Unknown storage type: {storage_type!r}")
return backend_class()
Ajouter un cinquième backend — Google Cloud Storage, par exemple — signifie : créer une classe implémentant upload(), generate_signed_url() et _delete(), l’ajouter à ce dictionnaire, ajouter la valeur à StorageType. Le reste de la codebase reste inchangé.
La suite
Le stockage étant 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 que les utilisateurs ne soient jamais facturés sans livraison.
→ Suite : L’API unifiée avec réservation de crédits
— Kékéli
