Microsoft Azure Storage Management Client Library for Python
npx @tessl/cli install tessl/pypi-azure-mgmt-storage@23.0.0Microsoft Azure Storage Management Client Library for Python provides comprehensive Python client library for managing Azure Storage services through the Azure Resource Manager APIs. It enables developers to programmatically create, configure, and manage Azure Storage accounts, blob containers, file shares, queues, and tables with full support for Azure's storage management features including access policies, encryption settings, network rules, and compliance configurations.
pip install azure-mgmt-storagefrom azure.mgmt.storage import StorageManagementClient
from azure.identity import DefaultAzureCredentialAsync version:
from azure.mgmt.storage.aio import StorageManagementClient
from azure.identity.aio import DefaultAzureCredentialfrom azure.mgmt.storage import StorageManagementClient
from azure.identity import DefaultAzureCredential
# Create client
credential = DefaultAzureCredential()
client = StorageManagementClient(
credential=credential,
subscription_id="your-subscription-id"
)
# List storage accounts
accounts = list(client.storage_accounts.list())
for account in accounts:
print(f"Account: {account.name}, Location: {account.location}")
# Create a storage account
from azure.mgmt.storage.models import StorageAccountCreateParameters, Sku, Kind
storage_account_params = StorageAccountCreateParameters(
sku=Sku(name="Standard_LRS"),
kind=Kind.STORAGE_V2,
location="East US"
)
# Begin long-running operation
poller = client.storage_accounts.begin_create(
resource_group_name="my-resource-group",
account_name="mystorageaccount123",
parameters=storage_account_params
)
# Wait for completion
account = poller.result()
print(f"Created storage account: {account.name}")The Azure Storage Management API is organized around resource management operations:
Both synchronous and asynchronous clients are available, with identical API surfaces but different execution models.
Complete lifecycle management of Azure Storage accounts including creation, deletion, configuration updates, key management, and failover operations.
class StorageAccountsOperations:
def check_name_availability(
self,
account_name: StorageAccountCheckNameAvailabilityParameters
) -> CheckNameAvailabilityResult
def begin_create(
self,
resource_group_name: str,
account_name: str,
parameters: StorageAccountCreateParameters
) -> LROPoller[StorageAccount]
def delete(
self,
resource_group_name: str,
account_name: str
) -> None
def get_properties(
self,
resource_group_name: str,
account_name: str
) -> StorageAccount
def list(self) -> ItemPaged[StorageAccount]
def list_keys(
self,
resource_group_name: str,
account_name: str
) -> StorageAccountListKeysResultManagement of blob services, containers, and blob inventory policies with support for access policies, lifecycle management, and container-level operations.
class BlobServicesOperations:
def get_service_properties(
self,
resource_group_name: str,
account_name: str
) -> BlobServiceProperties
def set_service_properties(
self,
resource_group_name: str,
account_name: str,
parameters: BlobServiceProperties
) -> BlobServiceProperties
class BlobContainersOperations:
def create(
self,
resource_group_name: str,
account_name: str,
container_name: str,
blob_container: BlobContainer
) -> BlobContainer
def list(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[ListContainerItem]Azure Files service and file share management including SMB/NFS configuration, access tier management, and snapshot operations.
class FileServicesOperations:
def get_service_properties(
self,
resource_group_name: str,
account_name: str
) -> FileServiceProperties
class FileSharesOperations:
def create(
self,
resource_group_name: str,
account_name: str,
share_name: str,
file_share: FileShare
) -> FileShare
def list(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[FileShareItem]Azure Queue service configuration and individual queue management operations.
class QueueServicesOperations:
def get_service_properties(
self,
resource_group_name: str,
account_name: str
) -> QueueServiceProperties
class QueueOperations:
def create(
self,
resource_group_name: str,
account_name: str,
queue_name: str,
queue: StorageQueue
) -> StorageQueueAzure Table service configuration and table management operations.
class TableServicesOperations:
def get_service_properties(
self,
resource_group_name: str,
account_name: str
) -> TableServiceProperties
class TableOperations:
def create(
self,
resource_group_name: str,
account_name: str,
table_name: str,
parameters: Table
) -> TableManagement of private endpoints, local users for SFTP access, encryption scopes, and network security configurations.
class PrivateEndpointConnectionsOperations:
def put(
self,
resource_group_name: str,
account_name: str,
private_endpoint_connection_name: str,
properties: PrivateEndpointConnection
) -> PrivateEndpointConnection
class LocalUsersOperations:
def create_or_update(
self,
resource_group_name: str,
account_name: str,
username: str,
properties: LocalUser
) -> LocalUser
class EncryptionScopesOperations:
def put(
self,
resource_group_name: str,
account_name: str,
encryption_scope_name: str,
encryption_scope: EncryptionScope
) -> EncryptionScope
class NetworkSecurityPerimeterConfigurationsOperations:
def get(
self,
resource_group_name: str,
account_name: str,
network_security_perimeter_configuration_name: str
) -> NetworkSecurityPerimeterConfiguration
def list(
self,
resource_group_name: str,
account_name: str
) -> ItemPaged[NetworkSecurityPerimeterConfiguration]Lifecycle management policies, object replication policies, and blob inventory policies for automated data management.
class ManagementPoliciesOperations:
def create_or_update(
self,
resource_group_name: str,
account_name: str,
management_policy_name: str,
properties: ManagementPolicy
) -> ManagementPolicy
class ObjectReplicationPoliciesOperations:
def create_or_update(
self,
resource_group_name: str,
account_name: str,
object_replication_policy_id: str,
properties: ObjectReplicationPolicy
) -> ObjectReplicationPolicyStorage task assignments for automated data processing and reporting on task execution.
class StorageTaskAssignmentsOperations:
def create(
self,
resource_group_name: str,
account_name: str,
storage_task_assignment_name: str,
parameters: StorageTaskAssignment
) -> StorageTaskAssignmentOperations for discovering available SKUs, checking usage limits, managing deleted accounts, and retrieving service metadata.
class SkusOperations:
def list(self) -> ItemPaged[SkuInformation]
class UsagesOperations:
def list_by_location(self, location: str) -> ItemPaged[Usage]
class DeletedAccountsOperations:
def list(self) -> ItemPaged[DeletedAccount]
class Operations:
def list(self) -> ItemPaged[Operation]class StorageAccount:
"""Storage account resource with all properties and configuration."""
id: str
name: str
type: str
location: str
sku: Sku
kind: Kind
properties: StorageAccountProperties
class Sku:
"""SKU information for storage account pricing tier."""
name: SkuName
tier: SkuTier
class StorageAccountCreateParameters:
"""Parameters for creating a new storage account."""
sku: Sku
kind: Kind
location: str
tags: Dict[str, str]
identity: Identity
properties: StorageAccountPropertiesCreateParameters
# Enums
class Kind(str, Enum):
STORAGE = "Storage"
STORAGE_V2 = "StorageV2"
BLOB_STORAGE = "BlobStorage"
FILE_STORAGE = "FileStorage"
BLOCK_BLOB_STORAGE = "BlockBlobStorage"
class SkuName(str, Enum):
STANDARD_LRS = "Standard_LRS"
STANDARD_GRS = "Standard_GRS"
STANDARD_RAGRS = "Standard_RAGRS"
STANDARD_ZRS = "Standard_ZRS"
PREMIUM_LRS = "Premium_LRS"
PREMIUM_ZRS = "Premium_ZRS"
STANDARD_GZRS = "Standard_GZRS"
STANDARD_RAGZRS = "Standard_RAGZRS"
class AccessTier(str, Enum):
HOT = "Hot"
COOL = "Cool"
PREMIUM = "Premium"
COLD = "Cold"All operations may raise Azure-specific exceptions:
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
from azure.mgmt.core.exceptions import ARMErrorFormat
try:
account = client.storage_accounts.get_properties("rg", "account")
except ResourceNotFoundError:
print("Storage account not found")
except HttpResponseError as e:
print(f"HTTP error: {e.status_code} - {e.message}")import asyncio
from azure.mgmt.storage.aio import StorageManagementClient
from azure.identity.aio import DefaultAzureCredential
async def main():
credential = DefaultAzureCredential()
async with StorageManagementClient(
credential=credential,
subscription_id="subscription-id"
) as client:
accounts = []
async for account in client.storage_accounts.list():
accounts.append(account)
print(f"Found {len(accounts)} storage accounts")
asyncio.run(main())