This tutorial provides a production-ready Python script for managing files in Azure Blob Storage directly from your command line. It's designed for develop
This tutorial provides a production-ready Python script for managing files in Azure Blob Storage directly from your command line. It's designed for developers, DevOps engineers, and sysadmins who need a quick, scriptable way to interact with their Azure Blob containers.
In many development and operations workflows, there's a recurring need to interact with cloud storage. Whether it's uploading application logs, downloading database backups, listing contents of a data lake, or cleaning up old files, manually navigating the Azure portal can be time-consuming and error-prone. Furthermore, integrating these operations into automated scripts or CI/CD pipelines requires a programmatic interface.
This script addresses the challenge of managing Azure Blob Storage assets efficiently from the command line. It provides a simple, unified interface to perform common operations like listing, uploading, downloading, and deleting blobs, making it ideal for automation, ad-hoc tasks, and quick data transfers without the overhead of a full-fledged SDK integration in a larger application.
The Azure Blob Storage File Manager script operates as a command-line interface (CLI) tool. Its architecture is straightforward:
argparse module to define various commands (list, upload, download, delete) and their respective arguments (e.g., container name, local file path, blob name). This allows for a flexible and intuitive user experience.AZURE_STORAGE_CONNECTION_STRING environment variable. If not provided, it falls back to using an account name and account key from AZURE_STORAGE_ACCOUNT_NAME and AZURE_STORAGE_ACCOUNT_KEY environment variables. This approach keeps sensitive credentials out of the codebase and allows for easy environment-specific configuration.azure-storage-blob Python SDK. A BlobManager class encapsulates the BlobServiceClient and provides methods for each storage operation. This abstracts away the underlying REST API calls and complex retry logic, offering a clean interface.BlobManager.BlobClient.upload_blob to transfer it to Azure.BlobClient.download_blob to fetch content and writes it to a local file in binary write mode.BlobClient.delete_blob to remove a specific blob.try-except blocks catch common Azure-specific exceptions (e.g., ResourceNotFoundError, ClientAuthenticationError) and general exceptions, providing user-friendly error messages and exiting with a non-zero status code for scripting purposes.This design pattern (CLI with a service manager class) makes the script modular, testable, and easy to extend.
#!/usr/bin/env python3
"""
Azure Blob Storage File Manager
This script provides command-line utilities to interact with Azure Blob Storage.
It supports listing blobs, uploading files, downloading blobs, and deleting blobs
from a specified container. Authentication is handled via environment variables.
"""
import os
import argparse
import logging
import sys
from typing import Optional, List
from pathlib import Path
# Azure Blob Storage specific imports
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
from azure.core.exceptions import ResourceNotFoundError, ClientAuthenticationError, HttpResponseError
# --- Configuration ---
# Configure logging for better output control
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO").upper(),
format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Environment variables for Azure Blob Storage authentication
# Prioritize connection string, then account name/key
AZURE_STORAGE_CONNECTION_STRING: Optional[str] = os.environ.get("AZURE_STORAGE_CONNECTION_STRING")
AZURE_STORAGE_ACCOUNT_NAME: Optional[str] = os.environ.get("AZURE_STORAGE_ACCOUNT_NAME")
AZURE_STORAGE_ACCOUNT_KEY: Optional[str] = os.environ.get("AZURE_STORAGE_ACCOUNT_KEY")
class BlobManager:
"""
Manages interactions with Azure Blob Storage.
Encapsulates the BlobServiceClient and provides methods for common operations.
"""
def __init__(self, container_name: str):
"""
Initializes the BlobManager with a container name and establishes a BlobServiceClient.
"""
self.container_name = container_name
self.blob_service_client: BlobServiceClient = self._get_blob_service_client()
logger.debug(f"Initialized BlobManager for container: {container_name}")
def _get_blob_service_client(self) -> BlobServiceClient:
"""
Establishes and returns a BlobServiceClient instance.
Authenticates using connection string or account name/key from environment variables.
"""
if AZURE_STORAGE_CONNECTION_STRING:
logger.debug("Attempting to connect using AZURE_STORAGE_CONNECTION_STRING.")
return BlobServiceClient.from_connection_string(AZURE_STORAGE_CONNECTION_STRING)
elif AZURE_STORAGE_ACCOUNT_NAME and AZURE_STORAGE_ACCOUNT_KEY:
logger.debug("Attempting to connect using AZURE_STORAGE_ACCOUNT_NAME and AZURE_STORAGE_ACCOUNT_KEY.")
account_url = f"https://{AZURE_STORAGE_ACCOUNT_NAME}.blob.core.windows.net"
return BlobServiceClient(account_url=account_url, credential=AZURE_STORAGE_ACCOUNT_KEY)
else:
logger.error("Azure Storage credentials not found. Please set AZURE_STORAGE_CONNECTION_STRING "
"or AZURE_STORAGE_ACCOUNT_NAME and AZURE_STORAGE_ACCOUNT_KEY environment variables.")
sys.exit(1)
def _get_container_client(self) -> ContainerClient:
"""
Returns a ContainerClient for the configured container.
"""
return self.blob_service_client.get_container_client(self.container_name)
def list_blobs(self, prefix: Optional[str] = None) -> List[str]:
"""
Lists blobs in the configured container, optionally filtered by a prefix.
Returns a list of blob names.
"""
logger.info(f"Listing blobs in container '{self.container_name}' with prefix '{prefix or ''}'...")
try:
container_client = self._get_container_client()
blob_list: List[str] = []
for blob in container_client.list_blobs(name_starts_with=prefix):
blob_list.append(blob.name)
logger.info(f"Found {len(blob_list)} blob(s).")
return blob_list
except ResourceNotFoundError:
logger.error(f"Container '{self.container_name}' not found.")
return []
except ClientAuthenticationError as e:
logger.error(f"Authentication failed: {e}")
sys.exit(1)
except HttpResponseError as e:
logger.error(f"Azure Blob Storage error during list operation: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"An unexpected error occurred during list operation: {e}")
sys.exit(1)
def upload_blob(self, local_file_path: Path, blob_name: Optional[str] = None, overwrite: bool = False) -> None:
"""
Uploads a local file to the configured container.
If blob_name is not provided, it defaults to the local file's name.
"""
if not local_file_path.is_file():
logger.error(f"Local file not found: {local_file_path}")
sys.exit(1)
target_blob_name = blob_name if blob_name else local_file_path.name
logger.info(f"Uploading '{local_file_path}' to '{self.container_name}/{target_blob_name}' (Overwrite: {overwrite})...")
try:
container_client = self._get_container_client()
# Ensure the container exists before attempting to upload
if not container_client.exists():
logger.warning(f"Container '{self.container_name}' does not exist. Attempting to create it.")
container_client.create_container()
logger.info(f"Container '{self.container_name}' created successfully.")
blob_client: BlobClient = container_client.get_blob_client(target_blob_name)
with open(local_file_path, "rb") as data:
blob_client.upload_blob(data, overwrite=overwrite)
logger.info(f"Successfully uploaded '{local_file_path}' to '{target_blob_name}'.")
except ResourceNotFoundError:
logger.error(f"Container '{self.container_name}' not found during upload.")
sys.exit(1)
except ClientAuthenticationError as e:
logger.error(f"Authentication failed: {e}")
sys.exit(1)
except HttpResponseError as e:
logger.error(f"Azure Blob Storage error during upload operation: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"An unexpected error occurred during upload operation: {e}")
sys.exit(1)
def download_blob(self, blob_name: str, output_path: Optional[Path] = None) -> None:
"""
Downloads a blob from the configured container to a local file.
If output_path is not provided, it defaults to the current directory with the blob's name.
"""
target_output_path = output_path if output_path else Path(blob_name)
logger.info(f"Downloading '{self.container_name}/{blob_name}' to '{target_output_path}'...")
try:
container_client = self._get_container_client()
blob_client: BlobClient = container_client.get_blob_client(blob_name)
# Ensure the blob exists before attempting to download
if not blob_client.exists():
logger.error(f"Blob '{blob_name}' not found in container '{self.container_name}'.")
sys.exit(1)
# Create parent directories if they don't exist
target_output_path.parent.mkdir(parents=True, exist_ok=True)
with open(target_output_path, "wb") as download_file:
download_stream = blob_client.download_blob()
download_file.write(download_stream.readall())
logger.info(f"Successfully downloaded '{blob_name}' to '{target_output_path}'.")
except ResourceNotFoundError:
logger.error(f"Container '{self.container_name}' not found during download.")
sys.exit(1)
except ClientAuthenticationError as e:
logger.error(f"Authentication failed: {e}")
sys.exit(1)
except HttpResponseError as e:
logger.error(f"Azure Blob Storage error during download operation: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"An unexpected error occurred during download operation: {e}")
sys.exit(1)
def delete_blob(self, blob_name: str) -> None:
"""
Deletes a blob from the configured container.
"""
logger.info(f"Deleting blob '{blob_name}' from container '{self.container_name}'...")
try:
container_client = self._get_container_client()
blob_client: BlobClient = container_client.get_blob_client(blob_name)
if not blob_client.exists():
logger.warning(f"Blob '{blob_name}' does not exist in container '{self.container_name}'. No action needed.")
return
blob_client.delete_blob()
logger.info(f"Successfully deleted blob '{blob_name}'.")
except ResourceNotFoundError:
logger.error(f"Container '{self.container_name}' not found during delete operation.")
sys.exit(1)
except ClientAuthenticationError as e:
logger.error(f"Authentication failed: {e}")
sys.exit(1)
except HttpResponseError as e:
logger.error(f"Azure Blob Storage error during delete operation: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"An unexpected error occurred during delete operation: {e}")
sys.exit(1)
def main():
"""
Main function to parse arguments and execute blob operations.
"""
parser = argparse.ArgumentParser(
description="Azure Blob Storage File Manager. Manage files (blobs) in Azure containers."
)
# Global argument for container name, required for all operations
parser.add_argument('--container', required=True, help='Name of the Azure Blob container.')
# Subparsers for different commands (list, upload, download, delete)
subparsers = parser.add_subparsers(dest='command', required=True, help='Available commands.')
# --- List Command ---
list_parser = subparsers.add_parser('list', help='List blobs in a container.')
list_parser.add_argument('--prefix', help='Filter blobs by a specific prefix (e.g., "logs/").')
# --- Upload Command ---
upload_parser = subparsers.add_parser('upload', help='Upload a file to a container.')
upload_parser.add_argument('--local-path', type=Path, required=True,
help='Path to the local file to upload.')
upload_parser.add_argument('--blob-name', help='Name of the blob in Azure (defaults to local file name).')
upload_parser.add_argument('--overwrite', action='store_true',
help='Overwrite blob if it already exists.')
# --- Download Command ---
download_parser = subparsers.add_parser('download', help='Download a blob from a container.')
download_parser.add_argument('--blob-name', required=True,
help='Name of the blob in Azure to download.')
download_parser.add_argument('--output-path', type=Path,
help='Local path to save the downloaded file (defaults to current directory with blob name).')
# --- Delete Command ---
delete_parser = subparsers.add_parser('delete', help='Delete a blob from a container.')
delete_parser.add_argument('--blob-name', required=True,
help='Name of the blob in Azure to delete.')
args = parser.parse_args()
# Initialize BlobManager
blob_manager = BlobManager(args.container)
# Execute the chosen command
if args.command == 'list':
blobs = blob_manager.list_blobs(args.prefix)
if blobs:
logger.info("\n--- Blobs in container ---")
for blob in blobs:
print(blob)
logger.info("--------------------------")
else:
logger.info("No blobs found or container is empty.")
elif args.command == 'upload':
blob_manager.upload_blob(args.local_path, args.blob_name, args.overwrite)
elif args.command == 'download':
blob_manager.download_blob(args.blob_name, args.output_path)
elif args.command == 'delete':
blob_manager.delete_blob(args.blob_name)
else:
logger.error(f"Unknown command: {args.command}")
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()
The script relies on environment variables for Azure Blob Storage authentication. You need to set one of the following combinations before running the script:
AZURE_STORAGE_CONNECTION_STRING: The full connection string for your Azure Storage account.key1 or key2.