Skip to main content

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"
    }
  ]
}
Implementation (storage.py:34-131):
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
}
Implementation (storage.py:133-168):
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"
}
Implementation (storage.py:170-195):
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"
}
Implementation (storage.py:197-219):
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"
}
Implementation (storage.py:221-242):
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

Build docs developers (and LLMs) love