Microsoft Azure Blob Storage Client Library for Python providing comprehensive APIs for blob storage operations.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Container-level operations including creation, deletion, property management, access control, and blob operations within a container. The ContainerClient provides comprehensive management capabilities for Azure Blob Storage containers.
Create ContainerClient instances with various authentication methods and from different sources including URLs and connection strings.
class ContainerClient:
def __init__(self, account_url: str, container_name: str, credential=None, **kwargs):
"""
Create a ContainerClient.
Args:
account_url (str): The URL to the storage account
container_name (str): Name of the container
credential: The credentials for authentication. Can be:
- str: Account key or SAS token string
- dict: Account name and key mapping
- AzureNamedKeyCredential: Named key credential
- AzureSasCredential: SAS credential
- TokenCredential: Azure AD token credential
- None: For anonymous access or SAS URLs
"""
@classmethod
def from_container_url(cls, container_url: str, credential=None, **kwargs) -> 'ContainerClient':
"""
Create a ContainerClient from a container URL.
Args:
container_url (str): Complete URL to the container
credential: Optional credential for authentication
Returns:
ContainerClient: Configured client instance
"""
@classmethod
def from_connection_string(cls, conn_str: str, container_name: str, credential=None, **kwargs) -> 'ContainerClient':
"""
Create a ContainerClient from a connection string.
Args:
conn_str (str): Azure Storage connection string
container_name (str): Name of the container
credential: Optional credential to override connection string auth
Returns:
ContainerClient: Configured client instance
"""Create, delete, and check container existence. These operations manage the container lifecycle within Azure Blob Storage.
def create_container(self, metadata=None, public_access=None, **kwargs) -> dict:
"""
Create the container.
Args:
metadata (dict, optional): Container metadata as key-value pairs
public_access (PublicAccess, optional): Public access level for the container
Returns:
dict: Container creation response with ETag and last modified time
"""
def delete_container(self, **kwargs) -> None:
"""
Delete the container and all its blobs.
Keyword Args:
lease (BlobLeaseClient or str, optional): Required if container has an active lease
if_modified_since (datetime, optional): Delete only if modified since this time
if_unmodified_since (datetime, optional): Delete only if not modified since this time
etag (str, optional): Delete only if ETag matches
match_condition (MatchConditions, optional): ETag matching condition
"""
def exists(self, **kwargs) -> bool:
"""
Check whether the container exists.
Returns:
bool: True if container exists, False otherwise
"""Retrieve and modify container properties, metadata, and access policies.
def get_container_properties(self, **kwargs) -> ContainerProperties:
"""
Get container properties and metadata.
Returns:
ContainerProperties: Container properties including metadata, lease state, and access policies
"""
def set_container_metadata(self, metadata=None, **kwargs) -> dict:
"""
Set container metadata.
Args:
metadata (dict, optional): Container metadata as key-value pairs
Returns:
dict: Response with ETag and last modified time
"""
def get_container_access_policy(self, **kwargs) -> dict:
"""
Get the container's stored access policies.
Returns:
dict: Dictionary containing:
- public_access: PublicAccess level
- signed_identifiers: List of AccessPolicy objects
"""
def set_container_access_policy(self, signed_identifiers=None, public_access=None, **kwargs) -> dict:
"""
Set the container's access policy and public access level.
Args:
signed_identifiers (dict, optional): Stored access policies by identifier
public_access (PublicAccess, optional): Public access level
Returns:
dict: Response with ETag and last modified time
"""Acquire and manage leases on containers to prevent concurrent modifications.
def acquire_lease(self, lease_duration=-1, lease_id=None, **kwargs) -> BlobLeaseClient:
"""
Acquire a lease on the container.
Args:
lease_duration (int): Lease duration in seconds (-1 for infinite)
lease_id (str, optional): Proposed lease identifier
Returns:
BlobLeaseClient: Lease client for managing the lease
"""List blobs within the container with various filtering and inclusion options, including hierarchical directory-style traversal.
def list_blobs(self, name_starts_with=None, include=None, **kwargs) -> ItemPaged[BlobProperties]:
"""
List blobs in the container.
Args:
name_starts_with (str, optional): Filter blobs by name prefix
include (list, optional): Additional properties to include:
- 'snapshots': Include blob snapshots
- 'metadata': Include blob metadata
- 'uncommittedblobs': Include uncommitted blobs
- 'copy': Include copy properties
- 'deleted': Include soft-deleted blobs
- 'tags': Include blob index tags
- 'versions': Include blob versions
- 'immutabilitypolicy': Include immutability policy
- 'legalhold': Include legal hold status
- 'deletedwithversions': Include deleted blobs with versions
Returns:
ItemPaged[BlobProperties]: Paginated list of blob properties
"""
def list_blob_names(self, **kwargs) -> ItemPaged[str]:
"""
List blob names only (more efficient than full properties).
Returns:
ItemPaged[str]: Paginated list of blob names
"""
def walk_blobs(self, name_starts_with=None, include=None, delimiter='/', **kwargs) -> ItemPaged[Union[BlobProperties, BlobPrefix]]:
"""
Walk blobs in a directory-like hierarchy using delimiter.
Args:
name_starts_with (str, optional): Filter by name prefix
include (list, optional): Additional properties to include
delimiter (str): Delimiter for hierarchical listing (default '/')
Returns:
ItemPaged[Union[BlobProperties, BlobPrefix]]: Paginated list of blobs and prefixes
"""
def find_blobs_by_tags(self, filter_expression: str, **kwargs) -> ItemPaged['FilteredBlob']:
"""
Find blobs in the container using tag-based filtering.
Args:
filter_expression (str): OData filter expression for blob tags
Returns:
ItemPaged[FilteredBlob]: Paginated list of matching blobs
"""Upload, download, and delete blobs through the container client for convenient blob management.
def upload_blob(self, name: str, data, blob_type='BlockBlob', **kwargs) -> BlobClient:
"""
Upload data as a blob in the container.
Args:
name (str): Name for the blob
data: Data to upload (bytes, str, or file-like object)
blob_type (BlobType): Type of blob to create
Keyword Args:
length (int, optional): Number of bytes to upload
metadata (dict, optional): Blob metadata
content_type (str, optional): MIME content type
content_encoding (str, optional): Content encoding
content_language (str, optional): Content language
content_disposition (str, optional): Content disposition
cache_control (str, optional): Cache control header
content_md5 (bytes, optional): MD5 hash of content
validate_content (bool): Validate content integrity
max_concurrency (int): Maximum concurrent uploads
overwrite (bool): Whether to overwrite existing blob
standard_blob_tier (StandardBlobTier, optional): Access tier
tags (dict, optional): Blob index tags
Returns:
BlobClient: Client for the uploaded blob
"""
def download_blob(self, blob: str, offset=None, length=None, **kwargs) -> StorageStreamDownloader:
"""
Download a blob from the container.
Args:
blob (str): Name of the blob to download
offset (int, optional): Start byte position
length (int, optional): Number of bytes to download
Returns:
StorageStreamDownloader: Streaming downloader for blob content
"""
def delete_blob(self, blob: str, delete_snapshots=None, **kwargs) -> None:
"""
Delete a blob from the container.
Args:
blob (str): Name of the blob to delete
delete_snapshots (str, optional): How to handle snapshots:
- 'include': Delete blob and all snapshots
- 'only': Delete snapshots but not the blob
"""Perform efficient batch operations on multiple blobs within the container.
def delete_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
"""
Delete multiple blobs in a batch operation.
Args:
*blobs: Blob names or BlobProperties objects to delete
Keyword Args:
delete_snapshots (str, optional): How to handle snapshots
raise_on_any_failure (bool): Whether to raise on any failure
Returns:
Iterator[HttpResponse]: Response for each delete operation
"""
def set_standard_blob_tier_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
"""
Set access tier for multiple standard blobs in batch.
Args:
*blobs: Tuples of (blob_name, standard_blob_tier) or BlobProperties with tier
Returns:
Iterator[HttpResponse]: Response for each tier operation
"""
def set_premium_page_blob_tier_blobs(self, *blobs, **kwargs) -> Iterator[HttpResponse]:
"""
Set access tier for multiple premium page blobs in batch.
Args:
*blobs: Tuples of (blob_name, premium_page_blob_tier) or BlobProperties with tier
Returns:
Iterator[HttpResponse]: Response for each tier operation
"""Create blob clients for individual blob operations within the container.
def get_blob_client(self, blob: str, snapshot=None) -> 'BlobClient':
"""
Get a BlobClient for a specific blob in this container.
Args:
blob (str): Name of the blob
snapshot (str, optional): Blob snapshot identifier
Returns:
BlobClient: Client for blob operations
"""Retrieve storage account information through the container client.
def get_account_information(self, **kwargs) -> dict:
"""
Get information about the storage account.
Returns:
dict: Account information including account kind and SKU
"""class PublicAccess:
"""Container public access levels."""
OFF: str # No public access
BLOB: str # Public read access for blobs only
CONTAINER: str # Public read access for container and blobs
class ContainerProperties:
"""Container properties and metadata."""
name: str
last_modified: datetime
etag: str
lease: LeaseProperties
public_access: PublicAccess
has_immutability_policy: bool
deleted: bool
version: str
has_legal_hold: bool
metadata: dict
encryption_scope: ContainerEncryptionScope
immutable_storage_with_versioning_enabled: bool
class BlobPrefix:
"""Blob name prefix for hierarchical listing."""
name: str
service_endpoint: str
container_name: str
class AccessPolicy:
"""Stored access policy for container."""
permission: str
expiry: datetime
start: datetime
class LeaseProperties:
"""Container or blob lease information."""
status: str
state: str
duration: str
class ContainerEncryptionScope:
"""Container encryption scope configuration."""
default_encryption_scope: str
deny_encryption_scope_override: boolInstall with Tessl CLI
npx tessl i tessl/pypi-azure-storage-blob