Microsoft Azure Storage Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Management of Azure Blob storage services including blob service configuration, container management, and blob inventory policies. Blob storage provides scalable object storage for unstructured data.
Configure blob service properties including CORS rules, delete retention policies, and versioning settings.
class BlobServicesOperations:
def get_service_properties(
self,
resource_group_name: str,
account_name: str
) -> BlobServiceProperties:
"""
Gets the properties of a storage account's Blob service.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
Returns:
BlobServiceProperties with current configuration
"""
def set_service_properties(
self,
resource_group_name: str,
account_name: str,
parameters: BlobServiceProperties
) -> BlobServiceProperties:
"""
Sets the properties of a storage account's Blob service.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- parameters: Blob service properties to set
Returns:
Updated BlobServiceProperties
"""
def list(
self,
resource_group_name: str,
account_name: str
) -> BlobServiceItems:
"""
List blob services for a storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
Returns:
BlobServiceItems containing the blob service
"""Usage example:
from azure.mgmt.storage.models import (
BlobServiceProperties, DeleteRetentionPolicy,
RestorePolicyProperties, CorsRules, CorsRule
)
# Get current blob service properties
blob_service = client.blob_services.get_service_properties(
resource_group_name="my-resource-group",
account_name="mystorageaccount123"
)
# Configure blob service with delete retention and CORS
cors_rule = CorsRule(
allowed_origins=["https://example.com"],
allowed_methods=["GET", "POST"],
allowed_headers=["*"],
exposed_headers=["*"],
max_age_in_seconds=3600
)
blob_properties = BlobServiceProperties(
properties=BlobServicePropertiesProperties(
delete_retention_policy=DeleteRetentionPolicy(
enabled=True,
days=30
),
restore_policy=RestorePolicyProperties(
enabled=True,
days=7,
minimum_days_to_retain=1
),
cors=CorsRules(cors_rules=[cors_rule]),
is_versioning_enabled=True,
change_feed=ChangeFeed(enabled=True, retention_in_days=90)
)
)
updated_service = client.blob_services.set_service_properties(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
parameters=blob_properties
)Create, manage, and configure blob containers with access policies, metadata, and lease operations.
class BlobContainersOperations:
def create(
self,
resource_group_name: str,
account_name: str,
container_name: str,
blob_container: BlobContainer
) -> BlobContainer:
"""
Creates a new container under the specified account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container (3-63 chars, lowercase)
- blob_container: Container properties and configuration
Returns:
Created BlobContainer
"""
def update(
self,
resource_group_name: str,
account_name: str,
container_name: str,
blob_container: BlobContainer
) -> BlobContainer:
"""
Updates container properties or metadata.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
- blob_container: Updated container properties
Returns:
Updated BlobContainer
"""
def get(
self,
resource_group_name: str,
account_name: str,
container_name: str
) -> BlobContainer:
"""
Gets properties of a specified container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
Returns:
BlobContainer with properties
"""
def delete(
self,
resource_group_name: str,
account_name: str,
container_name: str
) -> None:
"""
Deletes the specified container and any blobs it contains.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
"""
def list(
self,
resource_group_name: str,
account_name: str,
maxpagesize: Optional[str] = None,
filter: Optional[str] = None,
include: Optional[ListContainersInclude] = None
) -> ItemPaged[ListContainerItem]:
"""
Lists all containers in a storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- maxpagesize: Maximum number of results per page
- filter: OData filter expression
- include: Include additional properties (deleted containers, metadata)
Returns:
Paginated list of ListContainerItem objects
"""Usage example:
from azure.mgmt.storage.models import (
BlobContainer, PublicAccess, BlobContainerProperties
)
# Create a private container
container = BlobContainer(
properties=BlobContainerProperties(
public_access=PublicAccess.NONE,
metadata={"Department": "Finance", "Project": "Reports"}
)
)
created_container = client.blob_containers.create(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
container_name="private-documents",
blob_container=container
)
# Create a public container for blob access
public_container = BlobContainer(
properties=BlobContainerProperties(
public_access=PublicAccess.BLOB,
metadata={"Type": "Public", "Purpose": "Static Content"}
)
)
client.blob_containers.create(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
container_name="public-assets",
blob_container=public_container
)
# List all containers
containers = list(client.blob_containers.list(
resource_group_name="my-resource-group",
account_name="mystorageaccount123"
))
for container in containers:
print(f"Container: {container.name}, Public Access: {container.public_access}")Manage stored access policies for fine-grained container permissions.
def set_legal_hold(
self,
resource_group_name: str,
account_name: str,
container_name: str,
legal_hold: LegalHold
) -> LegalHold:
"""
Sets legal hold tags on a container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
- legal_hold: Legal hold configuration
Returns:
Applied LegalHold configuration
"""
def clear_legal_hold(
self,
resource_group_name: str,
account_name: str,
container_name: str,
legal_hold: LegalHold
) -> LegalHold:
"""
Clears legal hold tags from a container.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
- legal_hold: Legal hold tags to clear
Returns:
Updated LegalHold configuration
"""Acquire, break, and manage leases on containers for exclusive access control.
def lease(
self,
resource_group_name: str,
account_name: str,
container_name: str,
parameters: LeaseContainerRequest
) -> LeaseContainerResponse:
"""
Leases a container for delete operations.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- container_name: Name of the blob container
- parameters: Lease operation parameters
Returns:
LeaseContainerResponse with lease information
"""Usage example:
from azure.mgmt.storage.models import (
LeaseContainerRequest, LeaseContainerRequestEnum, LegalHold
)
# Acquire a lease on container
lease_request = LeaseContainerRequest(
action=LeaseContainerRequestEnum.ACQUIRE,
lease_duration=30 # seconds, -1 for infinite
)
lease_response = client.blob_containers.lease(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
container_name="private-documents",
parameters=lease_request
)
print(f"Lease ID: {lease_response.lease_id}")
# Set legal hold on container
legal_hold = LegalHold(
tags=["LegalCase2024", "Compliance"],
allow_protected_append_writes_all=False
)
client.blob_containers.set_legal_hold(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
container_name="legal-documents",
legal_hold=legal_hold
)Configure and manage blob inventory policies for automated reporting on blob data.
class BlobInventoryPoliciesOperations:
def create_or_update(
self,
resource_group_name: str,
account_name: str,
blob_inventory_policy_name: BlobInventoryPolicyName,
properties: BlobInventoryPolicy
) -> BlobInventoryPolicy:
"""
Sets the blob inventory policy for the specified storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- blob_inventory_policy_name: Name of the blob inventory policy
- properties: Blob inventory policy properties
Returns:
Created or updated BlobInventoryPolicy
"""
def get(
self,
resource_group_name: str,
account_name: str,
blob_inventory_policy_name: BlobInventoryPolicyName
) -> BlobInventoryPolicy:
"""
Gets the blob inventory policy associated with the specified storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- blob_inventory_policy_name: Name of the blob inventory policy
Returns:
BlobInventoryPolicy configuration
"""
def delete(
self,
resource_group_name: str,
account_name: str,
blob_inventory_policy_name: BlobInventoryPolicyName
) -> None:
"""
Deletes the blob inventory policy associated with the specified storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
- blob_inventory_policy_name: Name of the blob inventory policy
"""
def list(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[ListBlobInventoryPolicy]:
"""
Gets the blob inventory policies for the specified storage account.
Parameters:
- resource_group_name: Name of the resource group
- account_name: Name of the storage account
Returns:
Paginated list of blob inventory policies
"""Usage example:
from azure.mgmt.storage.models import (
BlobInventoryPolicy, BlobInventoryPolicyDefinition,
BlobInventoryPolicyRule, BlobInventoryPolicyFilter,
BlobInventoryPolicyName, InventoryRuleType, Format
)
# Create blob inventory policy
inventory_rule = BlobInventoryPolicyRule(
enabled=True,
name="DailyInventory",
destination="inventory-reports",
definition=BlobInventoryPolicyDefinition(
format=Format.CSV,
schedule="Daily",
object_type=InventoryRuleType.BLOB,
schema_fields=[
"Name", "Content-Length", "LastModified",
"Content-Type", "AccessTier", "BlobType"
],
filters=BlobInventoryPolicyFilter(
prefix_match=["data/", "logs/"],
blob_types=["blockBlob", "appendBlob"]
)
)
)
inventory_policy = BlobInventoryPolicy(
properties=BlobInventoryPolicyProperties(
policy=BlobInventoryPolicySchema(
enabled=True,
type_="Inventory",
rules=[inventory_rule]
)
)
)
created_policy = client.blob_inventory_policies.create_or_update(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
blob_inventory_policy_name=BlobInventoryPolicyName.DEFAULT,
properties=inventory_policy
)class BlobServiceProperties:
"""Blob service properties configuration."""
id: str
name: str
type_: str
properties: BlobServicePropertiesProperties
class BlobContainer:
"""Blob container resource."""
id: str
name: str
type_: str
etag: str
properties: BlobContainerProperties
class BlobContainerProperties:
"""Properties of a blob container."""
public_access: PublicAccess
last_modified_time: datetime
lease_status: LeaseStatus
lease_state: LeaseState
lease_duration: LeaseDuration
metadata: Dict[str, str]
immutability_policy: ImmutabilityPolicyProperties
legal_hold: LegalHoldProperties
has_legal_hold: bool
has_immutability_policy: bool
class ListContainerItem:
"""Container item in list results."""
name: str
deleted: bool
version: str
properties: BlobContainerProperties
class DeleteRetentionPolicy:
"""Delete retention policy configuration."""
enabled: bool
days: int
allow_permanent_delete: bool
class BlobInventoryPolicy:
"""Blob inventory policy resource."""
id: str
name: str
type_: str
system_data: SystemData
properties: BlobInventoryPolicyProperties
class PublicAccess(str, Enum):
"""Container public access levels."""
NONE = "None"
BLOB = "Blob"
CONTAINER = "Container"
class LeaseStatus(str, Enum):
"""Container lease status."""
LOCKED = "Locked"
UNLOCKED = "Unlocked"
class LeaseState(str, Enum):
"""Container lease state."""
AVAILABLE = "Available"
LEASED = "Leased"
EXPIRED = "Expired"
BREAKING = "Breaking"
BROKEN = "Broken"Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-storage