Cloud Storage Manager

One async interface over AWS S3, Google Cloud Storage and Azure Blob Storage. Application code calls the same methods everywhere; switching cloud is a configuration change.

Features

• One interface for AWS S3, Google Cloud Storage and Azure Blob Storage
• Async operations
• Upload from a file or from bytes, download, list, delete, signed read URLs
• upload_bytes stores a content type, so a signed video URL plays in a browser
• Type-safe configuration using Pydantic
• A drop-in upgrade from 0.1.x (see below)

Installation

The base install covers AWS S3 and Google Cloud Storage, as 0.1.x did. Azure is an extra:

pip install 10x-cloud-storage-manager           # AWS S3 + Google Cloud Storage
pip install "10x-cloud-storage-manager[azure]"  # adds Azure Blob Storage
pip install "10x-cloud-storage-manager[all]"    # all three

The import name is still cloud_storage_manager. From 0.2.0 the package is published as 10x-cloud-storage-manager; cloud-storage-manager on PyPI stops at 0.1.2.

Requires Python 3.10+.

Quick start

import asyncio
from cloud_storage_manager import CloudStorageFactory, StorageConfig


async def main(config: StorageConfig):
    storage = CloudStorageFactory.from_config(config)

    await storage.upload("local/clip.mp4", "recordings/clip.mp4")
    await storage.upload_bytes(b"# Notes", "notes/a.md", content_type="text/markdown")

    data = await storage.download_bytes("notes/a.md")
    names = await storage.list_objects("recordings/")
    sizes = await storage.list_object_sizes("recordings/")  # {name: bytes}

    url = await storage.get_public_url("recordings/clip.mp4", expiration=3600)

    deleted = await storage.delete_if_exists("notes/a.md")  # False if it was already gone

CloudStorageFactory.from_config picks the provider from the config. CloudStorageFactory.get_storage(StorageProvider.GCP, config) still works.

Configuration

CloudStorageFactory.get_storage(provider, config) uses the provider you name, so a config may hold more than one. config.provider and from_config infer it, and raise ValueError unless exactly one is set.

AWS S3

from cloud_storage_manager import AwsConfig, StorageConfig

config = StorageConfig(
    aws=AwsConfig(
        bucket_name="my-bucket",
        access_key_id="...",
        secret_access_key="...",
        region="us-east-1",  # stored but not used; boto resolves the region, as in 0.1.x
    )
)

Google Cloud Storage

from cloud_storage_manager import GcpConfig, StorageConfig

config = StorageConfig(
    gcp=GcpConfig(
        bucket_name="my-bucket",
        project_id="my-project",
        credentials={...},  # service account JSON as a dict
    )
)

Azure Blob Storage

Give exactly one of account_key or connection_string.

from cloud_storage_manager import AzureConfig, StorageConfig

config = StorageConfig(
    azure=AzureConfig(
        account_name="mystorageaccount",
        container_name="my-container",
        account_key="...",
        # or: connection_string="DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;"
    )
)

get_public_url on Azure returns a read-only blob SAS URL, signed with the account key. A connection string that carries a SAS instead of an AccountKey can read and write but cannot sign URLs.

Interface

Every provider implements BaseCloudStorage:

Error handling

Every error is a CloudStorageError: UploadError, DownloadError, NotFoundError, ListError, DeleteError. The SDK's own exception is kept as __cause__.

from cloud_storage_manager import NotFoundError

try:
    data = await storage.download_bytes("missing.txt")
except NotFoundError:
    data = None

Upgrading from 0.1.x

Change only the requirement: swap cloud-storage-manager for 10x-cloud-storage-manager (uninstall the old package first, since both ship the cloud_storage_manager module). Your code does not change: every 0.1.x call behaves as before.

pip uninstall -y cloud-storage-manager
pip install 10x-cloud-storage-manager

• The base install pulls the same packages as 0.1.x; Azure is the [azure] extra.
• StorageConfig accepts every combination 0.1.x accepted.
• upload, delete and get_public_url behave as in 0.1.x on S3 and GCP; the new delete_if_exists is the one to use for the same answer on every cloud.
• Only those three methods are abstract, so a subclass written for 0.1.x still instantiates.
• AWSClientStorage and GcpClientStorage are still importable from cloud_storage_manager.storage_provider.
• Python 3.10+ (0.1.x already used 3.10 syntax and did not import on 3.8 or 3.9).

Development

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
pytest --cov
ruff check . && ruff format --check .

Tests mock every cloud SDK and never call a real cloud.

Branches and releases

• dev is the default branch. Open feature PRs against it.
• prod is the release branch. Every merge into prod publishes to PyPI and creates a v<version> GitHub release (.github/workflows/release.yml).
• Bump [project].version in pyproject.toml in the PR to prod. CI fails the PR if PyPI already has that version.
• Publishing uses the PYPI_API_TOKEN secret on the pypi GitHub environment, which only the prod branch can deploy to. Never upload to PyPI by hand.

License

This project is licensed under the MIT License - see the LICENSE file for details.