1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import io
17import os
18import tempfile
19from collections.abc import Callable, Iterator
20from datetime import datetime, timedelta, timezone
21from typing import IO, Any, Optional, TypeVar, Union
22from urllib.parse import urlparse
23
24from azure.core import MatchConditions
25from azure.core.exceptions import AzureError, HttpResponseError
26from azure.identity import DefaultAzureCredential
27from azure.storage.blob import (
28 BlobPrefix,
29 BlobSasPermissions,
30 BlobServiceClient,
31 PartialBatchErrorException,
32 generate_blob_sas,
33)
34
35from ..constants import DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT
36from ..signers.base import URLSigner
37from ..telemetry import Telemetry
38from ..types import (
39 AWARE_DATETIME_MIN,
40 Credentials,
41 CredentialsProvider,
42 ObjectMetadata,
43 PreconditionFailedError,
44 Range,
45 SignerType,
46 SymlinkHandling,
47)
48from ..utils import safe_makedirs, split_path, validate_attributes
49from .base import BaseStorageProvider
50
51_T = TypeVar("_T")
52
53PROVIDER = "azure"
54AZURE_CONNECTION_STRING_KEY = "connection"
55AZURE_CREDENTIAL_KEY = "azure_credential"
56
57MiB = 1024 * 1024
58
59MULTIPART_THRESHOLD = 64 * MiB
60MULTIPART_CHUNKSIZE = 32 * MiB
61IO_CHUNKSIZE = 32 * MiB
62PYTHON_MAX_CONCURRENCY = 8
63
64# Azure REST API only returns ``Content-MD5`` for GET ranges up to 4 MiB,
65# so download chunks must stay at or below this limit when ``validate_content`` is enabled.
66AZURE_CONTENT_MD5_RANGE_LIMIT_BYTES = 4 * MiB
67
68DEFAULT_PRESIGN_EXPIRES_IN = 3600
69
70# How long before delegation key expiry we treat the cached key as stale.
71_DELEGATION_KEY_REFRESH_BUFFER = timedelta(minutes=5)
72
73# Azure's maximum allowed delegation key lifetime is 7 days.
74_DELEGATION_KEY_LIFETIME = timedelta(days=7)
75
76
77def _sas_permissions_for_method(method: str) -> BlobSasPermissions:
78 """Return the minimal :class:`BlobSasPermissions` needed for *method*."""
79 m = method.upper()
80 if m in ("PUT", "POST"):
81 return BlobSasPermissions(write=True, create=True)
82 elif m == "DELETE":
83 return BlobSasPermissions(delete=True)
84 else:
85 # GET, HEAD, and any unrecognised method → read-only
86 return BlobSasPermissions(read=True)
87
88
89def _parse_account_name_from_url(account_url: str) -> str:
90 """Extract the storage account name from an Azure Blob Storage account URL."""
91 hostname = urlparse(account_url).hostname
92 if hostname is None:
93 raise ValueError(f"Invalid Azure account URL: {account_url!r}")
94 return hostname.split(".")[0]
95
96
97def _parse_connection_string(conn_str: str) -> dict[str, str]:
98 """Parse an Azure connection string (``AccountName=foo;AccountKey=bar;...``) into a dict."""
99 return dict(part.split("=", 1) for part in conn_str.split(";") if "=" in part)
100
101
[docs]
102class AzureURLSigner(URLSigner):
103 """
104 Generates Azure Blob Storage SAS (Shared Access Signature) URLs.
105
106 Supports two signing paths depending on which credential is provided:
107
108 * **Account key** – uses a static storage account key (parsed from a connection string).
109 * **User delegation key** – uses a time-limited key obtained via Azure Identity (e.g. workload
110 identity, managed identity). Callers are responsible for refreshing the signer when the
111 delegation key approaches expiry; see :py:meth:`AzureBlobStorageProvider._generate_presigned_url`.
112 """
113
114 def __init__(
115 self,
116 account_name: str,
117 account_url: str,
118 *,
119 account_key: Optional[str] = None,
120 user_delegation_key: Optional[Any] = None,
121 expires_in: int = DEFAULT_PRESIGN_EXPIRES_IN,
122 ) -> None:
123 if account_key is None and user_delegation_key is None:
124 raise ValueError("Either account_key or user_delegation_key must be provided.")
125 self._account_name = account_name
126 self._account_url = account_url.rstrip("/")
127 self._account_key = account_key
128 self._user_delegation_key = user_delegation_key
129 self._expires_in = expires_in
130
[docs]
131 def generate_presigned_url(self, path: str, *, method: str = "GET") -> str:
132 """
133 Generate a SAS URL for the given blob path.
134
135 :param path: Path in the form ``container/blob/name``.
136 :param method: HTTP method requested by the caller.
137 :return: A fully-qualified SAS URL.
138 """
139 container_name, blob_name = split_path(path)
140 expiry = datetime.now(timezone.utc) + timedelta(seconds=self._expires_in)
141
142 sas_kwargs: dict[str, Any] = {
143 "account_name": self._account_name,
144 "container_name": container_name,
145 "blob_name": blob_name,
146 "permission": _sas_permissions_for_method(method),
147 "expiry": expiry,
148 }
149
150 if self._account_key is not None:
151 sas_kwargs["account_key"] = self._account_key
152 else:
153 sas_kwargs["user_delegation_key"] = self._user_delegation_key
154
155 sas_token = generate_blob_sas(**sas_kwargs)
156 blob_url = f"{self._account_url}/{container_name}/{blob_name}"
157 return f"{blob_url}?{sas_token}"
158
159
[docs]
160class StaticAzureCredentialsProvider(CredentialsProvider):
161 """
162 A concrete implementation of the :py:class:`multistorageclient.types.CredentialsProvider` that provides static Azure credentials.
163 """
164
165 _connection: str
166
167 def __init__(self, connection: str):
168 """
169 Initializes the :py:class:`StaticAzureCredentialsProvider` with the provided connection string.
170
171 :param connection: The connection string for Azure Blob Storage authentication.
172 """
173 self._connection = connection
174
[docs]
175 def get_credentials(self) -> Credentials:
176 return Credentials(
177 access_key=self._connection,
178 secret_key="",
179 token=None,
180 expiration=None,
181 custom_fields={AZURE_CONNECTION_STRING_KEY: self._connection},
182 )
183
[docs]
184 def refresh_credentials(self) -> None:
185 pass
186
187
[docs]
188class DefaultAzureCredentialsProvider(CredentialsProvider):
189 """
190 A concrete implementation of the :py:class:`multistorageclient.types.CredentialsProvider` that uses Azure Identity's :py:class:`azure.identity.DefaultAzureCredential` to authenticate with Blob Storage.
191
192 See :py:class:`azure.identity.DefaultAzureCredential` for provider options.
193 """
194
195 def __init__(self, **kwargs: dict[str, Any]):
196 self._credential = DefaultAzureCredential(**kwargs)
197
[docs]
198 def get_credentials(self) -> Credentials:
199 return Credentials(
200 access_key="",
201 secret_key="",
202 token=None,
203 expiration=None,
204 custom_fields={AZURE_CREDENTIAL_KEY: self._credential},
205 )
206
[docs]
207 def refresh_credentials(self) -> None:
208 pass
209
210
[docs]
211class AzureBlobStorageProvider(BaseStorageProvider):
212 """
213 A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with Azure Blob Storage.
214 """
215
216 def __init__(
217 self,
218 endpoint_url: str,
219 base_path: str = "",
220 credentials_provider: Optional[CredentialsProvider] = None,
221 config_dict: Optional[dict[str, Any]] = None,
222 telemetry_provider: Optional[Callable[[], Telemetry]] = None,
223 **kwargs: Any,
224 ):
225 """
226 Initializes the :py:class:`AzureBlobStorageProvider` with the endpoint URL and optional credentials provider.
227
228 :param endpoint_url: The Azure storage account URL.
229 :param base_path: The root prefix path within the container where all operations will be scoped.
230 :param credentials_provider: The provider to retrieve Azure credentials.
231 :param config_dict: Resolved MSC config.
232 :param telemetry_provider: A function that provides a telemetry instance.
233 :param kwargs: Additional options including:
234 - ``multipart_threshold`` (int): File size threshold (bytes) for switching to parallel chunked transfers. Defaults to 64 MiB.
235 - ``multipart_chunksize`` (int): Block size (bytes) for chunked uploads. Defaults to 32 MiB.
236 - ``io_chunksize`` (int): Chunk size (bytes) for chunked downloads. Defaults to 32 MiB.
237 - ``max_concurrency`` (int): Number of parallel threads for chunked transfers. Defaults to 8.
238 - ``validate_content`` (bool): Opt-in client-side MD5 verification. Defaults to False.
239 """
240 super().__init__(
241 base_path=base_path,
242 provider_name=PROVIDER,
243 config_dict=config_dict,
244 telemetry_provider=telemetry_provider,
245 )
246
247 self._account_url = endpoint_url
248 self._credentials_provider = credentials_provider
249 # Cache static connection-string signing material used for per-request signers.
250 self._account_key_signing_material: Optional[tuple[str, str]] = None
251 # Cached delegation key and its expiry for DefaultAzureCredentialsProvider.
252 self._delegation_user_key: Optional[Any] = None
253 self._delegation_signer_expiry: Optional[datetime] = None
254 self._multipart_threshold = int(kwargs.get("multipart_threshold", MULTIPART_THRESHOLD))
255 self._multipart_chunksize = int(kwargs.get("multipart_chunksize", MULTIPART_CHUNKSIZE))
256 self._io_chunksize = int(kwargs.get("io_chunksize", IO_CHUNKSIZE))
257 self._max_concurrency = int(kwargs.get("max_concurrency", PYTHON_MAX_CONCURRENCY))
258 self._validate_content = kwargs.get("validate_content", False)
259 if not isinstance(self._validate_content, bool):
260 raise ValueError("Option 'validate_content' must be a boolean.")
261 if self._validate_content and self._io_chunksize > AZURE_CONTENT_MD5_RANGE_LIMIT_BYTES:
262 raise ValueError(
263 "Option 'validate_content=True' requires 'io_chunksize' to be "
264 f"<= {AZURE_CONTENT_MD5_RANGE_LIMIT_BYTES} bytes (4 MiB) because Azure only "
265 f"returns Content-MD5 for GET ranges within that limit. Got io_chunksize={self._io_chunksize}."
266 )
267
268 # https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/storage/azure-storage-blob#optional-configuration
269 client_optional_configuration_keys = {
270 "retry_total",
271 "retry_connect",
272 "retry_read",
273 "retry_status",
274 "connection_timeout",
275 "read_timeout",
276 }
277 self._client_optional_configuration: dict[str, Any] = {
278 key: value for key, value in kwargs.items() if key in client_optional_configuration_keys
279 }
280 if "connection_timeout" not in self._client_optional_configuration:
281 self._client_optional_configuration["connection_timeout"] = DEFAULT_CONNECT_TIMEOUT
282 if "read_timeout" not in self._client_optional_configuration:
283 self._client_optional_configuration["read_timeout"] = DEFAULT_READ_TIMEOUT
284
285 self._transfer_configuration: dict[str, Any] = {
286 "max_single_put_size": self._multipart_threshold,
287 "max_block_size": self._multipart_chunksize,
288 "max_single_get_size": self._multipart_threshold,
289 "max_chunk_get_size": self._io_chunksize,
290 }
291
292 self._blob_service_client = self._create_blob_service_client()
293
294 def _create_blob_service_client(self) -> BlobServiceClient:
295 """
296 Creates and configures the Azure BlobServiceClient using the current credentials.
297
298 :return: The configured BlobServiceClient.
299 """
300 combined_config = {**self._client_optional_configuration, **self._transfer_configuration}
301
302 if self._credentials_provider:
303 credentials = self._credentials_provider.get_credentials()
304
305 if isinstance(self._credentials_provider, StaticAzureCredentialsProvider):
306 return BlobServiceClient.from_connection_string(
307 credentials.get_custom_field(AZURE_CONNECTION_STRING_KEY), **combined_config
308 )
309 elif isinstance(self._credentials_provider, DefaultAzureCredentialsProvider):
310 return BlobServiceClient(
311 account_url=self._account_url,
312 credential=credentials.get_custom_field(AZURE_CREDENTIAL_KEY),
313 **combined_config,
314 )
315 else:
316 # Fallback to connection string if no built-in credentials provider is provided
317 return BlobServiceClient.from_connection_string(credentials.access_key, **combined_config)
318 else:
319 return BlobServiceClient(account_url=self._account_url, **combined_config)
320
321 def _refresh_blob_service_client_if_needed(self) -> None:
322 """
323 Refreshes the BlobServiceClient if the current credentials are expired.
324 """
325 if self._credentials_provider:
326 credentials = self._credentials_provider.get_credentials()
327 if credentials.is_expired():
328 self._credentials_provider.refresh_credentials()
329 self._blob_service_client = self._create_blob_service_client()
330
331 def _translate_errors(
332 self,
333 func: Callable[[], _T],
334 operation: str,
335 container: str,
336 blob: str,
337 ) -> _T:
338 """
339 Translates errors like timeouts and client errors.
340
341 :param func: The function that performs the actual Azure Blob Storage operation.
342 :param operation: The type of operation being performed (e.g., "PUT", "GET", "DELETE").
343 :param container: The name of the Azure container involved in the operation.
344 :param blob: The name of the blob within the Azure container.
345
346 :return The result of the Azure Blob Storage operation, typically the return value of the `func` callable.
347 """
348 try:
349 return func()
350 except HttpResponseError as error:
351 status_code = error.status_code if error.status_code else -1
352 error_info = f"status_code: {error.status_code}, reason: {error.reason}"
353 if status_code == 404:
354 raise FileNotFoundError(f"Object {container}/{blob} does not exist.") # pylint: disable=raise-missing-from
355 elif status_code == 412:
356 # raised when If-Match or If-Modified fails
357 raise PreconditionFailedError(
358 f"Failed to {operation} object(s) at {container}/{blob}. {error_info}"
359 ) from error
360 else:
361 raise RuntimeError(f"Failed to {operation} object(s) at {container}/{blob}. {error_info}") from error
362 except AzureError as error:
363 error_info = f"message: {error.message}"
364 raise RuntimeError(f"Failed to {operation} object(s) at {container}/{blob}. {error_info}") from error
365 except FileNotFoundError:
366 raise
367 except Exception as error:
368 raise RuntimeError(
369 f"Failed to {operation} object(s) at {container}/{blob}. error_type: {type(error).__name__}, error: {error}"
370 ) from error
371
372 def _put_object(
373 self,
374 path: str,
375 body: bytes,
376 if_match: Optional[str] = None,
377 if_none_match: Optional[str] = None,
378 attributes: Optional[dict[str, str]] = None,
379 ) -> int:
380 """
381 Uploads an object to Azure Blob Storage.
382
383 :param path: The path to the object to upload.
384 :param body: The content of the object to upload.
385 :param if_match: Optional ETag to match against the object.
386 :param if_none_match: Optional ETag to match against the object.
387 :param attributes: Optional attributes to attach to the object.
388 """
389 container_name, blob_name = split_path(path)
390 self._refresh_blob_service_client_if_needed()
391
392 def _invoke_api() -> int:
393 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
394
395 kwargs: dict[str, Any] = {
396 "data": body,
397 "overwrite": True,
398 "max_concurrency": self._max_concurrency,
399 "validate_content": self._validate_content,
400 }
401
402 validated_attributes = validate_attributes(attributes)
403 if validated_attributes:
404 kwargs["metadata"] = validated_attributes
405
406 if if_match:
407 kwargs["match_condition"] = MatchConditions.IfNotModified
408 kwargs["etag"] = if_match
409
410 if if_none_match:
411 if if_none_match == "*":
412 raise NotImplementedError("if_none_match='*' is not supported for Azure")
413 kwargs["match_condition"] = MatchConditions.IfModified
414 kwargs["etag"] = if_none_match
415
416 blob_client.upload_blob(**kwargs)
417
418 return len(body)
419
420 return self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
421
422 def _get_object(self, path: str, byte_range: Optional[Range] = None) -> bytes:
423 container_name, blob_name = split_path(path)
424 self._refresh_blob_service_client_if_needed()
425
426 def _invoke_api() -> bytes:
427 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
428 if byte_range:
429 stream = blob_client.download_blob(
430 offset=byte_range.offset,
431 length=byte_range.size,
432 validate_content=self._validate_content,
433 )
434 else:
435 stream = blob_client.download_blob(
436 max_concurrency=self._max_concurrency,
437 validate_content=self._validate_content,
438 )
439 return stream.readall()
440
441 return self._translate_errors(_invoke_api, operation="GET", container=container_name, blob=blob_name)
442
443 def _copy_object(self, src_path: str, dest_path: str) -> int:
444 src_container, src_blob = split_path(src_path)
445 dest_container, dest_blob = split_path(dest_path)
446 self._refresh_blob_service_client_if_needed()
447
448 src_object = self._get_object_metadata(src_path)
449
450 def _invoke_api() -> int:
451 src_blob_client = self._blob_service_client.get_blob_client(container=src_container, blob=src_blob)
452 dest_blob_client = self._blob_service_client.get_blob_client(container=dest_container, blob=dest_blob)
453 dest_blob_client.start_copy_from_url(src_blob_client.url)
454
455 return src_object.content_length
456
457 return self._translate_errors(_invoke_api, operation="COPY", container=src_container, blob=src_blob)
458
459 def _delete_object(self, path: str, if_match: Optional[str] = None) -> None:
460 container_name, blob_name = split_path(path)
461 self._refresh_blob_service_client_if_needed()
462
463 def _invoke_api() -> None:
464 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
465 # If if_match is provided, use if_match for conditional deletion
466 if if_match:
467 blob_client.delete_blob(etag=if_match, match_condition=MatchConditions.IfNotModified)
468 else:
469 # No if_match provided, perform unconditional deletion
470 blob_client.delete_blob()
471
472 return self._translate_errors(_invoke_api, operation="DELETE", container=container_name, blob=blob_name)
473
474 def _delete_objects(self, paths: list[str]) -> None:
475 if not paths:
476 return
477
478 by_container: dict[str, list[str]] = {}
479 for p in paths:
480 container_name, blob_name = split_path(p)
481 by_container.setdefault(container_name, []).append(blob_name)
482 self._refresh_blob_service_client_if_needed()
483
484 AZURE_BATCH_LIMIT = 256
485
486 def _invoke_api() -> None:
487 for container_name, blob_names in by_container.items():
488 container_client = self._blob_service_client.get_container_client(container=container_name)
489 for i in range(0, len(blob_names), AZURE_BATCH_LIMIT):
490 chunk = blob_names[i : i + AZURE_BATCH_LIMIT]
491 try:
492 responses = container_client.delete_blobs(*chunk, raise_on_any_failure=False)
493 except PartialBatchErrorException as error:
494 responses = error.parts
495 for response in responses:
496 status_code = response.status_code
497 if 200 <= status_code < 300 or status_code == 404:
498 continue
499 raise RuntimeError(
500 f"Azure batch delete failed with status_code: {status_code}, response: {response.text}"
501 )
502
503 container_desc = "(" + "|".join(by_container) + ")"
504 blob_desc = "(" + "|".join(str(len(blob_names)) for blob_names in by_container.values()) + " keys)"
505 self._translate_errors(_invoke_api, operation="DELETE_MANY", container=container_desc, blob=blob_desc)
506
507 def _is_dir(self, path: str) -> bool:
508 # Ensure the path ends with '/' to mimic a directory
509 path = self._append_delimiter(path)
510
511 container_name, prefix = split_path(path)
512 self._refresh_blob_service_client_if_needed()
513
514 def _invoke_api() -> bool:
515 # List objects with the given prefix
516 container_client = self._blob_service_client.get_container_client(container=container_name)
517 blobs = container_client.walk_blobs(name_starts_with=prefix, delimiter="/")
518 # Check if there are any contents or common prefixes
519 return any(True for _ in blobs)
520
521 return self._translate_errors(_invoke_api, operation="LIST", container=container_name, blob=prefix)
522
523 def _make_symlink(self, path: str, target: str) -> None:
524 container_name, blob_name = split_path(path)
525 target_container, target_key = split_path(target)
526 if container_name != target_container:
527 raise ValueError(f"Cannot create cross-container symlink: '{container_name}' -> '{target_container}'.")
528 relative_target = ObjectMetadata.encode_symlink_target(blob_name, target_key)
529 self._refresh_blob_service_client_if_needed()
530
531 def _invoke_api() -> None:
532 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
533 blob_client.upload_blob(
534 data=b"",
535 overwrite=True,
536 metadata={"msc_symlink_target": relative_target},
537 )
538
539 self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
540
541 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
542 container_name, blob_name = split_path(path)
543 if path.endswith("/") or (container_name and not blob_name):
544 # If path ends with "/" or empty blob name is provided, then assume it's a "directory",
545 # which metadata is not guaranteed to exist for cases such as
546 # "virtual prefix" that was never explicitly created.
547 if self._is_dir(path):
548 return ObjectMetadata(
549 key=self._append_delimiter(path),
550 type="directory",
551 content_length=0,
552 last_modified=AWARE_DATETIME_MIN,
553 )
554 else:
555 raise FileNotFoundError(f"Directory {path} does not exist.")
556 else:
557 self._refresh_blob_service_client_if_needed()
558
559 def _invoke_api() -> ObjectMetadata:
560 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
561 properties = blob_client.get_blob_properties()
562 user_metadata = dict(properties.metadata) if properties.metadata else None
563 symlink_target = user_metadata.get("msc_symlink_target") if user_metadata else None
564 return ObjectMetadata(
565 key=path,
566 content_length=properties.size,
567 content_type=properties.content_settings.content_type,
568 last_modified=properties.last_modified,
569 etag=properties.etag.strip('"') if properties.etag else "",
570 metadata=user_metadata,
571 symlink_target=symlink_target,
572 )
573
574 try:
575 return self._translate_errors(_invoke_api, operation="HEAD", container=container_name, blob=blob_name)
576 except FileNotFoundError as error:
577 if strict:
578 # If the object does not exist on the given path, we will append a trailing slash and
579 # check if the path is a directory.
580 path = self._append_delimiter(path)
581 if self._is_dir(path):
582 return ObjectMetadata(
583 key=path,
584 type="directory",
585 content_length=0,
586 last_modified=AWARE_DATETIME_MIN,
587 )
588 raise error
589
590 def _list_objects(
591 self,
592 path: str,
593 start_after: Optional[str] = None,
594 end_at: Optional[str] = None,
595 include_directories: bool = False,
596 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
597 ) -> Iterator[ObjectMetadata]:
598 container_name, prefix = split_path(path)
599
600 # Get the prefix of the start_after and end_at paths relative to the bucket.
601 if start_after:
602 _, start_after = split_path(start_after)
603 if end_at:
604 _, end_at = split_path(end_at)
605
606 self._refresh_blob_service_client_if_needed()
607
608 def _invoke_api() -> Iterator[ObjectMetadata]:
609 container_client = self._blob_service_client.get_container_client(container=container_name)
610 # Azure has no start key option like other object stores.
611 if include_directories:
612 blobs = container_client.walk_blobs(name_starts_with=prefix, delimiter="/", include=["metadata"])
613 else:
614 blobs = container_client.list_blobs(name_starts_with=prefix, include=["metadata"])
615 # Azure guarantees lexicographical order.
616 for blob in blobs:
617 if isinstance(blob, BlobPrefix):
618 prefix_key = blob.name.rstrip("/")
619 # Filter by start_after and end_at if specified
620 if (start_after is None or start_after < prefix_key) and (end_at is None or prefix_key <= end_at):
621 yield ObjectMetadata(
622 key=os.path.join(container_name, prefix_key),
623 type="directory",
624 content_length=0,
625 last_modified=AWARE_DATETIME_MIN,
626 )
627 elif end_at is not None and end_at < prefix_key:
628 return
629 else:
630 key = blob.name
631 if (start_after is None or start_after < key) and (end_at is None or key <= end_at):
632 if key.endswith("/"):
633 if include_directories:
634 yield ObjectMetadata(
635 key=os.path.join(container_name, key.rstrip("/")),
636 type="directory",
637 content_length=0,
638 last_modified=blob.last_modified,
639 )
640 else:
641 user_metadata = dict(blob.metadata) if blob.metadata else None
642 symlink_target = user_metadata.get("msc_symlink_target") if user_metadata else None
643 yield ObjectMetadata(
644 key=os.path.join(container_name, key),
645 content_length=blob.size,
646 content_type=blob.content_settings.content_type,
647 last_modified=blob.last_modified,
648 etag=blob.etag.strip('"') if blob.etag else "",
649 symlink_target=symlink_target,
650 )
651 elif end_at is not None and end_at < key:
652 return
653
654 return self._translate_errors(_invoke_api, operation="LIST", container=container_name, blob=prefix)
655
656 def _generate_presigned_url(
657 self,
658 path: str,
659 *,
660 method: str = "GET",
661 signer_type: Optional[SignerType] = None,
662 signer_options: Optional[dict[str, Any]] = None,
663 ) -> str:
664 """
665 Generate a SAS URL for a blob in Azure Blob Storage.
666
667 :param path: Path in the form ``container/blob/name``.
668 :param method: HTTP method requested by the caller.
669 :param signer_type: Must be ``None`` or :py:attr:`SignerType.AZURE`.
670 :param signer_options: Optional dict; supports ``expires_in`` (int, seconds).
671 :return: A fully-qualified SAS URL.
672 :raises ValueError: If *signer_type* is not ``None`` / ``SignerType.AZURE``, or if the
673 configured credential type does not support SAS generation.
674 """
675 if signer_type is not None and signer_type != SignerType.AZURE:
676 raise ValueError(f"Unsupported signer type for Azure provider: {signer_type!r}")
677
678 options = signer_options or {}
679 expires_in = int(options.get("expires_in", DEFAULT_PRESIGN_EXPIRES_IN))
680
681 self._refresh_blob_service_client_if_needed()
682
683 if isinstance(self._credentials_provider, StaticAzureCredentialsProvider):
684 # Account key path: cache parsed AccountName + AccountKey, then sign per request.
685 if self._account_key_signing_material is None:
686 conn_str = self._credentials_provider.get_credentials().get_custom_field(AZURE_CONNECTION_STRING_KEY)
687 parsed = _parse_connection_string(conn_str)
688 self._account_key_signing_material = (parsed["AccountName"], parsed["AccountKey"])
689 account_name, account_key = self._account_key_signing_material
690 signer = AzureURLSigner(
691 account_name=account_name,
692 account_url=self._account_url,
693 account_key=account_key,
694 expires_in=expires_in,
695 )
696
697 elif isinstance(self._credentials_provider, DefaultAzureCredentialsProvider):
698 # User delegation key path: refresh when the cached key is within the
699 # refresh buffer of its own expiry or has not been fetched yet.
700 now = datetime.now(timezone.utc)
701 if (
702 self._delegation_user_key is None
703 or self._delegation_signer_expiry is None
704 or now >= self._delegation_signer_expiry - _DELEGATION_KEY_REFRESH_BUFFER
705 ):
706 key_expiry = now + _DELEGATION_KEY_LIFETIME
707 self._delegation_user_key = self._blob_service_client.get_user_delegation_key(
708 key_start_time=now,
709 key_expiry_time=key_expiry,
710 )
711 self._delegation_signer_expiry = key_expiry
712 signer = AzureURLSigner(
713 account_name=_parse_account_name_from_url(self._account_url),
714 account_url=self._account_url,
715 user_delegation_key=self._delegation_user_key,
716 expires_in=expires_in,
717 )
718
719 else:
720 raise ValueError(
721 "Azure presigned URLs require StaticAzureCredentialsProvider (connection string) or "
722 "DefaultAzureCredentialsProvider (Azure Identity). "
723 f"Got: {type(self._credentials_provider).__name__!r}"
724 )
725
726 return signer.generate_presigned_url(path, method=method)
727
728 @property
729 def supports_parallel_listing(self) -> bool:
730 return True
731
732 def _upload_file(self, remote_path: str, f: Union[str, IO], attributes: Optional[dict[str, str]] = None) -> int:
733 container_name, blob_name = split_path(remote_path)
734 file_size: int = 0
735 self._refresh_blob_service_client_if_needed()
736
737 validated_attributes = validate_attributes(attributes)
738 if isinstance(f, str):
739 file_size = os.path.getsize(f)
740
741 if file_size <= self._multipart_threshold:
742
743 def _invoke_api() -> int:
744 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
745 with open(f, "rb") as data:
746 blob_client.upload_blob(
747 data,
748 overwrite=True,
749 metadata=validated_attributes or {},
750 validate_content=self._validate_content,
751 )
752 return file_size
753
754 return self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
755
756 def _invoke_api() -> int:
757 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
758 with open(f, "rb") as data:
759 blob_client.upload_blob(
760 data,
761 overwrite=True,
762 metadata=validated_attributes or {},
763 max_concurrency=self._max_concurrency,
764 validate_content=self._validate_content,
765 )
766 return file_size
767
768 return self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
769 else:
770 if isinstance(f, io.StringIO):
771 fp: IO = io.BytesIO(f.getvalue().encode("utf-8")) # type: ignore
772 else:
773 fp = f
774
775 fp.seek(0, io.SEEK_END)
776 file_size = fp.tell()
777 fp.seek(0)
778
779 if file_size <= self._multipart_threshold:
780
781 def _invoke_api() -> int:
782 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
783 blob_client.upload_blob(
784 fp,
785 overwrite=True,
786 metadata=validated_attributes or {},
787 validate_content=self._validate_content,
788 )
789 return file_size
790
791 return self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
792
793 def _invoke_api() -> int:
794 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
795 blob_client.upload_blob(
796 fp,
797 overwrite=True,
798 metadata=validated_attributes or {},
799 max_concurrency=self._max_concurrency,
800 validate_content=self._validate_content,
801 )
802 return file_size
803
804 return self._translate_errors(_invoke_api, operation="PUT", container=container_name, blob=blob_name)
805
806 def _download_file(self, remote_path: str, f: Union[str, IO], metadata: Optional[ObjectMetadata] = None) -> int:
807 if metadata is None:
808 metadata = self._get_object_metadata(remote_path)
809
810 container_name, blob_name = split_path(remote_path)
811 self._refresh_blob_service_client_if_needed()
812
813 if isinstance(f, str):
814 if os.path.dirname(f):
815 safe_makedirs(os.path.dirname(f))
816
817 if metadata.content_length <= self._multipart_threshold:
818
819 def _invoke_api() -> int:
820 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
821 temp_file_path: str | None = None
822 try:
823 with tempfile.NamedTemporaryFile(
824 mode="wb", delete=False, dir=os.path.dirname(f), prefix="."
825 ) as fp:
826 temp_file_path = fp.name
827 stream = blob_client.download_blob(validate_content=self._validate_content)
828 fp.write(stream.readall())
829 os.rename(src=temp_file_path, dst=f)
830 except BaseException:
831 if temp_file_path and os.path.exists(temp_file_path):
832 os.unlink(temp_file_path)
833 raise
834 return metadata.content_length
835
836 return self._translate_errors(_invoke_api, operation="GET", container=container_name, blob=blob_name)
837
838 def _invoke_api() -> int:
839 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
840 temp_file_path: str | None = None
841 try:
842 with tempfile.NamedTemporaryFile(mode="wb", delete=False, dir=os.path.dirname(f), prefix=".") as fp:
843 temp_file_path = fp.name
844 stream = blob_client.download_blob(
845 max_concurrency=self._max_concurrency,
846 validate_content=self._validate_content,
847 )
848 stream.readinto(fp)
849 os.rename(src=temp_file_path, dst=f)
850 except BaseException:
851 if temp_file_path and os.path.exists(temp_file_path):
852 os.unlink(temp_file_path)
853 raise
854 return metadata.content_length
855
856 return self._translate_errors(_invoke_api, operation="GET", container=container_name, blob=blob_name)
857 else:
858 if metadata.content_length <= self._multipart_threshold:
859
860 def _invoke_api() -> int:
861 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
862 stream = blob_client.download_blob(validate_content=self._validate_content)
863 if isinstance(f, io.StringIO):
864 f.write(stream.readall().decode("utf-8"))
865 else:
866 f.write(stream.readall())
867 return metadata.content_length
868
869 return self._translate_errors(_invoke_api, operation="GET", container=container_name, blob=blob_name)
870
871 def _invoke_api() -> int:
872 blob_client = self._blob_service_client.get_blob_client(container=container_name, blob=blob_name)
873 stream = blob_client.download_blob(
874 max_concurrency=self._max_concurrency,
875 validate_content=self._validate_content,
876 )
877 if isinstance(f, io.StringIO):
878 temp_file_path: str | None = None
879 try:
880 with tempfile.NamedTemporaryFile(mode="wb", delete=False, prefix=".") as tmp:
881 temp_file_path = tmp.name
882 stream.readinto(tmp)
883 with open(temp_file_path, "r") as tmp_read:
884 f.write(tmp_read.read())
885 finally:
886 if temp_file_path and os.path.exists(temp_file_path):
887 os.unlink(temp_file_path)
888 else:
889 stream.readinto(f)
890 return metadata.content_length
891
892 return self._translate_errors(_invoke_api, operation="GET", container=container_name, blob=blob_name)