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
directory-client.mddocs/
ShareDirectoryClient - Directory Operations
The ShareDirectoryClient provides operations for managing directories within file shares, including creating and deleting directories, listing contents, managing metadata, and handling file operations within directories.
Import and Initialization
from azure.storage.fileshare import ShareDirectoryClient, ShareFileClient
from azure.core.credentials import AzureNamedKeyCredential
from typing import Optional, Union, Dict, Any, List, IO, AnyStr, IterableConstructor
class ShareDirectoryClient:
def __init__(
self,
account_url: str,
share_name: str,
directory_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 ShareDirectoryClient for a specific directory.
Parameters:
account_url: The URL to the file service endpoint
share_name: Name of the share containing the directory
directory_path: Path to the directory 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_directory_url(
cls,
directory_url: str,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Create ShareDirectoryClient from directory URL.
Parameters:
directory_url: Complete URL to the directory
credential: Authentication credential
**kwargs: Additional client configuration options
Returns:
ShareDirectoryClient: Configured client instance
"""
@classmethod
def from_connection_string(
cls,
conn_str: str,
share_name: str,
directory_path: str,
credential: Optional[Union[str, Dict[str, str], AzureNamedKeyCredential, AzureSasCredential, TokenCredential]] = None,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Create ShareDirectoryClient from connection string.
Parameters:
conn_str: Azure Storage connection string
share_name: Name of the share
directory_path: Path to the directory from share root
credential: Optional credential to override connection string auth
**kwargs: Additional client configuration options
Returns:
ShareDirectoryClient: Configured client instance
"""Client Factory Methods
def get_file_client(
self,
file_name: str,
**kwargs: Any
) -> ShareFileClient:
"""
Get a client to interact with a specific file in this directory.
Parameters:
file_name: Name of the file within this directory
Returns:
ShareFileClient: Client configured for the file
"""
def get_subdirectory_client(
self,
directory_name: str,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Get a client to interact with a subdirectory.
Parameters:
directory_name: Name of the subdirectory within this directory
Returns:
ShareDirectoryClient: Client configured for the subdirectory
"""Directory Lifecycle Operations
def create_directory(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Creates a new directory under the share.
Parameters:
metadata: Dict of name-value pairs for directory metadata
file_attributes: File attributes for the directory (NTFSAttributes)
file_creation_time: Creation time for the file/directory
file_last_write_time: Last write time for the file/directory
file_permission: Security descriptor string or permission key
file_permission_key: Key for a previously created permission
timeout: Request timeout in seconds
Returns:
Dict containing directory creation response with ETag and properties
"""
def delete_directory(
self,
**kwargs: Any
) -> None:
"""
Marks the specified directory for deletion.
Parameters:
timeout: Request timeout in seconds
"""
def rename_directory(
self,
new_name: str,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Rename a directory and return a client for the renamed directory.
Parameters:
new_name: New name for the directory
overwrite: Whether to overwrite existing directory with same name
ignore_readonly: Whether to ignore readonly attribute on destination
timeout: Request timeout in seconds
Returns:
ShareDirectoryClient: Client for the renamed directory
"""
def exists(
self,
**kwargs: Any
) -> bool:
"""
Returns True if the directory exists, False otherwise.
Parameters:
timeout: Request timeout in seconds
Returns:
bool: True if directory exists
"""Directory Content Management
def list_directories_and_files(
self,
name_starts_with: Optional[str] = None,
marker: Optional[str] = None,
**kwargs: Any
) -> ItemPaged[Dict[str, Any]]:
"""
Lists all directories and files under this directory.
Parameters:
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
Note:
Each dict contains:
- 'name': Name of the item
- 'is_directory': True for directories, False for files
- 'content_length': Size in bytes (files only)
- Additional properties for files and directories
"""
def create_subdirectory(
self,
directory_name: str,
**kwargs: Any
) -> ShareDirectoryClient:
"""
Creates a new subdirectory under this directory.
Parameters:
directory_name: Name of the subdirectory to create
metadata: Dict of name-value pairs for directory metadata
file_attributes: File attributes for the directory
file_creation_time: Creation time for the directory
file_last_write_time: Last write time for the directory
file_permission: Security descriptor string or permission key
timeout: Request timeout in seconds
Returns:
ShareDirectoryClient: Client for the newly created subdirectory
"""
def delete_subdirectory(
self,
directory_name: str,
**kwargs: Any
) -> None:
"""
Deletes a subdirectory under this directory.
Parameters:
directory_name: Name of the subdirectory to delete
timeout: Request timeout in seconds
"""File Operations
def upload_file(
self,
file_name: str,
data: Union[bytes, str, Iterable[AnyStr], IO[AnyStr]],
length: Optional[int] = None,
**kwargs: Any
) -> ShareFileClient:
"""
Creates a new file in the directory and returns a ShareFileClient.
Parameters:
file_name: Name of the file to create
data: File content to upload
length: Length of the data in bytes (required for streams)
overwrite: Whether to overwrite existing file
max_concurrency: Number of parallel upload threads
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
timeout: Request timeout in seconds
Returns:
ShareFileClient: Client for the newly created file
"""
def delete_file(
self,
file_name: str,
**kwargs: Any
) -> None:
"""
Marks the specified file for deletion.
Parameters:
file_name: Name of the file to delete
timeout: Request timeout in seconds
"""Directory Properties and Metadata
def get_directory_properties(
self,
**kwargs: Any
) -> DirectoryProperties:
"""
Returns all user-defined metadata and system properties for the directory.
Parameters:
timeout: Request timeout in seconds
Returns:
DirectoryProperties: Object containing all directory properties
"""
def set_directory_metadata(
self,
metadata: Dict[str, Any],
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets user-defined metadata for the directory.
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
"""
def set_http_headers(
self,
**kwargs: Any
) -> Dict[str, Any]:
"""
Sets HTTP headers on the directory.
Parameters:
file_attributes: File attributes for the directory
file_creation_time: Creation time for the directory
file_last_write_time: Last write time for the directory
file_permission: Security descriptor string or permission key
file_permission_key: Key for a previously created permission
timeout: Request timeout in seconds
Returns:
Dict containing response with ETag and last modified time
"""Handle Management
def list_handles(
self,
recursive: bool = False,
**kwargs: Any
) -> ItemPaged[Handle]:
"""
Lists handles for directory, subdirectories, and files.
Parameters:
recursive: Whether to list handles recursively in subdirectories
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,
recursive: bool = False,
**kwargs: Any
) -> Dict[str, int]:
"""
Closes all open file handles for the directory, subdirectories, and files.
Parameters:
recursive: Whether to close handles recursively in subdirectories
timeout: Request timeout in seconds
Returns:
Dict containing number of handles closed
"""Properties
@property
def url(self) -> str:
"""The full endpoint URL to the directory."""
@property
def share_name(self) -> str:
"""The name of the share containing this directory."""
@property
def directory_path(self) -> str:
"""The path to the directory 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 Directory Operations
from azure.storage.fileshare import ShareDirectoryClient
from azure.core.credentials import AzureNamedKeyCredential
# Initialize directory client
credential = AzureNamedKeyCredential("myaccount", "mykey")
directory_client = ShareDirectoryClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
directory_path="projects/2024",
credential=credential
)
# Check if directory exists
if not directory_client.exists():
# Create directory with metadata
directory_client.create_directory(
metadata={"created_by": "automation", "project": "data_migration"}
)
print("Directory created successfully")
else:
print("Directory already exists")
# Get directory properties
properties = directory_client.get_directory_properties()
print(f"Directory created: {properties.creation_time}")
print(f"Last modified: {properties.last_modified}")
print(f"Metadata: {properties.metadata}")Directory Content Management
# List all contents
contents = list(directory_client.list_directories_and_files())
for item in contents:
item_type = "Directory" if item.get('is_directory') else "File"
size_info = f" ({item.get('content_length', 0)} bytes)" if not item.get('is_directory') else ""
print(f"{item_type}: {item['name']}{size_info}")
# Filter by prefix
reports = list(directory_client.list_directories_and_files(
name_starts_with="report_"
))
# Create subdirectories with different attributes
from azure.storage.fileshare import NTFSAttributes
from datetime import datetime
archive_dir = directory_client.create_subdirectory(
"archive",
metadata={"type": "archive", "retention": "7years"},
file_attributes=NTFSAttributes.Directory | NTFSAttributes.ReadOnly
)
temp_dir = directory_client.create_subdirectory(
"temp",
file_creation_time=datetime.utcnow()
)File Operations Within Directory
from azure.storage.fileshare import ContentSettings
# Upload text file
text_content = "This is a sample document."
file_client = directory_client.upload_file(
file_name="readme.txt",
data=text_content.encode('utf-8'),
content_settings=ContentSettings(
content_type="text/plain",
content_encoding="utf-8"
),
metadata={"author": "system", "version": "1.0"}
)
# Upload binary file
with open("local_image.jpg", "rb") as image_data:
image_client = directory_client.upload_file(
file_name="image.jpg",
data=image_data,
content_settings=ContentSettings(content_type="image/jpeg"),
overwrite=True
)
# Get file client for existing file
existing_file = directory_client.get_file_client("config.json")
# Download file through directory client
config_data = existing_file.download_file().readall()
print(f"Downloaded {len(config_data)} bytes")Batch Operations
# Upload multiple files
files_to_upload = [
("report1.pdf", "reports/monthly_report.pdf"),
("report2.pdf", "reports/quarterly_report.pdf"),
("summary.txt", "summaries/executive_summary.txt")
]
uploaded_clients = []
for file_name, local_path in files_to_upload:
with open(local_path, "rb") as file_data:
client = directory_client.upload_file(
file_name=file_name,
data=file_data,
overwrite=True
)
uploaded_clients.append(client)
print(f"Uploaded: {file_name}")
# Delete multiple files
files_to_delete = ["temp1.txt", "temp2.txt", "old_backup.zip"]
for file_name in files_to_delete:
try:
directory_client.delete_file(file_name)
print(f"Deleted: {file_name}")
except Exception as e:
print(f"Failed to delete {file_name}: {e}")Directory Tree Operations
# Create nested directory structure
base_dirs = ["projects", "archives", "temp"]
sub_dirs = ["2024", "2023", "2022"]
for base_dir in base_dirs:
base_client = directory_client.create_subdirectory(base_dir)
for sub_dir in sub_dirs:
year_client = base_client.create_subdirectory(sub_dir)
print(f"Created: {base_dir}/{sub_dir}")
# Recursive content listing
def list_directory_tree(dir_client, prefix=""):
"""Recursively list directory contents."""
items = list(dir_client.list_directories_and_files())
for item in items:
print(f"{prefix}{item['name']}")
if item.get('is_directory'):
sub_client = dir_client.get_subdirectory_client(item['name'])
list_directory_tree(sub_client, prefix + " ")
print("Directory tree:")
list_directory_tree(directory_client)Handle Management
# List open handles
handles = list(directory_client.list_handles(recursive=True))
print(f"Found {len(handles)} open handles")
for handle in handles:
print(f"Handle {handle.id}:")
print(f" Path: {handle.path}")
print(f" Client: {handle.client_name}")
print(f" Client IP: {handle.client_ip}")
print(f" Open time: {handle.open_time}")
print(f" Access rights: {handle.access_rights}")
# Close specific handle
if handles:
closed_count = directory_client.close_handle(handles[0])
print(f"Closed {closed_count['closed_handles_count']} handle(s)")
# Close all handles (useful for maintenance)
all_closed = directory_client.close_all_handles(recursive=True)
print(f"Closed {all_closed['closed_handles_count']} total handle(s)")Metadata and Properties Management
from azure.storage.fileshare import NTFSAttributes
from datetime import datetime, timezone
# Update directory metadata
directory_client.set_directory_metadata({
"department": "engineering",
"project_code": "PROJ-2024-001",
"retention_policy": "5years",
"last_review": datetime.now(timezone.utc).isoformat()
})
# Set directory attributes and timestamps
directory_client.set_http_headers(
file_attributes=NTFSAttributes.Directory | NTFSAttributes.Archive,
file_creation_time=datetime(2024, 1, 1),
file_last_write_time=datetime.now(timezone.utc)
)
# Get updated properties
properties = directory_client.get_directory_properties()
print(f"Attributes: {properties.file_attributes}")
print(f"Creation time: {properties.creation_time}")
print(f"Last write time: {properties.last_write_time}")
print(f"Updated metadata: {properties.metadata}")Directory Renaming and Movement
# Rename directory
try:
new_directory_client = directory_client.rename_directory(
new_name="projects_2024_archived",
overwrite=False # Fail if destination exists
)
print(f"Directory renamed to: {new_directory_client.directory_path}")
except Exception as e:
print(f"Rename failed: {e}")
# Move directory (rename with path change)
current_client = ShareDirectoryClient(
account_url="https://myaccount.file.core.windows.net",
share_name="documents",
directory_path="temp/project_a",
credential=credential
)
moved_client = current_client.rename_directory(
new_name="archive/completed/project_a"
)
print(f"Directory moved to: {moved_client.directory_path}")Error Handling and Validation
from azure.core.exceptions import ResourceExistsError, ResourceNotFoundError
# Safe directory creation
try:
directory_client.create_directory()
except ResourceExistsError:
print("Directory already exists, updating metadata...")
directory_client.set_directory_metadata({
"updated": datetime.now().isoformat()
})
# Validate before operations
if directory_client.exists():
properties = directory_client.get_directory_properties()
# Check if directory is empty before deletion
contents = list(directory_client.list_directories_and_files())
if not contents:
directory_client.delete_directory()
print("Empty directory deleted")
else:
print(f"Directory contains {len(contents)} items, skipping deletion")
else:
print("Directory does not exist")
# Handle file operation errors
try:
directory_client.delete_file("nonexistent.txt")
except ResourceNotFoundError:
print("File not found for deletion")The ShareDirectoryClient provides comprehensive directory management capabilities, enabling you to create hierarchical folder structures, manage directory contents, handle file operations, and maintain directory metadata and properties efficiently.
Install with Tessl CLI
npx tessl i tessl/pypi-azure-storage-file-share