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 Limitation de Débit — 27 avril
- Stratégie de Tests : Unitaires vs E2E — 20 avril
- 5 Cas Limites qui Brisent les APIs d’Image — 1er juin
- Documentation API : Swagger + Postman — 30 mars
Dans l’article 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’amplifie à mesure que le système grandit. Voici l’implémentation.
La conception repose sur un patron stratégie classique : une classe de base abstraite qui définit le contrat, quatre classes concrètes qui l’implémentent, et une fonction fabrique qui distribue les requêtes en fonction de la préférence de l’appelant. Chaque appelant utilise la même interface. Seule la fabrique sait quel backend est actif.
Le Contrat : La Classe de Base Abstraite
La classe de base est là où la conception est encodée. Chaque backend doit implémenter upload() et generate_signed_url(). Tout le reste est un 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 comme 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 nécessite de modifier que la fabrique, pas le code des appelants.
"""
@abstractmethod
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
"""Stocke les bytes de l'image et renvoie une URL ou un URI de données.
Le paramètre visibility contrôle l'accès :
- 'public' : URL ouverte, sans authentification requise
- 'private' : nécessite une URL signée pour y 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 en aval, mais suffisamment courte pour expirer
avant que des liens obsolètes ne circulent.
"""
...
def cleanup_failed_assets(self, asset_ids: list[str]) -> None:
"""Nettoyage au mieux des uploads partiels après un échec de lot.
Lorsqu'une génération multi-taille échoue partiellement — certaines images
téléversées avant l'erreur, d'autres non — ces uploads orphelins doivent être supprimés.
Cette méthode est conçue pour ne pas lever d'exception : l'échec du nettoyage
est journalisé, 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 d’être examinées.
Le paramètre visibility sur 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 le fait apparaître afin que les appelants puissent exprimer leur intention, même quand certains backends le traitent comme une opération sans effet.
La méthode cleanup_failed_assets() est concrète, pas abstraite. C’est intentionnel : la logique de nettoyage (itérer sur les IDs d’assets, appeler _delete(), avaler les exceptions) est identique pour tous les backends. Seul _delete() varie. Rendre le nettoyage concret sur 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 nous essayons 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. On journalise les échecs de nettoyage ; on ne les re-lève jamais.
Le Backend Cloudinary
Cloudinary est la valeur par défaut en production : les images téléversées ici obtiennent des URLs CDN, une optimisation automatique et le support de transformation 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 “private” 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 sur site (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 démarrage 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 fait que generate_signed_url renvoie asset_id inchangé est intentionnel. Le stockage local est utilisé dans des environnements de développement et des environnements sur site de confiance où le concept de “signer” une URL ne s’applique pas. Renvoyer 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 planifié.
Au lieu de stocker quoi que ce soit, il encode les bytes de l’image en base64 et renvoie un URI de données directement dans la réponse API. Pas de credentials. Pas de coûts de stockage. Pas d’aller-retour.
import base64
class DirectDownloadStorage(StorageBackend):
"""Renvoie les bytes de l'image sous forme d'URI de données encodé en base64.
Pas de stockage externe. Pas de credentials requis.
Idéal pour : les pipelines CI/CD, le développement et les tests, les consommateurs
d'API qui gèrent leur propre stockage en aval.
"""
# Détection du format par signature de bytes plutôt que par extension de fichier.
# Les appelants ne fournissent pas toujours les bonnes extensions ; les bytes 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 URIs de données 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 de signature de bytes dans _FORMAT_SIGNATURES mérite d’être soulignée. Les appelants passent un paramètre filename, mais les noms de fichiers mentent — un appelant peut téléverser un PNG avec une extension .jpg, et si le type MIME dans l’URI de données est image/jpeg, certains navigateurs refuseront de l’afficher. Lire les premiers bytes 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 du fichier.
À retenir : Détectez le format de l’image à partir des signatures de bytes, pas des extensions de fichiers. Les magic bytes
b"\x89PNG"au début d’un PNG sont toujours présents ; l’extension du nom de fichier est whatever l’appelant a fourni.
La Fonction Fabrique
Une seule fonction. C’est le seul endroit dans le codebase qui sait quelle 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:
"""Renvoie le bon backend de stockage pour la chaîne de type donnée.
Lève ValueError si le type n'est pas reconnu — les appelants doivent valider
le paramètre storage par rapport à 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 — par exemple Google Cloud Storage — signifie : ajouter une classe qui implémente upload(), generate_signed_url() et _delete(), l’ajouter à ce dictionnaire, ajouter la valeur à StorageType. Le reste du codebase reste inchangé.
La Suite
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 que les utilisateurs ne soient jamais facturés sans livraison.
