Azure Blob Storage Skill

Azure Blob Storage Skill

Overview

This skill enables interaction with Azure Blob Storage and Azure Data Lake Storage Gen2 (ADLS Gen2). It provides capabilities for authenticating, listing containers/​blobs, reading blob content, and managing blob metadata.

When to Use

• User mentions Azure Blob Storage, Azure Storage, or ADLS
• Need to list containers or blobs in a storage account
• Read data files from blob storage (CSV, JSON, Parquet)
• Upload or download files to/​from blob storage
• Get blob metadata (size, last modified, content type)
• Work with hierarchical namespace (ADLS Gen2) directories

Prerequisites

Environment variables must be configured:

• AZURE_STORAGE_ACCOUNT_NAME: Storage account name
• AZURE_STORAGE_ACCOUNT_KEY: Storage account key OR
• AZURE_STORAGE_CONNECTION_STRING: Complete connection string

Authentication

Option 1: Account Name + Key

from azure.storage.blob import BlobServiceClient

account_name = os.getenv("AZURE_STORAGE_ACCOUNT_NAME")
account_key = os.getenv("AZURE_STORAGE_ACCOUNT_KEY")
account_url = f"https://{account_name}.blob.core.windows.net"

blob_service_client = BlobServiceClient(account_url, credential=account_key)

Option 2: Connection String

from azure.storage.blob import BlobServiceClient

connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
blob_service_client = BlobServiceClient.from_connection_string(connection_string)

Option 3: Managed Identity (for Azure-hosted apps)

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

account_name = os.getenv("AZURE_STORAGE_ACCOUNT_NAME")
account_url = f"https://{account_name}.blob.core.windows.net"
credential = DefaultAzureCredential()

blob_service_client = BlobServiceClient(account_url, credential=credential)

Common Operations

List Containers

from azure.storage.blob import BlobServiceClient

blob_service_client = BlobServiceClient(account_url, credential=account_key)

print("Containers in storage account:")
containers = blob_service_client.list_containers()
for container in containers:
    print(f"  - {container.name}")

List Blobs in a Container

container_client = blob_service_client.get_container_client("my-container")

print("Blobs in container 'my-container':")
blobs = container_client.list_blobs()
for blob in blobs:
    print(f"  - {blob.name} ({blob.size} bytes, modified: {blob.last_modified})")

Filter Blobs by Prefix (Directory-like)

# List all blobs in "2024/​sales/" path
blobs = container_client.list_blobs(name_starts_with="2024/​sales/")
for blob in blobs:
    print(f"  - {blob.name}")

Download Blob Content

blob_client = blob_service_client.get_blob_client(
    container="my-container",
    blob="data/​file.csv"
)

# Download to stream
download_stream = blob_client.download_blob()
content = download_stream.readall()
print(f"Downloaded {len(content)} bytes")

# Download to file
with open("local_file.csv", "wb") as file:
    download_stream = blob_client.download_blob()
    file.write(download_stream.readall())

Read CSV/​JSON/​Parquet from Blob

import pandas as pd
from io import BytesIO

blob_client = blob_service_client.get_blob_client(
    container="data-container",
    blob="sales/​2024/​sales.csv"
)

# Download blob content
download_stream = blob_client.download_blob()
content = download_stream.readall()

# Read with pandas
if blob.name.endswith('.csv'):
    df = pd.read_csv(BytesIO(content))
elif blob.name.endswith('.json'):
    df = pd.read_json(BytesIO(content))
elif blob.name.endswith('.parquet'):
    df = pd.read_parquet(BytesIO(content))

print(f"Loaded DataFrame with {len(df)} rows and {len(df.columns)} columns")

Get Blob Metadata

blob_client = blob_service_client.get_blob_client(
    container="my-container",
    blob="data/​file.parquet"
)

properties = blob_client.get_blob_properties()
print(f"Blob: {properties.name}")
print(f"Size: {properties.size} bytes")
print(f"Content Type: {properties.content_settings.content_type}")
print(f"Last Modified: {properties.last_modified}")
print(f"ETag: {properties.etag}")

Upload Blob

blob_client = blob_service_client.get_blob_client(
    container="my-container",
    blob="output/​result.csv"
)

# Upload from local file
with open("local_result.csv", "rb") as data:
    blob_client.upload_blob(data, overwrite=True)

# Upload from string/​bytes
content = "col1,col2\nvalue1,value2"
blob_client.upload_blob(content.encode('utf-8'), overwrite=True)

Check if Blob Exists

blob_client = blob_service_client.get_blob_client(
    container="my-container",
    blob="data/​file.csv"
)

if blob_client.exists():
    print("Blob exists")
else:
    print("Blob does not exist")

Error Handling

from azure.core.exceptions import ResourceNotFoundError, AzureError

try:
    blob_client = blob_service_client.get_blob_client(
        container="my-container",
        blob="data/​file.csv"
    )
    content = blob_client.download_blob().readall()
except ResourceNotFoundError:
    print("Blob not found - check container and blob name")
except AzureError as e:
    print(f"Azure error: {e.message}")
except Exception as e:
    print(f"Unexpected error: {str(e)}")

Best Practices

1. Use Specific Paths: Always specify full blob paths including container
2. Handle Large Files: For large files (>100MB), use streaming downloads
3. Batch Operations: When listing many blobs, use pagination
4. Connection Reuse: Reuse BlobServiceClient instance for multiple operations
5. Error Handling: Always wrap blob operations in try/​except blocks
6. Authentication: Prefer managed identity in production over account keys

Security Notes

• Never commit account keys or connection strings to version control
• Use environment variables or Azure Key Vault for credentials
• In production, use managed identities instead of account keys
• Apply least-privilege access using SAS tokens or Azure RBAC

Extensibility

This skill is designed for Azure Blob Storage. For other cloud providers:

• AWS S3: Create separate aws-s3 skill following similar pattern
• Google Cloud Storage: Create gcs-storage skill
• MinIO: Create minio skill for S3-compatible storage

Common Use Cases

1. Data Discovery: List containers and blobs to find datasets
2. Data Ingestion: Download files for processing or profiling
3. Data Export: Upload transformed data back to blob storage
4. Metadata Collection: Get file sizes, types, and modification dates
5. Directory Traversal: Use prefixes to navigate ADLS Gen2 directories

Troubleshooting

• Authentication Errors: Verify env variables are set correctly
• Permission Denied: Check storage account access policies and RBAC roles
• Blob Not Found: Confirm container and blob names (case-sensitive)
• Network Issues: Check firewall rules and network connectivity
• Large File Timeouts: Increase timeout or use chunked downloads