Azure Storage Blob Py

Azure Blob Storage SDK for Python. Use for uploading, downloading, listing blobs, managing containers, and blob lifecycle. Triggers: "blob storage", "BlobServiceClient", "ContainerClient", "BlobClient", "upload blob", "download blob".

Security Scan

The skill is an instruction-only, coherent guide for using the Azure Blob Storage Python SDK; its requirements and instructions match its stated purpose and there are no obvious incoherences or unexplained privileges.

Purpose & Capability

The name and description (Azure Blob Storage SDK for Python) match the SKILL.md content: examples for upload, download, list, delete, SAS generation, async clients, and performance tuning. No unrelated services, binaries, or environment requests are present.

Instruction Scope

Runtime instructions are limited to installing the Azure SDK (pip), instantiating clients, and performing blob operations and examples. File operations (open/read/write) are explicitly about blob upload/download and are expected. The document does not instruct reading arbitrary system files or sending data to unexpected external endpoints.

Install Mechanism

This is an instruction-only skill with no formal install spec, which is low risk. The SKILL.md tells users to run `pip install azure-storage-blob azure-identity` (PyPI). That is expected for a Python SDK but means the runtime must allow pip/network access to fetch packages; consider pinning versions in real deployments.

Credentials

The skill declares no required env vars but documents AZURE_STORAGE_ACCOUNT_NAME / AZURE_STORAGE_ACCOUNT_URL and shows use of DefaultAzureCredential and an account_key example for SAS generation. Those environment/credential suggestions are proportional to the task. One caveat: DefaultAzureCredential will attempt to use any ambient Azure credentials available in the runtime (environment variables, Azure CLI token, managed identity), so the agent using this skill could access whatever Azure identity is present — expected behavior but worth awareness.

Persistence & Privilege

The skill does not request persistent presence (always:false), does not modify other skills, and does not require system-wide privileges. Autonomous invocation is allowed by default but is not combined with other concerning flags.

Assessment

This skill is a straightforward usage guide for the Azure Blob Storage Python SDK and appears internally consistent. Before installing/using it: 1) Be aware it instructs pip installs (network access to PyPI) — in sensitive environments prefer pinned package versions or vetted wheels. 2) DefaultAzureCredential will use any Azure identity available to the runtime (env vars, CLI login, managed identity); ensure the agent runtime does not have unintended Azure permissions. 3) The SAS example uses an account key — never paste real long-lived account keys into untrusted channels; prefer short-lived SAS tokens or managed identities with least privilege. 4) If you plan to let the agent run this skill autonomously, confirm it should have the Azure access the runtime identity affords.

Azure Blob Storage SDK for Python

Client library for Azure Blob Storage — object storage for unstructured data.

Installation

pip install azure-storage-blob azure-identity

Environment Variables

AZURE_STORAGE_ACCOUNT_NAME=<your-storage-account>
# Or use full URL
AZURE_STORAGE_ACCOUNT_URL=https://<account>.blob.core.windows.net

Authentication

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

credential = DefaultAzureCredential()
account_url = "https://<account>.blob.core.windows.net"

blob_service_client = BlobServiceClient(account_url, credential=credential)

Client Hierarchy

Client	Purpose	Get From
BlobServiceClient	Account-level operations	Direct instantiation
ContainerClient	Container operations	blob_service_client.get_container_client()
BlobClient	Single blob operations	container_client.get_blob_client()

Core Workflow

Create Container

container_client = blob_service_client.get_container_client("mycontainer")
container_client.create_container()

Upload Blob

# From file path
blob_client = blob_service_client.get_blob_client(
    container="mycontainer",
    blob="sample.txt"
)

with open("./local-file.txt", "rb") as data:
    blob_client.upload_blob(data, overwrite=True)

# From bytes/string
blob_client.upload_blob(b"Hello, World!", overwrite=True)

# From stream
import io
stream = io.BytesIO(b"Stream content")
blob_client.upload_blob(stream, overwrite=True)

Download Blob

blob_client = blob_service_client.get_blob_client(
    container="mycontainer",
    blob="sample.txt"
)

# To file
with open("./downloaded.txt", "wb") as file:
    download_stream = blob_client.download_blob()
    file.write(download_stream.readall())

# To memory
download_stream = blob_client.download_blob()
content = download_stream.readall()  # bytes

# Read into existing buffer
stream = io.BytesIO()
num_bytes = blob_client.download_blob().readinto(stream)

List Blobs

container_client = blob_service_client.get_container_client("mycontainer")

# List all blobs
for blob in container_client.list_blobs():
    print(f"{blob.name} - {blob.size} bytes")

# List with prefix (folder-like)
for blob in container_client.list_blobs(name_starts_with="logs/"):
    print(blob.name)

# Walk blob hierarchy (virtual directories)
for item in container_client.walk_blobs(delimiter="/"):
    if item.get("prefix"):
        print(f"Directory: {item['prefix']}")
    else:
        print(f"Blob: {item.name}")

Delete Blob

blob_client.delete_blob()

# Delete with snapshots
blob_client.delete_blob(delete_snapshots="include")

Performance Tuning

# Configure chunk sizes for large uploads/downloads
blob_client = BlobClient(
    account_url=account_url,
    container_name="mycontainer",
    blob_name="large-file.zip",
    credential=credential,
    max_block_size=4 * 1024 * 1024,  # 4 MiB blocks
    max_single_put_size=64 * 1024 * 1024  # 64 MiB single upload limit
)

# Parallel upload
blob_client.upload_blob(data, max_concurrency=4)

# Parallel download
download_stream = blob_client.download_blob(max_concurrency=4)

SAS Tokens

from datetime import datetime, timedelta, timezone
from azure.storage.blob import generate_blob_sas, BlobSasPermissions

sas_token = generate_blob_sas(
    account_name="<account>",
    container_name="mycontainer",
    blob_name="sample.txt",
    account_key="<account-key>",  # Or use user delegation key
    permission=BlobSasPermissions(read=True),
    expiry=datetime.now(timezone.utc) + timedelta(hours=1)
)

# Use SAS token
blob_url = f"https://<account>.blob.core.windows.net/mycontainer/sample.txt?{sas_token}"

Blob Properties and Metadata

# Get properties
properties = blob_client.get_blob_properties()
print(f"Size: {properties.size}")
print(f"Content-Type: {properties.content_settings.content_type}")
print(f"Last modified: {properties.last_modified}")

# Set metadata
blob_client.set_blob_metadata(metadata={"category": "logs", "year": "2024"})

# Set content type
from azure.storage.blob import ContentSettings
blob_client.set_http_headers(
    content_settings=ContentSettings(content_type="application/json")
)

Async Client

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

async def upload_async():
    credential = DefaultAzureCredential()
    
    async with BlobServiceClient(account_url, credential=credential) as client:
        blob_client = client.get_blob_client("mycontainer", "sample.txt")
        
        with open("./file.txt", "rb") as data:
            await blob_client.upload_blob(data, overwrite=True)

# Download async
async def download_async():
    async with BlobServiceClient(account_url, credential=credential) as client:
        blob_client = client.get_blob_client("mycontainer", "sample.txt")
        
        stream = await blob_client.download_blob()
        data = await stream.readall()

Best Practices

1. Use DefaultAzureCredential instead of connection strings
2. Use context managers for async clients
3. Set overwrite=True explicitly when re-uploading
4. Use max_concurrency for large file transfers
5. Prefer readinto() over readall() for memory efficiency
6. Use walk_blobs() for hierarchical listing
7. Set appropriate content types for web-served blobs