Enable DefaultAzureCredential authentication for Azure filesystem block (PrefectHQ#7513)

Co-authored-by: Sam Dyson <[email protected]>
Co-authored-by: Terrence Dorsey <[email protected]>
Co-authored-by: Alexander Streed <[email protected]>
Co-authored-by: Alexander Streed <[email protected]>

Author: 4 people

docs/concepts/filesystems.md

@@ -105,6 +105,7 @@ The `Azure` file system block enables interaction with Azure Datalake and Azure
 | azure_storage_tenant_id | Azure storage tenant ID. |
 | azure_storage_client_id | Azure storage client ID. |
 | azure_storage_client_secret | Azure storage client secret. |
+| azure_storage_anon | Anonymous authentication, disable to use `DefaultAzureCredential`. |
 
 
 To create a block:

src/prefect/filesystems.py

@@ -653,6 +653,15 @@ class Azure(WritableFileSystem, WritableDeploymentStorage):
         title="Azure storage client secret",
         description="Equivalent to the AZURE_CLIENT_SECRET environment variable.",
     )
+    azure_storage_anon: bool = Field(
+        default=True,
+        title="Azure storage anonymous connection",
+        description=(
+            "Set the 'anon' flag for ADLFS. This should be False for systems that"
+            " require ADLFS to use DefaultAzureCredentials."
+        ),
+    )
+
     _remote_file_system: RemoteFileSystem = None
 
     @property
@@ -680,6 +689,7 @@ def filesystem(self) -> RemoteFileSystem:
             settings["client_secret"] = (
                 self.azure_storage_client_secret.get_secret_value()
             )
+        settings["anon"] = self.azure_storage_anon
         self._remote_file_system = RemoteFileSystem(
             basepath=f"az://{self.bucket_path}", settings=settings
         )

tests/test_filesystems.py

@@ -7,8 +7,8 @@
 
 import prefect
 from prefect.exceptions import InvalidRepositoryURLError
-from prefect.filesystems import GitHub, LocalFileSystem, RemoteFileSystem
-from prefect.testing.utilities import AsyncMock
+from prefect.filesystems import Azure, GitHub, LocalFileSystem, RemoteFileSystem
+from prefect.testing.utilities import AsyncMock, MagicMock
 from prefect.utilities.filesystem import tmpchdir
 
 TEST_PROJECTS_DIR = prefect.__root_path__ / "tests" / "test-projects"
@@ -622,3 +622,52 @@ class p:
             await g.get_directory(local_path=tmp_dst)
 
             assert any(".git" in f for f in os.listdir(tmp_dst)) == expect_git_objects
+
+
+class TestAzure:
+    def test_init(self, monkeypatch):
+        remote_storage_mock = MagicMock()
+        monkeypatch.setattr("prefect.filesystems.RemoteFileSystem", remote_storage_mock)
+        Azure(
+            azure_storage_tenant_id="tenant",
+            azure_storage_account_name="account",
+            azure_storage_client_id="client_id",
+            azure_storage_account_key="key",
+            azure_storage_client_secret="secret",
+            bucket_path="bucket",
+        ).filesystem
+        remote_storage_mock.assert_called_once_with(
+            basepath="az://bucket",
+            settings={
+                "account_name": "account",
+                "account_key": "key",
+                "tenant_id": "tenant",
+                "client_id": "client_id",
+                "client_secret": "secret",
+                "anon": True,
+            },
+        )
+
+    def test_init_with_anon(self, monkeypatch):
+        remote_storage_mock = MagicMock()
+        monkeypatch.setattr("prefect.filesystems.RemoteFileSystem", remote_storage_mock)
+        Azure(
+            azure_storage_tenant_id="tenant",
+            azure_storage_account_name="account",
+            azure_storage_client_id="client_id",
+            azure_storage_account_key="key",
+            azure_storage_client_secret="secret",
+            bucket_path="bucket",
+            azure_storage_anon=False,
+        ).filesystem
+        remote_storage_mock.assert_called_once_with(
+            basepath="az://bucket",
+            settings={
+                "account_name": "account",
+                "account_key": "key",
+                "tenant_id": "tenant",
+                "client_id": "client_id",
+                "client_secret": "secret",
+                "anon": False,
+            },
+        )