Overview
Azure Blob Storage is Microsoft’s object storage solution for the cloud. Multi-Cloud Manager provides comprehensive APIs for managing storage accounts, containers, and blobs.Storage Account Operations
List Storage Accounts
Retrieve all storage accounts across your Azure subscriptions with detailed metrics and properties. Endpoint:GET /api/azure/storage/accounts
Authentication: Requires valid Azure access token in session
Response:
{
"value": [
{
"name": "mystorageaccount",
"resourceGroup": "my-resource-group",
"location": "westeurope",
"sku": "Standard_LRS",
"accessTier": "Hot",
"storageType": "Blob + Files + Queues + Tables",
"httpsOnly": true,
"usage": "2.45 GiB",
"publicAccess": "Wyłączony",
"subscriptionId": "sub-123-456",
"Keys": "account-key-here"
}
]
}
from azure.mgmt.storage import StorageManagementClient
from azure.mgmt.monitor import MonitorManagementClient
from datetime import datetime, timedelta
def list_storage_accounts():
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
credential = FlaskCredential()
sub_client = SubscriptionClient(credential)
items = []
try:
for sub in sub_client.subscriptions.list():
subscription_id = sub.subscription_id
# Create storage and monitor clients
storage_acc_client = StorageManagementClient(
credential, subscription_id, api_version="2025-06-01"
)
resource_client = ResourceManagementClient(credential, subscription_id)
monitor_client = MonitorManagementClient(credential, subscription_id)
for rg in resource_client.resource_groups.list():
accounts = storage_acc_client.storage_accounts.list_by_resource_group(rg.name)
for acc in accounts:
keys = storage_acc_client.storage_accounts.list_keys(rg.name, acc.name)
props = storage_acc_client.storage_accounts.get_properties(rg.name, acc.name)
# Fetch usage metrics
usage_str = "N/A"
try:
metric_result = monitor_client.metrics.list(
resource_uri=props.id,
timespan=f"{datetime.utcnow() - timedelta(days=1)}/{datetime.utcnow()}",
interval="PT1H",
metricnames="UsedCapacity",
aggregation="Average"
)
if metric_result.value and len(metric_result.value) > 0:
timeseries = metric_result.value[0].timeseries
if timeseries and len(timeseries) > 0:
data = timeseries[0].data
if data and len(data) > 0:
capacity_bytes = data[-1].average
if capacity_bytes is not None:
# Format bytes to human-readable
if capacity_bytes > (1024**4):
usage_str = f"{capacity_bytes / (1024**4):.2f} TiB"
elif capacity_bytes > (1024**3):
usage_str = f"{capacity_bytes / (1024**3):.2f} GiB"
elif capacity_bytes > (1024**2):
usage_str = f"{capacity_bytes / (1024**2):.2f} MiB"
else:
usage_str = f"{capacity_bytes:.0f} Bytes"
except Exception as e:
print(f"[ERROR] Metric for {acc.name}: {e}")
usage_str = f"Error: {str(e)}"
items.append({
"name": acc.name,
"resourceGroup": rg.name,
"location": acc.location,
"Keys": keys.keys[0].value,
"sku": props.sku.name,
"accessTier": props.access_tier,
"storageType": classify_storage_type(props.kind).value,
"httpsOnly": props.enable_https_traffic_only,
"subscriptionId": subscription_id,
"usage": usage_str,
"publicAccess": "Włączony" if props.allow_blob_public_access else "Wyłączony"
})
return jsonify({"value": items})
except Exception as e:
return jsonify({"error": str(e)}), 500
The
classify_storage_type() helper function maps Azure storage kind values to human-readable types (storage.py:21-31).Create Storage Account
Create a new Azure storage account with custom configuration. Endpoint:POST /api/azure/storage/accounts
Request Body:
{
"subscriptionId": "sub-123-456",
"rgName": "my-resource-group",
"accountName": "mystorageaccount",
"location": "westeurope",
"sku": "Standard_LRS",
"kind": "StorageV2",
"accessTier": "Hot",
"enable_https_traffic_only": true
}
def create_storage_account():
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
subscription_id = data.get("subscriptionId")
rg_name = data.get("rgName")
account_name = data.get("accountName")
location = data.get("location", "westeurope")
sku = {"name": data.get("sku", "Standard_LRS")}
kind = data.get("kind", "StorageV2")
access_tier = data.get("accessTier", "Hot")
enable_https_traffic_only = data.get("enable_https_traffic_only", True)
if not all([subscription_id, rg_name, location, account_name, sku, kind, access_tier, enable_https_traffic_only]):
return jsonify({"error": "Brak wymaganych danych"}), 400
try:
credential = FlaskCredential()
storage_client = StorageManagementClient(credential, subscription_id)
result = storage_client.storage_accounts.begin_create(
resource_group_name=rg_name,
account_name=account_name,
parameters={
"location": location,
"sku": sku,
"kind": kind,
"access_tier": access_tier,
"enable_https_traffic_only": enable_https_traffic_only
}
).result()
return jsonify({
"message": f"Storage account '{account_name}' utworzony pomyślnie",
"storageAccID": result.id
}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
Storage account names must be globally unique, lowercase, and 3-24 characters. They can only contain letters and numbers.
Delete Storage Account
Endpoint:DELETE /api/azure/storage/accounts
Request Body:
{
"subscriptionId": "sub-123-456",
"resourceGroup": "my-resource-group",
"accountName": "mystorageaccount"
}
def delete_storage_account():
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
subscription_id = data.get("subscriptionId")
rg_name = data.get("resourceGroup")
account_name = data.get("accountName")
if not all([subscription_id, rg_name, account_name]):
return jsonify({"error": "Brak wymaganych danych"}), 400
try:
credential = FlaskCredential()
storage_client = StorageManagementClient(credential, subscription_id)
storage_client.storage_accounts.delete(
resource_group_name=rg_name,
account_name=account_name
)
return jsonify({"message": f"Storage account '{account_name}' został usunięty"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
Container Operations
List Containers
List all blob containers in a storage account. Endpoint:GET /api/azure/storage/containers
Request Body:
{
"accountName": "mystorageaccount",
"accountKey": "storage-account-key"
}
from azure.storage.blob import BlobServiceClient
def list_blob_containers(storage_account_id):
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
account_name = data.get("accountName")
account_key = data.get("accountKey")
if not all([account_name, account_key]):
return jsonify({"error": "Brak danych konta"}), 400
try:
blob_service = BlobServiceClient(
account_url=f"https://{account_name}.blob.core.windows.net",
credential=account_key
)
containers = blob_service.list_containers()
result = [{
"name": c.name,
"last_modified": c.last_modified.isoformat()
} for c in containers]
return jsonify({"value": result})
except Exception as e:
return jsonify({"error": str(e)}), 500
Create Container
Endpoint:POST /api/azure/storage/containers
Request Body:
{
"accountName": "mystorageaccount",
"accountKey": "storage-account-key",
"containerName": "my-container"
}
def create_blob_container(storage_account_id):
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
account_name = data.get("accountName")
account_key = data.get("accountKey")
container_name = data.get("containerName")
if not all([account_name, account_key, container_name]):
return jsonify({"error": "Brak wymaganych danych"}), 400
try:
blob_service = BlobServiceClient(
account_url=f"https://{account_name}.blob.core.windows.net",
credential=account_key
)
blob_service.create_container(container_name)
return jsonify({"message": f"Kontener '{container_name}' utworzony"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
Delete Container
Endpoint:DELETE /api/azure/storage/containers
Implementation (storage.py:244-265):
def delete_blob_container(storage_account_id):
if "access_token" not in session:
return jsonify({"error": "Unauthorized"}), 401
data = request.get_json()
account_name = data.get("accountName")
account_key = data.get("accountKey")
container_name = data.get("containerName")
if not all([account_name, account_key, container_name]):
return jsonify({"error": "Brak wymaganych danych"}), 400
try:
blob_service = BlobServiceClient(
account_url=f"https://{account_name}.blob.core.windows.net",
credential=account_key
)
blob_service.delete_container(container_name)
return jsonify({"message": f"Kontener '{container_name}' usunięty"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
List Blobs
List all blobs in a container. Implementation (storage.py:268-281):def list_blobs(storage_account_id):
data = request.get_json()
account_name = data.get("accountName")
account_key = data.get("accountKey")
container_name = data.get("containerName")
blob_service = BlobServiceClient(
account_url=f"https://{account_name}.blob.core.windows.net",
credential=account_key
)
container_client = blob_service.get_container_client(container_name)
blobs = container_client.list_blobs()
result = [{
"name": b.name,
"size": b.size,
"last_modified": b.last_modified.isoformat()
} for b in blobs]
return jsonify({"value": result})
Next Steps
Blob Management
Learn how to upload, download, and delete blobs