Storage Module
The Storage module provides a unified interface for file operations across local filesystems, AWS S3, Google Cloud Storage, and Azure Blob Storage. It supports presigned URLs, CDN invalidation, and automatic content-type detection.
Configuration
example.py
python
Copied!
| 1 | from vorte import Vorte |
| 2 | |
| 3 | app = Vorte( |
| 4 | auto_load=True, |
| 5 | config={ |
| 6 | "storage": { |
| 7 | "default_backend": "s3", |
| 8 | "backends": { |
| 9 | "local": { |
| 10 | "type": "local", |
| 11 | "base_path": "./uploads", |
| 12 | "max_file_size": 104857600, |
| 13 | }, |
| 14 | "s3": { |
| 15 | "type": "s3", |
| 16 | "bucket": "my-app-storage", |
| 17 | "region": "us-east-1", |
| 18 | "access_key": "${'${AWS_ACCESS_KEY}'}", |
| 19 | "secret_key": "${'${AWS_SECRET_KEY}'}", |
| 20 | "max_file_size": 5368709120, |
| 21 | }, |
| 22 | "cdn": { |
| 23 | "type": "s3", |
| 24 | "bucket": "my-cdn-bucket", |
| 25 | "region": "us-east-1", |
| 26 | "access_key": "${'${AWS_ACCESS_KEY}'}", |
| 27 | "secret_key": "${'${AWS_SECRET_KEY}'}", |
| 28 | "cloudfront_distribution": "E1234567890", |
| 29 | "public_url": "https://cdn.myapp.com", |
| 30 | }, |
| 31 | }, |
| 32 | }, |
| 33 | }, |
| 34 | ) |
Basic Usage
example.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage() |
| 4 | |
| 5 | content = b"Hello, World!" |
| 6 | await storage.upload("uploads/hello.txt", content) |
| 7 | |
| 8 | data = await storage.download("uploads/hello.txt") |
| 9 | print(data.decode()) |
| 10 | |
| 11 | exists = await storage.exists("uploads/hello.txt") |
| 12 | print(exists) |
| 13 | |
| 14 | await storage.delete("uploads/hello.txt") |
| 15 | |
| 16 | size = await storage.size("uploads/report.pdf") |
| 17 | print(f"File size: {size} bytes") |
| 18 | |
| 19 | url = await storage.url("uploads/image.jpg") |
| 20 | print(url) |
Local Filesystem
example.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage(backend="local") |
| 4 | |
| 5 | await storage.upload("documents/report.pdf", pdf_bytes) |
| 6 | await storage.upload("images/photo.jpg", image_bytes) |
| 7 | |
| 8 | files = await storage.list("documents/") |
| 9 | for f in files: |
| 10 | print(f"{f.name} ({f.size} bytes, modified: {f.last_modified})") |
| 11 | |
| 12 | await storage.copy("documents/report.pdf", "archive/report-2026.pdf") |
| 13 | await storage.move("temp/upload.csv", "data/upload.csv") |
AWS S3
example.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage(backend="s3") |
| 4 | |
| 5 | await storage.upload( |
| 6 | "users/1/avatar.jpg", |
| 7 | image_bytes, |
| 8 | content_type="image/jpeg", |
| 9 | metadata={"user_id": "1", "uploaded_by": "admin"}, |
| 10 | ) |
| 11 | |
| 12 | await storage.upload_multipart( |
| 13 | "backups/database.sql.gz", |
| 14 | large_file_path="/tmp/db_backup.sql.gz", |
| 15 | chunk_size=8388608, |
| 16 | ) |
| 17 | |
| 18 | objects = await storage.list("users/", recursive=True) |
| 19 | print(f"Found {len(objects)} objects") |
Presigned URLs
Generate temporary URLs for direct file access without exposing credentials. Useful for client-side uploads and downloads.
example.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage(backend="s3") |
| 4 | |
| 5 | download_url = await storage.presigned_url( |
| 6 | "reports/financial-q1.pdf", |
| 7 | expires_in=3600, |
| 8 | operation="download", |
| 9 | ) |
| 10 | |
| 11 | upload_url = await storage.presigned_url( |
| 12 | "uploads/user-upload.csv", |
| 13 | expires_in=600, |
| 14 | operation="upload", |
| 15 | content_type="text/csv", |
| 16 | ) |
| 17 | |
| 18 | print(f"Download: {download_url}") |
| 19 | print(f"Upload: {upload_url}") |
CDN Integration
example.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage(backend="cdn") |
| 4 | |
| 5 | await storage.upload("assets/logo.png", logo_bytes) |
| 6 | |
| 7 | public_url = storage.public_url("assets/logo.png") |
| 8 | print(public_url) |
| 9 | |
| 10 | await storage.invalidate_cdn(["assets/logo.png", "assets/style.css"]) |
| 11 | |
| 12 | await storage.invalidate_cdn_all() |
Streaming
upload_stream.py
python
Copied!
| 1 | from vorte.storage import Storage |
| 2 | |
| 3 | storage = Storage() |
| 4 | |
| 5 | async for chunk in storage.stream("videos/tutorial.mp4", chunk_size=65536): |
| 6 | yield chunk |
| 7 | |
| 8 | async def upload_stream(file): |
| 9 | async for chunk in file.chunks(): |
| 10 | await storage.append(f"temp/{file.name}", chunk) |
| 11 | await storage.finalize(f"temp/{file.name}", move_to=f"uploads/{file.name}") |
File Validation
upload_file.py
python
Copied!
| 1 | from vorte.storage import Storage, FileValidator |
| 2 | |
| 3 | storage = Storage() |
| 4 | |
| 5 | validator = FileValidator( |
| 6 | allowed_types=["image/jpeg", "image/png", "image/webp"], |
| 7 | max_size=5242880, |
| 8 | allowed_extensions=[".jpg", ".jpeg", ".png", ".webp"], |
| 9 | ) |
| 10 | |
| 11 | result = validator.validate(filename="photo.jpg", content_type="image/jpeg", size=1048576) |
| 12 | if not result.valid: |
| 13 | print(result.errors) |
| 14 | |
| 15 | @router.post("/upload") |
| 16 | async def upload_file(file: UploadFile): |
| 17 | validator = FileValidator( |
| 18 | allowed_types=["image/jpeg", "image/png"], |
| 19 | max_size=5242880, |
| 20 | ) |
| 21 | result = validator.validate( |
| 22 | filename=file.filename, |
| 23 | content_type=file.content_type, |
| 24 | size=0, |
| 25 | ) |
| 26 | if not result.valid: |
| 27 | return error_response(code="INVALID_FILE", status_code=400) |
| 28 | |
| 29 | content = await file.read() |
| 30 | await storage.upload(f"uploads/{file.filename}", content) |
| 31 | url = await storage.url(f"uploads/{file.filename}") |
| 32 | return success_response(data={"url": url}) |