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