Callbacks¶
filestore supports dynamic resolution of filenames, destinations, metadata, and filters via callbacks. Every callback receives a single UploadContext:
from filestore import UploadContext
def callback(ctx: UploadContext):
ctx.request # The Starlette/FastAPI request object
ctx.form # The parsed multipart form data
ctx.field_name # Name of the upload field being processed
ctx.file # The UploadFile being processed
All callbacks can be sync or async — filestore handles both transparently.
Dynamic Destination¶
Route uploads to different directories based on request context:
from pathlib import Path
from filestore import FileStore, LocalEngine
async def user_directory(ctx):
"""Store uploads in per-user directories."""
user_id = ctx.request.headers.get("X-User-ID", "anonymous")
return Path("uploads") / user_id
storage = FileStore(
"file",
engine=LocalEngine(),
config={"destination": user_directory},
)
For cloud engines, the destination becomes the key prefix:
from filestore import FileStore, S3Engine
async def tenant_prefix(ctx):
tenant = ctx.request.headers.get("X-Tenant-ID")
return f"tenants/{tenant}/uploads"
storage = FileStore(
"file",
engine=S3Engine(bucket="my-bucket"),
config={"destination": tenant_prefix},
)
Dynamic Filename¶
Override the stored filename:
import uuid
from pathlib import Path
from filestore import FileStore
def unique_name(ctx):
"""Generate a UUID-based filename, preserving the extension."""
suffix = Path(ctx.file.filename or "").suffix
return f"{uuid.uuid4()}{suffix}"
storage = FileStore("file", config={"destination": "uploads", "filename": unique_name})
Subdirectory in Filename¶
The callback can return a path with subdirectories:
from datetime import date
def dated_name(ctx):
today = date.today().isoformat()
return f"{today}/{ctx.file.filename}"
This creates uploads/2026-07-12/report.pdf.
Returning an UploadFile¶
For advanced use, the filename callback can return a modified UploadFile
(for example, a re-encoded or transformed file):
Static Filename¶
You can also set a fixed string instead of a callback:
Sanitization
Whatever the callback returns is still normalized: traversal segments are stripped and,
when sanitize_filename=True (the default), unsafe characters are replaced with
underscores. The original client filename is always preserved in
FileData.original_filename.
Metadata Callbacks¶
Attach custom metadata to each uploaded file:
def request_metadata(ctx):
return {
"request_id": ctx.request.headers.get("X-Request-ID"),
"uploaded_by": ctx.request.headers.get("X-User-ID"),
"ip_address": ctx.request.client.host if ctx.request.client else None,
}
storage = FileStore("file", config={"destination": "uploads", "metadata": request_metadata})
The returned dict is merged into FileData.metadata:
@app.post("/upload")
async def upload(store: Store = Depends(storage)):
file_data = store.first("file")
print(file_data.metadata)
# {"request_id": "abc-123", "uploaded_by": "u42", ...}
Static Metadata¶
Pass a dict directly for fixed metadata:
Filter Callbacks¶
See the Validation guide for full details on filter callbacks.
Async Callbacks¶
All callbacks support async:
async def resolve_destination(ctx):
user = await get_current_user(ctx.request)
return f"uploads/{user.id}"
async def resolve_filename(ctx):
digest = await compute_hash(ctx.file)
return f"{digest}{Path(ctx.file.filename or '').suffix}"
async def resolve_metadata(ctx):
return {"processed_at": datetime.now(timezone.utc).isoformat()}
Per-Field Callbacks¶
Callbacks can be set at the store level (applies to all fields) or per-field (overrides the store level):
from filestore import FileField, FileStore
storage = FileStore(
fields=[
FileField(
name="avatar",
config={
"destination": "uploads/avatars",
"filename": avatar_namer, # Only for avatars
},
),
FileField(
name="document",
config={
"destination": "uploads/docs",
"filename": document_namer, # Only for documents
},
),
],
config={"metadata": shared_metadata}, # Applied to all fields
)