Storage (lexigram-storage)
Unified blob storage abstraction for Lexigram Framework — S3, GCS, Azure Blob, and local filesystem
Overview
Section titled “Overview”lexigram-storage provides a unified BlobStore interface over Amazon S3, Google Cloud Storage, Azure Blob Storage, Cloudflare R2, and local disk — with signed URLs, streaming uploads, content-type negotiation, and server-side encryption. All services are wired via StorageProvider, which registers the blob store protocol with the DI container.
Install
Section titled “Install”uv add lexigram-storage# Optional extrasuv add "lexigram-storage[s3]" # Amazon S3 (aiobotocore)uv add "lexigram-storage[gcp]" # Google Cloud Storageuv add "lexigram-storage[azure]" # Azure Blob StorageQuick Start
Section titled “Quick Start”from lexigram import Applicationfrom lexigram.di.module import Module, module
# Import the module from the packagefrom lexigram.storage import StorageModule
@module(imports=[StorageModule.configure(...)])class AppModule(Module): pass
app = Application(modules=[AppModule])if __name__ == "__main__": app.run()Configuration
Section titled “Configuration”Zero-config usage: Call
StorageModule.configure()with no arguments to use defaults.
Option 1 — YAML file
Section titled “Option 1 — YAML file”storage: default_driver: "s3" drivers: s3: bucket: "my-app-uploads" region: "us-east-1" access_key: "${AWS_ACCESS_KEY_ID}" secret_key: "${AWS_SECRET_ACCESS_KEY}"Option 2 — Profiles + Environment Variables (recommended)
Section titled “Option 2 — Profiles + Environment Variables (recommended)”export LEX_STORAGE__ENABLED=true# Environment variables for each fieldOption 3 — Python
Section titled “Option 3 — Python”from lexigram.storage.config import StorageConfigfrom lexigram.storage import StorageModule
config = StorageConfig(default_driver="s3", ...)StorageModule.configure(config)Config reference
Section titled “Config reference”| Field | Default | Env var | Description |
|---|---|---|---|
default_driver | "local" | LEX_STORAGE__DEFAULT_DRIVER | Active driver: s3, gcs, azure, r2, local, memory |
drivers.s3.bucket | — | LEX_STORAGE__DRIVERS__S3__BUCKET | S3 bucket name |
drivers.s3.region | "us-east-1" | LEX_STORAGE__DRIVERS__S3__REGION | AWS region |
drivers.s3.access_key | — | LEX_STORAGE__DRIVERS__S3__ACCESS_KEY | AWS access key ID |
drivers.s3.secret_key | — | LEX_STORAGE__DRIVERS__S3__SECRET_KEY | AWS secret access key |
drivers.s3.endpoint_url | null | LEX_STORAGE__DRIVERS__S3__ENDPOINT_URL | Override endpoint (MinIO, LocalStack) |
drivers.gcs.bucket | — | LEX_STORAGE__DRIVERS__GCS__BUCKET | GCS bucket name |
drivers.gcs.credentials_path | — | LEX_STORAGE__DRIVERS__GCS__CREDENTIALS_PATH | Path to service account JSON |
drivers.azure.account_name | — | LEX_STORAGE__DRIVERS__AZURE__ACCOUNT_NAME | Azure storage account name |
drivers.azure.container | — | LEX_STORAGE__DRIVERS__AZURE__CONTAINER | Azure blob container |
drivers.local.root_dir | — | LEX_STORAGE__DRIVERS__LOCAL__ROOT_DIR | Root directory for local storage |
service.max_file_size_mb | 100 | LEX_STORAGE__SERVICE__MAX_FILE_SIZE_MB | Maximum upload size in MB |
Module Factory Methods
Section titled “Module Factory Methods”| Method | Description |
|---|---|
StorageModule.configure(config, enable_encryption) | Configure with explicit StorageConfig |
StorageModule.stub() | Minimal config for testing |
Key Features
Section titled “Key Features”- S3 — Multi-part upload, server-side encryption, lifecycle rules
- GCS — Resumable uploads, CMEK, uniform bucket-level access
- Azure — Block blobs, shared access signatures (SAS), CDN integration
- R2 — Cloudflare R2 (S3-compatible) support
- Local driver — On-disk storage for development and CI
- Memory driver — In-process storage for unit tests
- Signed URLs — Time-limited GET / PUT pre-signed URLs for direct browser upload
- Content negotiation — Automatic MIME type detection from extension and magic bytes
- Key-value store — Lightweight
KVStorefor small blobs (config, flags) - Streaming —
put_stream()/get_stream()for arbitrarily large files
Testing
Section titled “Testing”async with Application.boot(modules=[StorageModule.stub()]) as app: # your test code ...Key Source Files
Section titled “Key Source Files”| File | What it contains |
|---|---|
src/lexigram/storage/module.py | StorageModule class with factory methods |
src/lexigram/storage/di/provider.py | StorageProvider — wires storage protocols into DI container |
src/lexigram/storage/config.py | StorageConfig and sub-config classes |
src/lexigram/storage/drivers/ | Driver implementations (S3, GCS, Azure, R2, local, memory) |
src/lexigram/storage/blob.py | BlobStore interface implementation |