tessl/pypi-azure-storage-file-share
Azure File Share storage client library for Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
share-client.mddocs/
ShareClient - Share-Level Operations
The ShareClient provides operations for managing individual file shares, including creating snapshots, managing quotas, setting permissions, and performing directory and file operations at the share root level.
Import and Initialization
from azure.storage.fileshare import ShareClient, ShareLeaseClient
from azure.core.credentials import AzureNamedKeyCredential
from typing import Optional, Union, Dict, Any, ListConstructor
class ShareClient:
def __init__(
self,
account_url: str,
share_name: str,
snapshot: Optional[str] = None,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
*,
token_intent: Optional[Literal['backup']] = None,
**kwargs: Any
) -> None:
"""
Create a ShareClient for a specific share.
Parameters:
account_url: The URL to the file service endpoint
share_name: Name of the share
snapshot: Optional share snapshot identifier
credential: Authentication credential
token_intent: Specifies the intent for all requests when using TokenCredential authentication
**kwargs: Additional client configuration options
"""Class Methods
@classmethod
def from_share_url(
cls,
share_url: str,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareClient:
"""
Create ShareClient from share URL.
Parameters:
share_url: Complete URL to the share
credential: Authentication credential
**kwargs: Additional client configuration options
Returns:
ShareClient: Configured client instance
"""
@classmethod
def from_connection_string(
cls,
conn_str: str,
share_name: str,
snapshot: Optional[str] = None,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareClient:
"""
Create ShareClient from connection string.
Parameters:
conn_str: Azure Storage connection string
share_name: Name of the share
snapshot: Optional share snapshot identifier
credential: Optional credential to override connection string auth
**kwargs: Additional client configuration options
Returns:
ShareClient: Configured client instance
"""Client Factory Methods
def get_directory_client(
self,
directory_path: Optional[str] = None
) -> ShareDirectoryClient:
"""
Get a client to interact with the specified directory.
Parameters:
directory_path: Path to the directory (None for root directory)
Returns:
ShareDirectoryClient: Client configured for the directory
"""
def get_file_client(
self,
file_path: str
) -> ShareFileClient:
"""
Get a client to interact with the specified file.
Parameters:
file_path: Path to the file from share root
Returns:
ShareFileClient: Client configured for the file
"""Share Lifecycle Operations
def create_share(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Creates a new share under the account.
Parameters:
metadata: Dict of name-value pairs for share metadata
quota: Share size quota in GB (1-102400 for standard accounts)
access_tier: Access tier ("Hot", "Cool", "TransactionOptimized", "Premium")
protocols: Enabled protocols (list of "SMB", "NFS")
root_squash: Root squash setting for NFS ("NoRootSquash", "RootSquash", "AllSquash")
timeout: Request timeout in seconds
Returns:
Dict containing share creation response with ETag and last modified time
"""
def delete_share(
self,
delete_snapshots: Optional[bool] = False,
**kwargs: Any
) -> None:
"""
Marks the specified share for deletion.
Parameters:
delete_snapshots: Whether to delete share snapshots
lease_id: Required if share has an active lease
if_modified_since: Delete only if modified since specified time
if_unmodified_since: Delete only if not modified since specified time
if_match: Delete only if ETag matches
if_none_match: Delete only if ETag does not match
timeout: Request timeout in seconds
"""
def create_snapshot(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Creates a snapshot of the share.
Parameters:
metadata: Dict of name-value pairs for snapshot metadata
timeout: Request timeout in seconds
Returns:
Dict containing snapshot information including snapshot ID and ETag
"""Share Properties and Metadata
def get_share_properties(
self,
**kwargs: Any
) -> ShareProperties:
"""
Returns all user-defined metadata and system properties for the share.
Parameters:
timeout: Request timeout in seconds
Returns:
ShareProperties: Object containing all share properties
"""
def set_share_quota(
self,
quota: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets the quota for the share.
Parameters:
quota: Share size limit in GB (1-102400 for standard accounts)
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""
def set_share_properties(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets share properties.
Parameters:
access_tier: Share access tier ("Hot", "Cool", "TransactionOptimized", "Premium")
quota: Share size quota in GB
root_squash: Root squash setting for NFS shares
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""
def set_share_metadata(
self,
metadata: Dict[str, str],
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets user-defined metadata for the share.
Parameters:
metadata: Dict of name-value pairs (keys must be valid metadata names)
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""Access Control and Permissions
def get_share_access_policy(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Gets the permissions for the share.
Parameters:
timeout: Request timeout in seconds
Returns:
Dict containing signed identifiers with their access policies
"""
def set_share_access_policy(
self,
signed_identifiers: Dict[str, AccessPolicy],
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets the permissions for the share.
Parameters:
signed_identifiers: Dict mapping policy IDs to AccessPolicy objects
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""
def create_permission_for_share(
self,
file_permission: str,
**kwargs: Any
) -> Optional[str]:
"""
Create a permission (security descriptor) at the share level.
Parameters:
file_permission: Security descriptor string in SDDL format
timeout: Request timeout in seconds
Returns:
Optional[str]: Permission key that can be used to reference this permission
"""
def get_permission_for_share(
self,
permission_key: str,
**kwargs: Any
) -> str:
"""
Get a permission (security descriptor) for a given key.
Parameters:
permission_key: Key returned from create_permission_for_share
timeout: Request timeout in seconds
Returns:
str: Security descriptor string in SDDL format
"""Share Statistics
def get_share_stats(
self,
**kwargs: Any
) -> int:
"""
Gets the approximate size of the data stored on the share.
Parameters:
timeout: Request timeout in seconds
Returns:
int: Size of share content in bytes
"""Directory and File Management
def list_directories_and_files(
self,
directory_name: Optional[str] = None,
name_starts_with: Optional[str] = None,
marker: Optional[str] = None,
**kwargs: Any
) -> ItemPaged[Dict[str, Any]]:
"""
Lists directories and files under the share.
Parameters:
directory_name: Directory path to list (None for root)
name_starts_with: Filter by name prefix
marker: Continuation token for pagination
results_per_page: Maximum items per page
timeout: Request timeout in seconds
Returns:
ItemPaged[Dict]: Paginated list of directory/file info dicts
"""
def create_directory(
self,
directory_name: str,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Creates a directory in the share.
Parameters:
directory_name: Name/path of the directory to create
metadata: Dict of name-value pairs for directory metadata
timeout: Request timeout in seconds
Returns:
ShareDirectoryClient: Client for the newly created directory
"""
def delete_directory(
self,
directory_name: str,
**kwargs: Any
) -> None:
"""
Deletes the specified directory.
Parameters:
directory_name: Name/path of the directory to delete
timeout: Request timeout in seconds
"""Lease Management
def acquire_lease(
self,
**kwargs: Any
) -> ShareLeaseClient:
"""
Requests a new lease for the share.
Parameters:
lease_duration: Duration of the lease in seconds (-1 for infinite)
lease_id: Proposed lease ID (UUID format)
timeout: Request timeout in seconds
Returns:
ShareLeaseClient: Lease client for managing the lease
"""Properties
@property
def url(self) -> str:
"""The full endpoint URL to the share."""
@property
def share_name(self) -> str:
"""The name of the share with which to interact."""
@property
def snapshot(self) -> Optional[str]:
"""An optional share snapshot on which to operate."""
@property
def account_name(self) -> str:
"""The storage account name."""
@property
def credential(self) -> Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]:
"""The credential used to authenticate."""Usage Examples
Basic Share Operations
from azure.storage.fileshare import ShareClient
from azure.core.credentials import AzureNamedKeyCredential
# Initialize client
credential = AzureNamedKeyCredential("myaccount", "mykey")
share_client = ShareClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
credential=credential
)
# Create share with properties
response = share_client.create_share(
quota=100,
access_tier="Hot",
metadata={"environment": "production", "department": "finance"}
)
print(f"Share created with ETag: {response['etag']}")
# Get share properties
properties = share_client.get_share_properties()
print(f"Share quota: {properties.quota} GB")
print(f"Last modified: {properties.last_modified}")
print(f"Metadata: {properties.metadata}")Share Configuration Management
# Update share quota
share_client.set_share_quota(200) # Increase to 200 GB
# Set share properties
share_client.set_share_properties(
access_tier="Cool", # Change to cooler tier for cost savings
quota=150
)
# Update metadata
share_client.set_share_metadata({
"environment": "production",
"cost_center": "12345",
"retention_policy": "7years"
})
# Get share statistics
used_space = share_client.get_share_stats()
properties = share_client.get_share_properties()
print(f"Share usage: {used_space / (1024**3):.2f} GB of {properties.quota} GB")Snapshot Management
from datetime import datetime
# Create snapshot with metadata
snapshot_response = share_client.create_snapshot(
metadata={"backup_date": datetime.now().isoformat()}
)
snapshot_id = snapshot_response['snapshot']
print(f"Snapshot created: {snapshot_id}")
# Work with snapshot
snapshot_client = ShareClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
snapshot=snapshot_id,
credential=credential
)
# List files in snapshot
snapshot_items = list(snapshot_client.list_directories_and_files())
print(f"Snapshot contains {len(snapshot_items)} items")Access Policy Management
from azure.storage.fileshare import AccessPolicy, ShareSasPermissions
from datetime import datetime, timedelta
# Define access policies
read_policy = AccessPolicy(
permission=ShareSasPermissions(read=True, list=True),
expiry=datetime.utcnow() + timedelta(days=30),
start=datetime.utcnow()
)
write_policy = AccessPolicy(
permission=ShareSasPermissions(read=True, write=True, create=True, delete=True),
expiry=datetime.utcnow() + timedelta(hours=24),
start=datetime.utcnow()
)
# Set access policies
share_client.set_share_access_policy({
"readonly": read_policy,
"fullaccess": write_policy
})
# Get current policies
policies = share_client.get_share_access_policy()
for policy_id, policy in policies.items():
print(f"Policy {policy_id}: expires {policy['expiry']}")Permission Management (Windows ACLs)
# Create a security descriptor for Windows ACL
sddl = "O:BAG:BAD:(A;;FA;;;SY)(A;;FA;;;BA)(A;;0x1200a9;;;BU)"
permission_key = share_client.create_permission_for_share(sddl)
print(f"Permission created with key: {permission_key}")
# Retrieve permission by key
retrieved_permission = share_client.get_permission_for_share(permission_key)
print(f"Retrieved permission: {retrieved_permission}")Directory and File Operations
# Create directories
documents_dir = share_client.create_directory("documents")
reports_dir = share_client.create_directory("reports/quarterly")
# List root contents
root_items = list(share_client.list_directories_and_files())
for item in root_items:
item_type = "Directory" if item.get('is_directory') else "File"
print(f"{item_type}: {item['name']}")
# Filter by prefix
pdf_files = list(share_client.list_directories_and_files(
name_starts_with="report"
))
# Get clients for further operations
file_client = share_client.get_file_client("document.pdf")
directory_client = share_client.get_directory_client("documents")Lease Management
# Acquire a lease on the share
lease_client = share_client.acquire_lease(lease_duration=60) # 60 second lease
print(f"Lease acquired with ID: {lease_client.id}")
# Use lease for operations
try:
share_client.set_share_metadata(
{"locked_by": "batch_process"},
lease_id=lease_client.id
)
# Perform operations while lease is held
# ...
finally:
# Always release the lease
lease_client.release()Error Handling and Validation
from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError, HttpResponseError
try:
# Attempt share creation
share_client.create_share(quota=50)
except ResourceExistsError:
print("Share already exists, checking properties...")
properties = share_client.get_share_properties()
print(f"Existing quota: {properties.quota} GB")
try:
# Try to get non-existent permission
permission = share_client.get_permission_for_share("invalid-key")
except HttpResponseError as e:
if e.error_code == "SharePermissionNotFound":
print("Permission key not found")
else:
raise
# Validate quota limits
properties = share_client.get_share_properties()
if properties.quota < 100:
share_client.set_share_quota(100)
print("Quota increased to minimum required size")The ShareClient provides comprehensive management capabilities for individual file shares, enabling you to configure properties, manage access control, create snapshots, and perform directory/file operations at the share level.
Install with Tessl CLI
npx tessl i tessl/pypi-azure-storage-file-share