azure-storage

import os from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient # Recommended: Use the modern azure-storage-blob package # Ensure AZURE_STORAGE_CONNECTION_STRING or AZURE_STORAGE_ACCOUNT_NAME is set # Or configure Azure Identity credentials for DefaultAzureCredential try: # Option 1: Using connection string (simpler for quickstart) connection_string = os.environ.get('AZURE_STORAGE_CONNECTION_STRING') if connection_string: blob_service_client = BlobServiceClient.from_connection_string(connection_string) else: # Option 2: Using DefaultAzureCredential (recommended for production) account_url = os.environ.get('AZURE_STORAGE_ACCOUNT_URL', 'https://<your_account_name>.blob.core.windows.net') if not account_url.startswith('https://') and 'AZURE_STORAGE_ACCOUNT_NAME' in os.environ: account_name = os.environ['AZURE_STORAGE_ACCOUNT_NAME'] account_url = f'https://{account_name}.blob.core.windows.net' credential = DefaultAzureCredential() blob_service_client = BlobServiceClient(account_url, credential=credential) container_name = 'mytestcontainer' blob_name = 'mytestblob.txt' local_file_name = 'sample.txt' data = 'Hello, Azure Blob Storage!' print(f"Creating container: {container_name}") container_client = blob_service_client.get_container_client(container_name) try: container_client.create_container() except Exception as e: if 'ContainerAlreadyExists' not in str(e): raise print(f"Container '{container_name}' already exists.") with open(local_file_name, 'w') as file: file.write(data) print(f"Uploading blob: {blob_name}") with open(local_file_name, 'rb') as data_file: container_client.upload_blob(name=blob_name, data=data_file, overwrite=True) print(f"Listing blobs in '{container_name}':") for blob in container_client.list_blobs(): print(f" - {blob.name}") print(f"Downloading blob: {blob_name}") download_blob_client = container_client.get_blob_client(blob_name) download_data = download_blob_client.download_blob().readall() print(f"Downloaded content: {download_data.decode('utf-8')}") # Clean up # print(f"Deleting blob: {blob_name}") # container_client.delete_blob(blob_name) # print(f"Deleting container: {container_name}") # container_client.delete_container() except Exception as ex: print(f"Error: {ex}") print("Please ensure environment variables (AZURE_STORAGE_CONNECTION_STRING or AZURE_STORAGE_ACCOUNT_URL/NAME and Azure Identity vars) are set correctly.")