|
1 | 1 | from datetime import datetime, timedelta |
2 | 2 | import mimetypes |
3 | 3 | import os |
| 4 | +from http import HTTPStatus |
4 | 5 | from pathlib import Path |
5 | 6 | from typing import Any, Callable, Dict, Iterable, Optional, Tuple, Union |
6 | 7 |
|
| 8 | +try: |
| 9 | + from typing import cast |
| 10 | +except ImportError: |
| 11 | + from typing_extensions import cast |
7 | 12 |
|
8 | 13 | from ..client import Client, register_client_class |
9 | 14 | from ..cloudpath import implementation_registry |
|
13 | 18 |
|
14 | 19 |
|
15 | 20 | try: |
16 | | - from azure.core.exceptions import ResourceNotFoundError |
| 21 | + from azure.core.exceptions import HttpResponseError, ResourceNotFoundError |
17 | 22 | from azure.core.credentials import AzureNamedKeyCredential |
18 | 23 |
|
19 | 24 | from azure.storage.blob import ( |
@@ -202,20 +207,29 @@ def _check_hns(self, cloud_path: AzureBlobPath) -> Optional[bool]: |
202 | 207 | try: |
203 | 208 | account_info = self.service_client.get_account_information() # type: ignore |
204 | 209 | self._hns_enabled = account_info.get("is_hns_enabled", False) # type: ignore |
| 210 | + |
| 211 | + # get_account_information() not supported with this credential; we have to fallback to |
| 212 | + # checking if the root directory exists and is a has 'metadata': {'hdi_isfolder': 'true'} |
205 | 213 | except ResourceNotFoundError: |
206 | | - # get_account_information() not supported with this credential; we have to fallback to |
207 | | - # checking if the root directory exists and is a has 'metadata': {'hdi_isfolder': 'true'} |
208 | | - root_dir = self.service_client.get_blob_client( |
209 | | - container=cloud_path.container, blob="/" |
210 | | - ) |
211 | | - self._hns_enabled = ( |
212 | | - root_dir.exists() |
213 | | - and root_dir.get_blob_properties().metadata.get("hdi_isfolder", False) |
214 | | - == "true" |
215 | | - ) |
| 214 | + return self._check_hns_root_metadata(cloud_path) |
| 215 | + except HttpResponseError as error: |
| 216 | + if error.status_code == HTTPStatus.FORBIDDEN: |
| 217 | + return self._check_hns_root_metadata(cloud_path) |
| 218 | + else: |
| 219 | + raise |
216 | 220 |
|
217 | 221 | return self._hns_enabled |
218 | 222 |
|
| 223 | + def _check_hns_root_metadata(self, cloud_path: AzureBlobPath) -> bool: |
| 224 | + root_dir = self.service_client.get_blob_client(container=cloud_path.container, blob="/") |
| 225 | + |
| 226 | + self._hns_enabled = ( |
| 227 | + root_dir.exists() |
| 228 | + and root_dir.get_blob_properties().metadata.get("hdi_isfolder", False) == "true" |
| 229 | + ) |
| 230 | + |
| 231 | + return cast(bool, self._hns_enabled) |
| 232 | + |
219 | 233 | def _get_metadata( |
220 | 234 | self, cloud_path: AzureBlobPath |
221 | 235 | ) -> Union["BlobProperties", "FileProperties", Dict[str, Any]]: |
|
0 commit comments