from fastapi import FastAPI, UploadFile, File
from dcd_storage_client import StorageClient
app = FastAPI()
storage = StorageClient()
@app.post("/upload")
async def handle_upload(photo: UploadFile = File(...)):
contents = await photo.read()
asset = storage.upload(contents, folder="avatars", filename=photo.filename)
optimized_url = (
storage.url(asset["publicId"])
.resize(width=500, height=500, crop="fill")
.format("webp")
.quality(80)
.to_string()
)
return {"id": asset["publicId"], "cdnUrl": optimized_url}The official DevCloudDesk Python library (dcd-storage-client) brings the full power of DevCloudDesk's media engine into the Python ecosystem with clean Pythonic idioms, type hints, and zero heavy binary dependencies.
1. Installation#
Install the package via pip or poetry:
bashpip install dcd-storage-client # or poetry add dcd-storage-client
2. Client Setup & Environment Variables#
Import StorageClient or BucketClient and initialize with your cluster details. The client automatically reads DCD_STORAGE_ENDPOINT and DCD_STORAGE_API_KEY if omitted.
pythonimport os from dcd_storage_client import StorageClient, BucketClient # Explicit initialization: storage = StorageClient( endpoint="https://cdn.mycompany.com", api_key="dcd_key_live_99a8b1c4e23", timeout=30.0, # Optional timeout in seconds ) # Or zero-config initialization (reads from env vars): # export DCD_STORAGE_ENDPOINT="https://cdn.mycompany.com" # export DCD_STORAGE_API_KEY="dcd_key_live_99a8b1c4e23" storage = StorageClient()
3. Fluent URL Transformation Builder#
Chain transformations naturally. The .to_string() or .build() method returns the full CDN URL.
python# 1. Fluent method chaining thumb_url = ( storage.url("products/sneakers-black.png") .resize(width=600, height=450, crop="fill") .format("webp") .quality(85) .blur(0) .to_string() ) print("CDN WebP URL:", thumb_url) # https://cdn.mycompany.com/image/upload/w_600,h_450,c_fill,f_webp,q_85/products/sneakers-black.webp # 2. Square Avatar with thumbnail gravity: avatar_url = ( storage.url("users/profile-99.jpg") .crop("thumb") .width(200) .height(200) .format("webp") .to_string() ) # 3. Direct public asset URL without transformations: raw_url = storage.get_public_url("documents/terms-of-service.pdf")
4. Uploading Disk Files, Streams & BytesIO#
The upload() method flexibly accepts string file paths, open file handles, io.BytesIO in-memory buffers, or raw bytes.
pythonimport io # A. Upload from file path: asset_1 = storage.upload("static/banner.png", folder="banners") print("Public ID:", asset_1["publicId"]) print("Bytes:", asset_1["bytes"]) # B. Upload from open file handle: with open("dataset/sample.jpg", "rb") as f: asset_2 = storage.upload(f, folder="datasets", access="public") # C. Upload from in-memory BytesIO (e.g. PIL / Pillow output or matplotlib chart): chart_buffer = io.BytesIO() # chart.save(chart_buffer, format="PNG") chart_buffer.seek(0) asset_3 = storage.upload( chart_buffer, folder="analytics/charts", filename="weekly-growth.png", access="private" # 🔒 Marked as private asset )
5. Generating Signed Download URLs#
Generate cryptographic time-expiring URLs for private assets (invoices, customer documents) with customizable validity:
python# Generate a 15-minute expiring link for a private contract: signed_url = storage.get_signed_url( "analytics/charts/weekly-growth.png", expires_in_seconds=900, resource_type="image", transform="w_800,f_webp" # Can even transform private images! ) print("Secure expiring link:", signed_url) # https://cdn.mycompany.com/cdn/w_800,f_webp/analytics/charts/weekly-growth.png?exp=1792600900&sig=c91a0...
6. Client Upload Tokens & upload_with_token#
Keep your master API key secret on the server while allowing web or mobile clients to upload directly:
# Backend server generates a signed authorization token:
token_data = storage.create_signed_upload_token(
folder="customer-kyc",
max_bytes=15 * 1024 * 1024, # 15MB limit
expires_in_seconds=600, # 10 minutes
)
# Return token_data to frontend:
# {"token": "sig_...", "uploadUrl": "https://cdn.mycompany.com/v1/upload", "expiresAt": 1792600600}
# Or upload using only the token in client microservices without API key:
client_uploader = StorageClient(endpoint="https://cdn.mycompany.com")
uploaded = client_uploader.upload_with_token(
file="kyc-doc.pdf",
token=token_data["token"],
folder="customer-kyc",
)python from fastapi import FastAPI, UploadFile, File, HTTPException from dcd_storage_client import StorageClient, StorageError
app = FastAPI() storage = StorageClient()
@app.post("/api/v1/photos/upload") async def upload_photo(file: UploadFile = File(...)): try: # Read file contents into memory or stream directly file_bytes = await file.read()
bashasset = storage.upload( file_bytes, folder="fastapi-uploads", filename=file.filename, access="public" ) # Build optimized thumbnail URL thumb_url = storage.url(asset["publicId"]).resize(300, 300, "fill").format("webp").to_string() return { "success": True, "publicId": asset["publicId"], "url": asset["url"], "thumbnailUrl": thumb_url } except StorageError as e: raise HTTPException(status_code=500, detail=str(e))
bash### 8. Django Model FileField Recipe Upload user media to DevCloudDesk when models are saved: ```python from django.db import models from dcd_storage_client import StorageClient storage = StorageClient() class UserProfile(models.Model): username = models.CharField(max_length=150) avatar_public_id = models.CharField(max_length=255, blank=True) def set_avatar(self, uploaded_file): asset = storage.upload( uploaded_file.file, folder=f"users/{self.username}", filename=uploaded_file.name ) self.avatar_public_id = asset["publicId"] self.save() @property def avatar_url(self): if not self.avatar_public_id: return "/static/default-avatar.png" return storage.url(self.avatar_public_id).resize(200, 200, "fill").format("webp").to_string()
9. Exception Hierarchy & Error Handling#
Catch granular errors in your application:
pythonfrom dcd_storage_client.exceptions import ( StorageError, AuthenticationError, NotFoundError, PayloadTooLargeError, UploadError, NetworkTimeoutError, ) try: storage.upload("heavy-dataset.zip", folder="backups") except AuthenticationError: print("Invalid API Key! Check your DCD_STORAGE_API_KEY setting.") except PayloadTooLargeError: print("The uploaded file exceeds the maximum allowed payload size.") except NetworkTimeoutError: print("Network connection to the storage server timed out.") except StorageError as err: print(f"DevCloudDesk Storage Exception: {err}")