azure

import os from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient # Replace with your storage account name or retrieve from environment variables STORAGE_ACCOUNT_NAME = os.environ.get('AZURE_STORAGE_ACCOUNT_NAME', 'yourstorageaccountname') CONTAINER_NAME = 'my-container' BLOB_NAME = 'hello_world.txt' try: # Authenticate using DefaultAzureCredential (recommended passwordless approach) credential = DefaultAzureCredential() # Construct the blob service client blob_service_client = BlobServiceClient(account_url=f"https://{STORAGE_ACCOUNT_NAME}.blob.core.windows.net/", credential=credential) # Get a client to interact with a specific container container_client = blob_service_client.get_container_client(CONTAINER_NAME) # Create the container if it doesn't exist try: container_client.create_container() print(f"Container '{CONTAINER_NAME}' created.") except Exception as e: if "ContainerAlreadyExists" in str(e): # Specific exception for idempotent create print(f"Container '{CONTAINER_NAME}' already exists.") else: raise # Upload data to a blob data = "Hello, Azure Blob Storage! This is a test blob." blob_client = container_client.get_blob_client(BLOB_NAME) blob_client.upload_blob(data, overwrite=True) print(f"Blob '{BLOB_NAME}' uploaded to container '{CONTAINER_NAME}'.") # Download the blob data downloaded_blob = blob_client.download_blob().readall() print(f"Downloaded blob content: {downloaded_blob.decode('utf-8')}") # Clean up (optional: delete the blob and container) # blob_client.delete_blob() # print(f"Blob '{BLOB_NAME}' deleted.") # container_client.delete_container() # print(f"Container '{CONTAINER_NAME}' deleted.") except Exception as ex: print(f"An error occurred: {ex}")