azpaddypy 0.9.76

Azure cloud services SDK with Storage (blob, file share, queue), Key Vault, Cosmos DB, OpenTelemetry tracing, and builder patterns.

azpaddypy

Azure cloud services SDK with Storage (blob, file share, queue), Key Vault, Cosmos DB, OpenTelemetry tracing, and builder patterns.

Designed for Python 3.11+ running in Dockerized Azure Function Apps and Web Apps.

Installation

uv add azpaddypy

Quick Start

from azpaddypy import AzureStorage, AzureIdentity, create_azure_storage

# Factory function (cached instances, auto-creates identity)
storage = create_azure_storage(
    account_url="https://myaccount.blob.core.windows.net/",
    service_name="my_service",
)

# Or explicit identity
identity = AzureIdentity(service_name="my_service")
storage = AzureStorage(
    account_url="https://myaccount.blob.core.windows.net/",
    azure_identity=identity,
    enable_file_storage=True,
)

Storage Operations

Blob Storage

# Upload
storage.upload_blob(
    container_name="documents",
    blob_name="report.pdf",
    data=pdf_bytes,
    content_type="application/pdf",
    metadata={"author": "team"},
)

# Download (returns None if not found)
data = storage.download_blob(container_name="documents", blob_name="report.pdf")

# Upload and get SAS URL
sas_url = storage.upload_blob_with_sas(
    container_name="documents",
    blob_name="report.pdf",
    data=pdf_bytes,
    sas_permission="r",
    sas_expiry_delta=timedelta(hours=3),
)

# List, exists, delete
blobs = storage.list_blobs(container_name="documents", name_starts_with="reports/")
exists = storage.blob_exists(container_name="documents", blob_name="report.pdf")
storage.delete_blob(container_name="documents", blob_name="report.pdf")

# Metadata upsert (merges with existing)
storage.upsert_blob_metadata(
    container_name="documents",
    blob_name="report.pdf",
    metadata={"status": "processed"},
)

# SAS token generation
blob_sas = storage.get_blob_sas(container_name="docs", blob_name="file.pdf")
container_sas = storage.get_container_sas(container_name="docs", permission="r")

File Share Storage

Requires enable_file_storage=True. Uses Azure File Shares (SMB/NFS), not blob storage.

storage = AzureStorage(
    account_url="https://myaccount.blob.core.windows.net/",
    azure_identity=identity,
    enable_file_storage=True,
)

# Upload (auto-creates parent directories)
storage.upload_share_file(
    share_name="myshare",
    file_path="reports/2026/q1.pdf",
    data=pdf_bytes,
    content_type="application/pdf",
)

# Download (returns None if not found)
data = storage.download_share_file(share_name="myshare", file_path="reports/2026/q1.pdf")

# List files and directories
items = storage.list_share_files(share_name="myshare", directory_path="reports/2026")
# Returns: [{"name": "q1.pdf", "is_directory": False, "size": 1024}, ...]

# Exists, properties, delete
exists = storage.share_file_exists(share_name="myshare", file_path="reports/2026/q1.pdf")
props = storage.get_share_file_properties(share_name="myshare", file_path="reports/2026/q1.pdf")
storage.delete_share_file(share_name="myshare", file_path="reports/2026/q1.pdf")

# Directory management
storage.create_share_directory(share_name="myshare", directory_path="reports/2026/q2")
storage.delete_share_directory(share_name="myshare", directory_path="reports/2026/q2")

# Metadata upsert (merges with existing)
storage.upsert_share_file_metadata(
    share_name="myshare",
    file_path="reports/2026/q1.pdf",
    metadata={"reviewed": "true"},
)

Queue Storage

# Send
storage.send_message(
    queue_name="tasks",
    content='{"task": "process"}',
    visibility_timeout=30,
    time_to_live=3600,
)

# Receive
messages = storage.receive_messages(queue_name="tasks", messages_per_page=5)
for msg in messages:
    print(msg["id"], msg["content"])
    storage.delete_message(
        queue_name="tasks",
        message_id=msg["id"],
        pop_receipt=msg["pop_receipt"],
    )

Builder Pattern

For complex multi-resource setups:

from azpaddypy.builder import AzureManagementBuilder, AzureResourceBuilder
from azpaddypy.builder.directors import ConfigurationSetupDirector

# One-liner setup with director
config = ConfigurationSetupDirector.default_setup(
    service_name="my_app",
    service_version="1.0.0",
)

# Or step-by-step with builders
mgmt = (
    AzureManagementBuilder()
    .with_logger(service_name="my_app")
    .with_identity()
    .with_keyvault(vault_url="https://myvault.vault.azure.net/")
    .build()
)

resources = (
    AzureResourceBuilder(mgmt, env_config)
    .with_storage("default", enable_blob=True, enable_queue=True)
    .with_storage("archive", account_url="https://archive.blob.core.windows.net/", enable_file=True)
    .build()
)

storage = resources.get_storage("default")
archive = resources.get_storage("archive")

Key Vault

from azpaddypy import AzureKeyVault, create_azure_keyvault

kv = create_azure_keyvault(
    vault_url="https://myvault.vault.azure.net/",
    service_name="my_service",
)

secret = kv.get_secret("database-connection-string")

Observability

All operations include OpenTelemetry spans and structured logging via Application Insights.

storage = AzureStorage(
    account_url="https://myaccount.blob.core.windows.net/",
    azure_identity=identity,
    connection_string="InstrumentationKey=...",  # App Insights
)

# Correlation tracking across distributed calls
storage.set_correlation_id("request-abc-123")

Feature Flags

Enable only the storage services you need:

Flag	Default	Service
enable_blob_storage	True	BlobServiceClient
enable_file_storage	False	ShareServiceClient (requires token_intent="backup" RBAC)
enable_queue_storage	True	QueueServiceClient

Dependencies

• azure-storage-blob - Blob operations
• azure-storage-file-share - File share operations
• azure-storage-queue - Queue operations
• azure-identity - Credential management
• azure-keyvault-secrets / keys / certificates - Key Vault
• azure-cosmos - Cosmos DB
• azure-monitor-opentelemetry - Telemetry