[azure] Allow creation of signed URLs for upload

CHANGELOG.rst

@@ -15,11 +15,13 @@ Azure
 -----
 
 - Fix ``collectstatic --clear`` (`#1403`_)
+- Add ``mode`` kwarg to ``.url()`` to support creation of signed URLs for upload (`#1414`_)
 
 .. _#1399: https://github.com/jschneier/django-storages/pull/1399
 .. _#1381: https://github.com/jschneier/django-storages/pull/1381
 .. _#1402: https://github.com/jschneier/django-storages/pull/1402
 .. _#1403: https://github.com/jschneier/django-storages/pull/1403
+.. _#1414: https://github.com/jschneier/django-storages/pull/1414
 
 
 1.14.3 (2024-05-04)

storages/backends/azure_storage.py

@@ -309,9 +309,10 @@ def _expire_at(self, expire):
         # azure expects time in UTC
         return datetime.utcnow() + timedelta(seconds=expire)
 
-    def url(self, name, expire=None, parameters=None):
+    def url(self, name, expire=None, parameters=None, mode="r"):
         name = self._get_valid_path(name)
         params = parameters or {}
+        permission = BlobSasPermissions.from_string(mode)
 
         if expire is None:
             expire = self.expiration_secs
@@ -326,7 +327,7 @@ def url(self, name, expire=None, parameters=None):
                 name,
                 account_key=self.account_key,
                 user_delegation_key=user_delegation_key,
-                permission=BlobSasPermissions(read=True),
+                permission=permission,
                 expiry=expiry,
                 **params,
             )

tests/test_azure.py

@@ -164,7 +164,9 @@ def test_url_expire(self, generate_blob_sas_mocked):
         fixed_time = make_aware(
             datetime.datetime(2016, 11, 6, 4), datetime.timezone.utc
         )
+
         with mock.patch("storages.backends.azure_storage.datetime") as d_mocked:
+            # Implicit read permission
             d_mocked.utcnow.return_value = fixed_time
             self.assertEqual(
                 self.storage.url("some blob", 100),
@@ -179,6 +181,26 @@ def test_url_expire(self, generate_blob_sas_mocked):
                 permission=mock.ANY,
                 expiry=fixed_time + timedelta(seconds=100),
             )
+            called_args, called_kwargs = generate_blob_sas_mocked.call_args
+            self.assertEqual(str(called_kwargs["permission"]), "r")
+
+            # Explicit write permission
+            d_mocked.utcnow.return_value = fixed_time
+            self.assertEqual(
+                self.storage.url("some blob", expire=100, mode="w"),
+                "https://ret_foo.blob.core.windows.net/test/some%20blob",
+            )
+            generate_blob_sas_mocked.assert_called_with(
+                self.account_name,
+                self.container_name,
+                "some blob",
+                account_key=self.account_key,
+                user_delegation_key=None,
+                permission=mock.ANY,
+                expiry=fixed_time + timedelta(seconds=100),
+            )
+            called_args, called_kwargs = generate_blob_sas_mocked.call_args
+            self.assertEqual(str(called_kwargs["permission"]), "w")
 
     @mock.patch("storages.backends.azure_storage.generate_blob_sas")
     def test_url_expire_user_delegation_key(self, generate_blob_sas_mocked):