Repository: fsspec/adlfs (language: python)

Issue:
Container exists / Permissions issue
Hi there

I have a sas_token that somehow does not allow asking for the container properties, only for folders inside. This breaks adlfs's _mkdir which is called by a third-party lib I don't control (pyarrow). Is there some way to not ignore the error on _container_exists if a sas_token was provided? 




Discussion / hints:
Monkey patching works as a workaround:

```python

async def _noop_container_exists(self = None, container_name = None):
        return True 

fs, path = fsspec.core.url_to_fs(url, **storage_options)
if "_container_exists" in dir(fs):
    # monkey patch adfs because of missing permissions 
    fs._container_exists = _noop_container_exists # https://github.com/fsspec/adlfs/issues/384

```
Yeah, this is pretty unfortunate. Even if you're using Azure AD RBAC for auth, you might only have permission within the container.

FWIW, `azcopy` will print out a warning in some cases like this.

I would propose two things:

1. Place the main part of `mkdir` in a `try / except` block. If we do get a permission error,  we emit a warning explaining that we couldn't create the storage container, but things may work if the container already exists
2. A new parameter in `AzureBlobFileSystem` that controls whether that warning is emitted. By default we should warn. I suppose this could be a 3-way switch: warn / ignore / raise an error.
This one is really a bad one. Would you accept a PR that implements the try / except fix and just warns?
Yep, if you have a chance to make a PR that'd be great.

Fix patch:
diff --git a/adlfs/spec.py b/adlfs/spec.py
index ee9430bb..3352ca3c 100644
--- a/adlfs/spec.py
+++ b/adlfs/spec.py
@@ -159,7 +159,12 @@ class AzureBlobFileSystem(AsyncFileSystem):
     version_aware : bool (False)
         Whether to support blob versioning.  If enable this will require the
         user to have the necessary permissions for dealing with versioned blobs.
-
+    assume_container_exists: Optional[bool] (None)
+        Set this to true to not check for existance of containers at all, assuming they exist.
+        None (default) means to warn in case of a failure when checking for existance of a container
+        False throws if retrieving container properties fails, which might happen if your
+        authentication is only valid at the storage container level, and not the
+        storage account level.
     Pass on to fsspec:
 
     skip_instance_cache:  to control reuse of instances
@@ -221,6 +226,7 @@ def __init__(
         default_fill_cache: bool = True,
         default_cache_type: str = "bytes",
         version_aware: bool = False,
+        assume_container_exists: Optional[bool] = None,
         **kwargs,
     ):
         super_kwargs = {
@@ -245,6 +251,7 @@ def __init__(
         self.location_mode = location_mode
         self.credential = credential
         self.request_session = request_session
+        self.assume_container_exists = assume_container_exists
         if socket_timeout is not _SOCKET_TIMEOUT_DEFAULT:
             warnings.warn(
                 "socket_timeout is deprecated and has no effect.", FutureWarning
@@ -1100,6 +1107,8 @@ async def _async_walk(self, path: str, maxdepth=None, **kwargs):
                 yield path, dirs, files
 
     async def _container_exists(self, container_name):
+        if self.assume_container_exists:
+            return True
         try:
             async with self.service_client.get_container_client(
                 container_name
@@ -1108,9 +1117,15 @@ async def _container_exists(self, container_name):
         except ResourceNotFoundError:
             return False
         except Exception as e:
-            raise ValueError(
-                f"Failed to fetch container properties for {container_name} for {e}"
-            ) from e
+            if self.assume_container_exists is None:
+                warnings.warn(
+                    f"Failed to fetch container properties for {container_name}. Assume it exists already",
+                )
+                return True
+            else:
+                raise ValueError(
+                    f"Failed to fetch container properties for {container_name} for {e}"
+                ) from e
         else:
             return True
 


Test patch:
diff --git a/adlfs/tests/test_spec.py b/adlfs/tests/test_spec.py
index 87e00d65..c226e835 100644
--- a/adlfs/tests/test_spec.py
+++ b/adlfs/tests/test_spec.py
@@ -779,6 +779,7 @@ def test_mkdir(storage):
     fs = AzureBlobFileSystem(
         account_name=storage.account_name,
         connection_string=CONN_STR,
+        assume_container_exists=False,
     )
 
     # Verify mkdir will create a new container when create_parents is True
@@ -810,6 +811,22 @@ def test_mkdir(storage):
     with pytest.raises(PermissionError):
         fs.mkdir("new-container/dir", create_parents=False)
 
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name,
+        connection_string=CONN_STR,
+        assume_container_exists=None,
+    )
+
+    with pytest.warns():
+        fs.mkdir("bad_container_name")
+
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name,
+        connection_string=CONN_STR,
+        assume_container_exists=True,
+    )
+    fs.mkdir("bad_container_name")  # should not throw as we assume it exists
+
 
 def test_makedir(storage):
     fs = AzureBlobFileSystem(