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
file-client.mddocs/
ShareFileClient - File Operations
The ShareFileClient provides comprehensive file management operations including upload/download, copying, range operations, metadata management, and advanced features like leasing and symbolic links.
Import and Initialization
from azure.storage.fileshare import ShareFileClient, ShareLeaseClient, ContentSettings
from azure.storage.fileshare import StorageStreamDownloader, Handle
from azure.core.credentials import AzureNamedKeyCredential
from typing import Optional, Union, Dict, Any, List, IO, AnyStr, Iterable, TupleConstructor
class ShareFileClient:
def __init__(
self,
account_url: str,
share_name: str,
file_path: str,
snapshot: Optional[Union[Dict[str, Any], str]] = None,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
*,
token_intent: Optional[Literal['backup']] = None,
**kwargs: Any
) -> None:
"""
Create a ShareFileClient for a specific file.
Parameters:
account_url: The URL to the file service endpoint
share_name: Name of the share containing the file
file_path: Path to the file from share root
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_file_url(
cls,
file_url: str,
snapshot: Optional[Union[str, Dict[str, Any]]] = None,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareFileClient:
"""
Create ShareFileClient from file URL.
Parameters:
file_url: Complete URL to the file
snapshot: Optional share snapshot identifier
credential: Authentication credential
**kwargs: Additional client configuration options
Returns:
ShareFileClient: Configured client instance
"""
@classmethod
def from_connection_string(
cls,
conn_str: str,
share_name: str,
file_path: str,
snapshot: Optional[Union[str, Dict[str, Any]]] = None,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareFileClient:
"""
Create ShareFileClient from connection string.
Parameters:
conn_str: Azure Storage connection string
share_name: Name of the share
file_path: Path to the file from share root
snapshot: Optional share snapshot identifier
credential: Optional credential to override connection string auth
**kwargs: Additional client configuration options
Returns:
ShareFileClient: Configured client instance
"""File Lifecycle Operations
def exists(
self,
**kwargs: Any
) -> bool:
"""
Returns True if the file exists, False otherwise.
Parameters:
timeout: Request timeout in seconds
Returns:
bool: True if file exists
"""
def create_file(
self,
size: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Creates a new file with the specified size.
Parameters:
size: Size of the file in bytes
content_settings: ContentSettings object with HTTP properties
metadata: Dict of name-value pairs for file metadata
file_attributes: File attributes (NTFSAttributes)
file_creation_time: Creation time for the file
file_last_write_time: Last write time for the file
file_permission: Security descriptor string or permission key
file_permission_key: Key for a previously created permission
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing file creation response with ETag and properties
"""
def delete_file(
self,
**kwargs: Any
) -> None:
"""
Marks the specified file for deletion.
Parameters:
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
"""
def rename_file(
self,
new_name: str,
**kwargs: Any
) -> ShareFileClient:
"""
Rename a file and return a client for the renamed file.
Parameters:
new_name: New path/name for the file
overwrite: Whether to overwrite existing file with same name
ignore_readonly: Whether to ignore readonly attribute on destination
file_permission: Security descriptor for the destination file
file_permission_key: Permission key for the destination file
timeout: Request timeout in seconds
Returns:
ShareFileClient: Client for the renamed file
"""File Upload Operations
def upload_file(
self,
data: Union[bytes, str, Iterable[AnyStr], IO[AnyStr]],
length: Optional[int] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""
Uploads a new file or overwrites an existing file.
Parameters:
data: File content to upload
length: Length of the data in bytes (required for streams)
overwrite: Whether to overwrite existing file (default: False)
max_concurrency: Number of parallel upload threads (default: 1)
content_settings: ContentSettings object with HTTP properties
metadata: Dict of name-value pairs for file metadata
validate_content: Whether to validate content with MD5 hash
file_attributes: File attributes (NTFSAttributes)
file_creation_time: Creation time for the file
file_last_write_time: Last write time for the file
file_permission: Security descriptor string or permission key
file_permission_key: Key for a previously created permission
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing upload response with ETag and properties
"""
def upload_range(
self,
data: bytes,
offset: int,
length: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Upload a range of bytes to a file.
Parameters:
data: Bytes to upload
offset: Start position in the file
length: Number of bytes to upload
validate_content: Whether to validate content with MD5 hash
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing range upload response
"""
def upload_range_from_url(
self,
source_url: str,
offset: int,
length: int,
source_offset: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Writes bytes from one Azure File endpoint into a range of another file.
Parameters:
source_url: URL of the source file (can include SAS token)
offset: Start position in the destination file
length: Number of bytes to copy
source_offset: Start position in the source file
timeout: Request timeout in seconds
Returns:
Dict containing range upload response
"""File Download Operations
def download_file(
self,
offset: Optional[int] = None,
length: Optional[int] = None,
**kwargs: Any
) -> StorageStreamDownloader:
"""
Downloads a file to a StorageStreamDownloader.
Parameters:
offset: Start position to download from
length: Number of bytes to download
validate_content: Whether to validate content with MD5 hash
max_concurrency: Number of parallel download threads (default: 1)
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
StorageStreamDownloader: Stream object for downloading file content
"""File Copy Operations
def start_copy_from_url(
self,
source_url: str,
**kwargs: Any
) -> Dict[str, Any]:
"""
Initiates copying data from a source URL into the file.
Parameters:
source_url: URL of the source file (can include SAS token)
metadata: Dict of name-value pairs for destination file metadata
file_permission: Security descriptor for destination file
file_permission_key: Permission key for destination file
ignore_readonly: Whether to ignore readonly attribute on destination
file_attributes: File attributes for destination file
file_creation_time: Creation time for destination file
file_last_write_time: Last write time for destination file
lease_id: Required if destination file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing copy operation details including copy_id
"""
def abort_copy(
self,
copy_id: Union[str, FileProperties],
**kwargs: Any
) -> None:
"""
Aborts a pending copy operation and leaves destination file with zero length.
Parameters:
copy_id: Copy identifier from start_copy_from_url response
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
"""File Properties and Metadata
def get_file_properties(
self,
**kwargs: Any
) -> FileProperties:
"""
Returns all user-defined metadata, standard HTTP properties, and system properties.
Parameters:
timeout: Request timeout in seconds
Returns:
FileProperties: Object containing all file properties
"""
def set_http_headers(
self,
content_settings: ContentSettings,
file_attributes: Optional[Union[str, NTFSAttributes]] = None,
file_creation_time: Optional[Union[str, datetime]] = None,
file_last_write_time: Optional[Union[str, datetime]] = None,
file_permission: Optional[str] = None,
permission_key: Optional[str] = None,
*,
lease: Optional[Union[ShareLeaseClient, str]] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets system properties on the file.
Parameters:
content_settings: ContentSettings object with HTTP properties
file_attributes: File attributes (NTFSAttributes)
file_creation_time: Creation time for the file
file_last_write_time: Last write time for the file
file_permission: Security descriptor string or permission key
permission_key: Key for a previously created permission
lease: Required if file has an active lease
**kwargs: Additional keyword arguments
Returns:
Dict containing response with ETag and last modified time
"""
def set_file_metadata(
self,
metadata: Optional[Dict[str, Any]] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets user-defined metadata for the file.
Parameters:
metadata: Dict of name-value pairs (keys must be valid metadata names)
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""Range Operations
def get_ranges(
self,
offset: Optional[int] = None,
length: Optional[int] = None,
**kwargs: Any
) -> List[Dict[str, int]]:
"""
Returns the list of valid page ranges for a file.
Parameters:
offset: Start position to get ranges from
length: Number of bytes to get ranges for
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
List[Dict]: List of range dicts with 'start' and 'end' keys
"""
def get_ranges_diff(
self,
previous_sharesnapshot: Union[str, Dict[str, Any]],
offset: Optional[int] = None,
length: Optional[int] = None,
*,
include_renames: Optional[bool] = None,
**kwargs: Any
) -> Tuple[List[Dict[str, int]], List[Dict[str, int]]]:
"""
Returns ranges that differ between a previous snapshot and current file.
Parameters:
previous_sharesnapshot: Snapshot identifier to compare against
offset: Start position to compare from
length: Number of bytes to compare
include_renames: Whether to include renamed ranges in the response
**kwargs: Additional keyword arguments
Returns:
Tuple[List, List]: (changed_ranges, cleared_ranges)
"""
def clear_range(
self,
offset: int,
length: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Clears the specified range and releases space used in storage.
Parameters:
offset: Start position of the range to clear
length: Number of bytes to clear
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""
def resize_file(
self,
size: int,
**kwargs: Any
) -> Dict[str, Any]:
"""
Resizes a file to the specified size.
Parameters:
size: New size of the file in bytes
lease_id: Required if file has an active lease
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""Handle Management
def list_handles(
self,
**kwargs: Any
) -> ItemPaged[Handle]:
"""
Lists handles for the file.
Parameters:
marker: Continuation token for pagination
results_per_page: Maximum handles per page
timeout: Request timeout in seconds
Returns:
ItemPaged[Handle]: Paginated list of open handles
"""
def close_handle(
self,
handle: Union[str, Handle],
**kwargs: Any
) -> Dict[str, int]:
"""
Closes an open file handle.
Parameters:
handle: Handle ID string or Handle object to close
timeout: Request timeout in seconds
Returns:
Dict containing number of handles closed
"""
def close_all_handles(
self,
**kwargs: Any
) -> Dict[str, int]:
"""
Closes all handles for the file.
Parameters:
timeout: Request timeout in seconds
Returns:
Dict containing number of handles closed
"""Advanced File Operations
def create_hardlink(
self,
target: str,
*,
lease: Optional[Union[ShareLeaseClient, str]] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""
Create a hard link to an existing file.
Parameters:
target: Path for the new hard link
lease: Required if file has an active lease
**kwargs: Additional keyword arguments
Returns:
Dict[str, Any]: Response containing operation metadata
"""
def create_symlink(
self,
target: str,
*,
metadata: Optional[Dict[str, str]] = None,
**kwargs: Any
) -> Dict[str, Any]:
"""
Create a symbolic link to an existing file or directory.
Parameters:
target: Path to the target file or directory
metadata: Metadata for the symbolic link
**kwargs: Additional keyword arguments
Returns:
Dict[str, Any]: Response containing operation metadata
"""
def get_symlink(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Get the target of a symbolic link.
Parameters:
**kwargs: Additional keyword arguments
Returns:
Dict[str, Any]: Response containing symlink target and metadata
"""Lease Management
def acquire_lease(
self,
lease_id: Optional[str] = None,
**kwargs: Any
) -> ShareLeaseClient:
"""
Requests a new lease for the file.
Parameters:
lease_id: Proposed lease ID (UUID format)
lease_duration: Duration of the lease in seconds (-1 for infinite)
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 file."""
@property
def share_name(self) -> str:
"""The name of the share containing this file."""
@property
def file_path(self) -> str:
"""The path to the file 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 File Operations
from azure.storage.fileshare import ShareFileClient, ContentSettings
from azure.core.credentials import AzureNamedKeyCredential
# Initialize file client
credential = AzureNamedKeyCredential("myaccount", "mykey")
file_client = ShareFileClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
file_path="reports/monthly_report.pdf",
credential=credential
)
# Check if file exists
if file_client.exists():
print("File already exists")
else:
print("File does not exist")
# Upload a new file
with open("local_report.pdf", "rb") as data:
upload_response = file_client.upload_file(
data=data,
overwrite=True,
content_settings=ContentSettings(
content_type="application/pdf",
content_disposition="attachment; filename=monthly_report.pdf"
),
metadata={"department": "finance", "month": "january", "year": "2024"}
)
print(f"File uploaded with ETag: {upload_response['etag']}")File Download and Content Processing
# Download entire file
download_stream = file_client.download_file()
# Method 1: Download to bytes
file_content = download_stream.readall()
print(f"Downloaded {len(file_content)} bytes")
# Method 2: Download to local file
with open("downloaded_report.pdf", "wb") as file_handle:
download_stream.readinto(file_handle)
# Method 3: Download with parallel connections
download_stream = file_client.download_file(max_concurrency=4)
content = download_stream.content_as_bytes(max_concurrency=4)
# Partial download (range)
partial_stream = file_client.download_file(offset=0, length=1024) # First 1KB
first_chunk = partial_stream.readall()
# Download and process as text
text_file_client = ShareFileClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
file_path="config.txt",
credential=credential
)
text_stream = text_file_client.download_file()
text_content = text_stream.content_as_text(encoding="utf-8")
print(f"File content:\n{text_content}")Advanced Upload Scenarios
from azure.storage.fileshare import NTFSAttributes
from datetime import datetime, timezone
# Large file upload with progress tracking
def upload_large_file(file_client, local_path, chunk_size=4*1024*1024):
"""Upload large file in chunks with progress."""
import os
file_size = os.path.getsize(local_path)
# Create empty file first
file_client.create_file(file_size)
with open(local_path, "rb") as file_handle:
offset = 0
while offset < file_size:
chunk = file_handle.read(chunk_size)
if not chunk:
break
file_client.upload_range(
data=chunk,
offset=offset,
length=len(chunk)
)
offset += len(chunk)
progress = (offset / file_size) * 100
print(f"Upload progress: {progress:.1f}%")
# Upload with custom attributes
upload_response = file_client.upload_file(
data=b"Sample file content",
overwrite=True,
file_attributes=NTFSAttributes.Archive | NTFSAttributes.ReadOnly,
file_creation_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
file_last_write_time=datetime.now(timezone.utc),
validate_content=True # Validates MD5 hash
)File Copy and Backup Operations
# Copy file from another Azure file share
source_url = "https://sourceaccount.file.core.windows.net/source-share/file.txt"
source_sas = "?sp=r&st=2024-01-01T00:00:00Z&se=2024-12-31T23:59:59Z&sv=2021-06-08&sr=f&sig=..."
copy_response = file_client.start_copy_from_url(
source_url=source_url + source_sas,
metadata={"copied_from": "backup_share", "copy_date": datetime.now().isoformat()}
)
copy_id = copy_response['copy_id']
print(f"Copy started with ID: {copy_id}")
# Monitor copy progress
import time
while True:
properties = file_client.get_file_properties()
copy_status = properties.copy.status
print(f"Copy status: {copy_status}")
if copy_status == 'success':
print("Copy completed successfully")
break
elif copy_status == 'failed':
print(f"Copy failed: {properties.copy.status_description}")
break
time.sleep(5) # Check again in 5 seconds
# Copy between ranges
source_file_client = ShareFileClient(
account_url="https://myaccount.file.core.windows.net",
share_name="source",
file_path="large_file.bin",
credential=credential
)
# Copy specific range from source to destination
file_client.upload_range_from_url(
source_url=source_file_client.url + "?" + sas_token,
offset=0, # Destination offset
length=1024*1024, # 1MB
source_offset=5*1024*1024 # Start from 5MB in source
)Range Operations and File Manipulation
# Get file ranges (for sparse files)
ranges = file_client.get_ranges()
print("File ranges:")
for range_info in ranges:
print(f" {range_info['start']}-{range_info['end']} ({range_info['end'] - range_info['start'] + 1} bytes)")
# Clear a range (create a hole in the file)
file_client.clear_range(offset=1024, length=2048)
# Compare with snapshot
snapshot_client = ShareFileClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
file_path="reports/monthly_report.pdf",
snapshot="2024-01-01T00:00:00Z",
credential=credential
)
changed_ranges, cleared_ranges = file_client.get_ranges_diff(
previous_sharesnapshot="2024-01-01T00:00:00Z"
)
print(f"Changed ranges: {len(changed_ranges)}")
print(f"Cleared ranges: {len(cleared_ranges)}")
# Resize file
current_size = file_client.get_file_properties().size
new_size = current_size + 1024*1024 # Add 1MB
file_client.resize_file(new_size)
print(f"File resized from {current_size} to {new_size} bytes")Metadata and Properties Management
# Get comprehensive file properties
properties = file_client.get_file_properties()
print(f"File: {properties.name}")
print(f"Size: {properties.size} bytes")
print(f"Content type: {properties.content_settings.content_type}")
print(f"Last modified: {properties.last_modified}")
print(f"ETag: {properties.etag}")
print(f"Metadata: {properties.metadata}")
print(f"Attributes: {properties.file_attributes}")
print(f"Creation time: {properties.creation_time}")
# Update HTTP headers
file_client.set_http_headers(
content_settings=ContentSettings(
content_type="application/json",
content_encoding="gzip",
content_language="en-US",
cache_control="max-age=3600"
)
)
# Update metadata
file_client.set_file_metadata({
"processed": "true",
"processing_date": datetime.now().isoformat(),
"version": "2.1",
"checksum": "abc123def456"
})Handle Management for Open Files
# List open handles on the file
handles = list(file_client.list_handles())
print(f"File has {len(handles)} open handle(s)")
for handle in handles:
print(f"Handle {handle.id}:")
print(f" Client: {handle.client_name} ({handle.client_ip})")
print(f" Session: {handle.session_id}")
print(f" Opened: {handle.open_time}")
print(f" Access: {handle.access_rights}")
# Close specific handle (useful for releasing locks)
if handles:
result = file_client.close_handle(handles[0])
print(f"Closed {result['closed_handles_count']} handle(s)")
# Close all handles (useful for maintenance)
result = file_client.close_all_handles()
print(f"Closed {result['closed_handles_count']} total handle(s)")Symbolic Links and Hard Links
# Create symbolic link
symlink_client = file_client.create_symlink(
target_path="../../shared/template.docx",
file_attributes=NTFSAttributes.ReparsePoint
)
print(f"Symbolic link created: {symlink_client.file_path}")
# Get symlink target
target = symlink_client.get_symlink()
print(f"Symlink points to: {target}")
# Create hard link (multiple names for same file)
hardlink_client = file_client.create_hardlink("backup_copy.pdf")
print(f"Hard link created: {hardlink_client.file_path}")
# Both clients point to the same file content
original_props = file_client.get_file_properties()
hardlink_props = hardlink_client.get_file_properties()
print(f"Same file ID: {original_props.file_id == hardlink_props.file_id}")Lease Management for Exclusive Access
# Acquire lease for exclusive access
lease_client = file_client.acquire_lease(lease_duration=60) # 60 second lease
print(f"Lease acquired: {lease_client.id}")
try:
# Perform operations with lease
file_client.set_file_metadata(
{"locked_by": "batch_process", "lock_time": datetime.now().isoformat()},
lease_id=lease_client.id
)
# Upload new content (requires lease)
file_client.upload_file(
data=b"Updated content with exclusive access",
overwrite=True,
lease_id=lease_client.id
)
finally:
# Always release the lease
lease_client.release()
print("Lease released")Error Handling and Recovery
from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError, HttpResponseError
# Safe file operations with error handling
try:
# Try to upload file
file_client.upload_file(data=b"content", overwrite=False)
except ResourceExistsError:
print("File exists, checking if update is needed...")
properties = file_client.get_file_properties()
if properties.size == 0:
# Empty file, safe to overwrite
file_client.upload_file(data=b"content", overwrite=True)
else:
print("File has content, skipping upload")
# Handle copy operation errors
try:
copy_response = file_client.start_copy_from_url(source_url)
copy_id = copy_response['copy_id']
# Monitor and handle copy failure
properties = file_client.get_file_properties()
if properties.copy.status == 'failed':
print(f"Copy failed: {properties.copy.status_description}")
file_client.abort_copy(copy_id)
except HttpResponseError as e:
if e.error_code == "CannotVerifyCopySource":
print("Source file not accessible or SAS token expired")
else:
print(f"Copy operation failed: {e.message}")
# Retry logic for transient failures
import time
from azure.core.exceptions import ServiceRequestError
def upload_with_retry(file_client, data, max_retries=3):
"""Upload with exponential backoff retry."""
for attempt in range(max_retries):
try:
return file_client.upload_file(data, overwrite=True)
except ServiceRequestError as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt # Exponential backoff
print(f"Upload failed, retrying in {wait_time} seconds...")
time.sleep(wait_time)
else:
raiseThe ShareFileClient provides comprehensive file management capabilities, enabling efficient file operations, content management, and advanced features like leasing and symbolic links for Azure File Share storage.
Install with Tessl CLI
npx tessl i tessl/pypi-azure-storage-file-share