google-cloud & azure-sdk

from google.cloud import storage
 
client = storage.Client()
bucket = client.bucket("my-gcs-bucket")
blob = bucket.blob("data/report.csv")
blob.upload_from_filename("report.csv")

from azure.storage.blob import BlobServiceClient
 
service = BlobServiceClient.from_connection_string(conn_str)
container = service.get_container_client("uploads")
container.upload_blob(name="report.csv", data=b"hello", overwrite=True)

from __future__ import annotations
 
from dataclasses import dataclass
from typing import Protocol
 
@dataclass
class ObjectRef:
    bucket: str
    key: str
 
class ObjectStore(Protocol):
    def upload_bytes(self, ref: ObjectRef, data: bytes, content_type: str) -> None: ...
 
class GCSStore:
    def __init__(self, project: str | None = None) -> None:
        from google.cloud import storage
        self._client = storage.Client(project=project)
 
    def upload_bytes(self, ref: ObjectRef, data: bytes, content_type: str) -> None:
        bucket = self._client.bucket(ref.bucket)
        blob = bucket.blob(ref.key)
        blob.upload_from_string(data, content_type=content_type)
 
class AzureBlobStore:
    def __init__(self, connection_string: str) -> None:
        from azure.storage.blob import BlobServiceClient
        self._service = BlobServiceClient.from_connection_string(connection_string)
 
    def upload_bytes(self, ref: ObjectRef, data: bytes, content_type: str) -> None:
        container = self._service.get_container_client(ref.bucket)
        container.upload_blob(name=ref.key, data=data, overwrite=True, content_type=content_type)
 
def persist_report(store: ObjectStore, ref: ObjectRef, payload: bytes) -> None:
    store.upload_bytes(ref, payload, "text/csv")
 
if __name__ == "__main__":
    ref = ObjectRef(bucket="reports", key="daily/summary.csv")
    # persist_report(GCSStore(), ref, b"date,total\n2026-07-09,42\n")
    print(ref)

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
 
credential = DefaultAzureCredential()
service = BlobServiceClient(account_url="https://acct.blob.core.windows.net", credential=credential)