Summary
[!note] 📚 Smart Assets Manager Series
- Why Storage Abstraction Matters
- Four Storage Backends, One Interface ← you are here
- The Unified API: Credit Reservation and Rate Limiting — April 1
- Testing Strategy: Unit Tests vs E2E Tests — April 3
- The 5 Edge Cases That Break Image APIs — April 8
- API Documentation: Swagger + Postman as Deliverables — April 10
In the previous post, I made the case for storage abstraction in image generation APIs. Here’s the implementation.
The core design: an abstract base class with two required methods, four concrete implementations, and one factory function that dispatches based on the request parameter. Every caller uses the same interface — only the factory knows which backend is active.
The Abstract Base Class
# 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):
"""Supported storage backends — passed as a request parameter."""
CLOUDINARY = "cloudinary"
LOCAL = "local"
S3 = "s3"
DIRECT = "direct"
class StorageBackend(ABC):
"""Abstract base class for all image storage backends.
Every backend must implement upload() and generate_signed_url().
This contract is the only thing callers depend on — switching
backends requires changing only the factory, not any caller code.
"""
@abstractmethod
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
"""Store image bytes and return a URL or data URI.
Args:
image_bytes: Raw image data (PNG, JPEG, or WebP)
filename: Suggested filename (backends may use or ignore this)
visibility: 'public' for open URLs, 'private' for signed access only
Returns:
str: URL, signed URL, or base64 data URI depending on backend
"""
...
@abstractmethod
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
"""Generate a time-limited signed URL for a private asset."""
...
def cleanup_failed_assets(self, asset_ids: list[str]) -> None:
"""Best-effort cleanup of partial uploads after a batch failure.
Called when a multi-size generation partially fails — some images
were uploaded before the error, and those orphaned assets should
be removed. Does not raise exceptions: failure to clean up is
logged as a warning but never re-raised.
"""
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:
"""Delete a stored asset by its identifier (internal use only)."""
...
The Cloudinary Backend
import time
import cloudinary.uploader
import cloudinary.utils
from pathlib import Path
class CloudinaryStorage(StorageBackend):
"""Cloudinary storage backend.
Supports public uploads (open CDN URL) and private uploads
(accessible only via signed URLs with configurable expiry).
"""
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Use 'authenticated' type for private assets so the URL
# requires a signature to access — preventing 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 defaults to 24 hours — long enough for most workflows,
# short enough to expire before stale links circulate.
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")
The Local File System Backend
from pathlib import Path
class LocalStorage(StorageBackend):
"""Local file system storage backend.
Saves images to a local directory and returns localhost URLs.
Intended for development and on-premise deployments where
external CDN dependency is unwanted or unavailable.
"""
BASE_PATH = Path("generated_assets")
BASE_URL = "http://localhost:8000/assets"
def upload(self, image_bytes: bytes, filename: str, visibility: str = "public") -> str:
# Ensure the target directory exists before writing
self.BASE_PATH.mkdir(parents=True, exist_ok=True)
target = self.BASE_PATH / filename
target.write_bytes(image_bytes)
# Return a localhost URL — the caller is responsible for serving this path
return f"{self.BASE_URL}/{filename}"
def generate_signed_url(self, asset_id: str, expiry_seconds: int = 86400) -> str:
# Local storage doesn't support signed URLs — return as-is
# This is acceptable because local storage is used in trusted environments
return asset_id
def _delete(self, asset_id: str) -> None:
path = self.BASE_PATH / Path(asset_id).name
if path.exists():
path.unlink()
The Direct Download Backend
import base64
class DirectDownloadStorage(StorageBackend):
"""Returns image bytes as a base64-encoded data URI.
No external storage. No credentials required. No storage costs.
The image is returned inline in the API response.
Ideal for:
- CI/CD pipelines (image is processed downstream, not stored)
- Development and testing (no cloud setup needed)
- API consumers who handle their own storage
"""
# Map common byte signatures to MIME type suffixes
_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:
# Detect format from byte signature rather than file extension
# (callers don't always provide correct extensions)
image_format = "png" # Safe default
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:
# Data URIs don't expire — they're ephemeral by nature
return asset_id
def _delete(self, asset_id: str) -> None:
pass # Nothing to delete — data URIs exist only in the response
The Factory Function
def get_storage_backend(storage_type: str) -> StorageBackend:
"""Return the correct storage backend for the given type.
This is the only place in the codebase that knows which class
maps to which storage type. All callers depend on the interface,
not the implementation.
"""
backends = {
StorageType.CLOUDINARY: CloudinaryStorage,
StorageType.LOCAL: LocalStorage,
StorageType.S3: S3Storage, # Implementation follows same pattern
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()
The Cleanup-on-Failure Pattern
One detail worth highlighting: cleanup_failed_assets() on the base class. When a batch generation (say, 16 sizes from a preset) partially fails — 7 succeed and upload, then the 8th raises an exception — the caller should clean up the 7 orphaned uploads before returning the error.
The base class provides this as a “best-effort” method: it tries to delete each asset, but if a deletion fails, it logs a warning and continues. It never raises exceptions, because a failed cleanup is a secondary problem — the primary error already happened.
This pattern is deliberately forgiving because the cleanup runs in an error context. Raising a new exception during error handling obscures the original error and makes debugging harder.
Key Takeaways
- The abstract base class (2 abstract methods + 1 concrete cleanup method) is the entire interface. Callers depend only on this.
- Direct download is the simplest backend and requires zero external dependencies — make it the development default.
- Detect image format from byte signatures, not file extensions. Callers lie; bytes don’t.
- Cleanup-on-failure should be best-effort and non-raising. Exceptions in error handlers hide original errors.
What’s Next
With storage handled, the next piece is the API layer: a unified endpoint that accepts all generation types, validates requests with Pydantic, and coordinates credit reservation to ensure users are never charged without delivery.
→ Next: The Unified API with Credit Reservation
← Previous: Why Storage Abstraction Matters
— Kékéli
