Azure Storage Blob Py

from azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.storage.blob import BlobServiceClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>credential = DefaultAzureCredential(require_envvar=True)# Or use a specific credential directly in production:# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes# credential = ManagedIdentityCredential()account_url = "https://<account>.blob.core.windows.net"
with BlobServiceClient(account_url, credential=credential) as blob_service_client:    # Use blob_service_client here (see following sections for operations)    ...

container_client = blob_service_client.get_container_client("mycontainer")container_client.create_container()

# From file pathblob_client = blob_service_client.get_blob_client(    container="mycontainer",    blob="sample.txt")
with open("./local-file.txt", "rb") as data:    blob_client.upload_blob(data, overwrite=True)
# From bytes/stringblob_client.upload_blob(b"Hello, World!", overwrite=True)
# From streamimport iostream = io.BytesIO(b"Stream content")blob_client.upload_blob(stream, overwrite=True)

blob_client = blob_service_client.get_blob_client(    container="mycontainer",    blob="sample.txt")
# To filewith open("./downloaded.txt", "wb") as file:    download_stream = blob_client.download_blob()    file.write(download_stream.readall())
# To memorydownload_stream = blob_client.download_blob()content = download_stream.readall()  # bytes
# Read into existing bufferstream = io.BytesIO()num_bytes = blob_client.download_blob().readinto(stream)

container_client = blob_service_client.get_container_client("mycontainer")
# List all blobsfor blob in container_client.list_blobs():    print(f"{blob.name} - {blob.size} bytes")
# List with prefix (folder-like)for blob in container_client.list_blobs(name_starts_with="logs/"):    print(blob.name)
# Walk blob hierarchy (virtual directories)for item in container_client.walk_blobs(delimiter="/"):    if item.get("prefix"):        print(f"Directory: {item['prefix']}")    else:        print(f"Blob: {item.name}")

blob_client.delete_blob()
# Delete with snapshotsblob_client.delete_blob(delete_snapshots="include")

# Configure chunk sizes for large uploads/downloadswith BlobClient(    account_url=account_url,    container_name="mycontainer",    blob_name="large-file.zip",    credential=credential,    max_block_size=4 * 1024 * 1024,  # 4 MiB blocks    max_single_put_size=64 * 1024 * 1024  # 64 MiB single upload limit) as blob_client:    # Parallel upload    blob_client.upload_blob(data, max_concurrency=4)
    # Parallel download    download_stream = blob_client.download_blob(max_concurrency=4)

from datetime import datetime, timedelta, timezonefrom azure.identity import DefaultAzureCredentialfrom azure.storage.blob import (    BlobServiceClient,    BlobSasPermissions,    generate_blob_sas,)
now = datetime.now(timezone.utc)account_url = "https://<account>.blob.core.windows.net"
with BlobServiceClient(account_url, credential=DefaultAzureCredential()) as service:    # Get a user delegation key (valid up to 7 days). Caller needs the    # "Storage Blob Delegator" role on the storage account.    udk = service.get_user_delegation_key(        key_start_time=now,        key_expiry_time=now + timedelta(hours=1),    )
    sas_token = generate_blob_sas(        account_name="<account>",        container_name="mycontainer",        blob_name="sample.txt",        user_delegation_key=udk,        permission=BlobSasPermissions(read=True),        expiry=now + timedelta(hours=1),    )
blob_url = f"{account_url}/mycontainer/sample.txt?{sas_token}"

# Get propertiesproperties = blob_client.get_blob_properties()print(f"Size: {properties.size}")print(f"Content-Type: {properties.content_settings.content_type}")print(f"Last modified: {properties.last_modified}")
# Set metadatablob_client.set_blob_metadata(metadata={"category": "logs", "year": "2024"})
# Set content typefrom azure.storage.blob import ContentSettingsblob_client.set_http_headers(    content_settings=ContentSettings(content_type="application/json"))

from azure.identity.aio import DefaultAzureCredentialfrom azure.storage.blob.aio import BlobServiceClient
async def upload_async():    async with DefaultAzureCredential() as credential:        async with BlobServiceClient(account_url, credential=credential) as client:            blob_client = client.get_blob_client("mycontainer", "sample.txt")                        with open("./file.txt", "rb") as data:                await blob_client.upload_blob(data, overwrite=True)
# Download asyncasync def download_async():    async with BlobServiceClient(account_url, credential=credential) as client:        blob_client = client.get_blob_client("mycontainer", "sample.txt")                stream = await blob_client.download_blob()        data = await stream.readall()