Repository: RaRe-Technologies/smart_open (python)

Issue:
Allow specifying metadata for azure uploads
#### Problem description

I would like to upload an Azure blob with metadata. This is currently possible with s3 and gcs:

```py
metadata = {"name": "bla"}
bucket = "test"
key = "test"
open(
    f"s3://{bucket}/{key}",
    "wb",
    transport_params={
        "client_kwargs": {"S3.Client.create_multipart_upload": {"Metadata": metadata},
    },
)
open(
    f"gs://{bucket}/{key}",
    "wb",
    transport_params={
        "blob_properties": {"metadata": metadata},
    },
)
```

For Azure, `smart_open` only has the possibility of passing a custom [`BlobClient`](https://docs.microsoft.com/en-us/python/api/azure-storage-blob/azure.storage.blob.blobclient?view=azure-python) but it looks like metadata needs to be [explicitly passed](https://github.com/Azure/azure-sdk-for-python/blob/azure-storage-blob_12.12.0/sdk/storage/azure-storage-blob/azure/storage/blob/_blob_client.py#L2631) to `BlobClient.commit_block_list`. 

I suggest introducing `blob_kwargs`:
```py
open(
    f"azure://{container}/{path}",
    "wb",
    transport_params={
        "blob_kwargs": {"metadata": metadata},
    },
)
```

#### Steps/code to reproduce the problem

N/A

#### Versions

Please provide the output of:

```pycon
>>> import platform, sys, smart_open
>>> print(platform.platform())
>>> print("Python", sys.version)
>>> print("smart_open", smart_open.__version__)
macOS-10.15.7-x86_64-i386-64bit
Python 3.8.10 (default, Sep  1 2021, 10:14:33)
[Clang 12.0.0 (clang-1200.0.32.29)]
smart_open 6.0.0
```

#### Checklist

Before you create the issue, please make sure you have:

- [x] Described the problem clearly
- [x] Provided a minimal reproducible example, including any required data
- [x] Provided the version numbers of the relevant software


Pull request description:
Allow passing additional kwargs for Azure writes
#### Title

Allow passing additional kwargs for Azure writes

#### Motivation

- Fixes #701 

#### Tests

Tests that `smart_open.azure.Writer` accepts metadata, and retrieving it afterwards (with `FakeBlobClient.get_blob_properties`).

#### Checklist

Before you create the PR, please make sure you have:

- [x] Picked a concise, informative and complete title
- [x] Clearly explained the motivation behind the PR
- [x] Linked to any existing issues that your PR will be solving
- [x] Included tests for any new functionality
- [x] Checked that all unit tests pass

#### Workflow

Please avoid rebasing and force-pushing to the branch of the PR once a review is in progress.
Rebasing can make your commits look a bit cleaner, but it also makes life more difficult from the reviewer, because they are no longer able to distinguish between code that has already been reviewed, and unreviewed code.


Patch:
diff --git a/smart_open/azure.py b/smart_open/azure.py
index b3a9bad..15168a9 100644
--- a/smart_open/azure.py
+++ b/smart_open/azure.py
@@ -70,6 +70,7 @@ def open(
         blob_id,
         mode,
         client=None,  # type: Union[azure.storage.blob.BlobServiceClient, azure.storage.blob.ContainerClient, azure.storage.blob.BlobClient] # noqa
+        blob_kwargs=None,
         buffer_size=DEFAULT_BUFFER_SIZE,
         min_part_size=_DEFAULT_MIN_PART_SIZE,
         max_concurrency=DEFAULT_MAX_CONCURRENCY,
@@ -86,10 +87,13 @@ def open(
         The mode for opening the object.  Must be either "rb" or "wb".
     client: azure.storage.blob.BlobServiceClient, ContainerClient, or BlobClient
         The Azure Blob Storage client to use when working with azure-storage-blob.
+    blob_kwargs: dict, optional
+        Additional parameters to pass to `BlobClient.commit_block_list`.
+        For writing only.
     buffer_size: int, optional
         The buffer size to use when performing I/O. For reading only.
     min_part_size: int, optional
-        The minimum part size for multipart uploads.  For writing only.
+        The minimum part size for multipart uploads. For writing only.
     max_concurrency: int, optional
         The number of parallel connections with which to download. For reading only.
 
@@ -111,6 +115,7 @@ def open(
             container_id,
             blob_id,
             client,
+            blob_kwargs=blob_kwargs,
             min_part_size=min_part_size
         )
     else:
@@ -387,12 +392,14 @@ class Writer(io.BufferedIOBase):
             container,
             blob,
             client,  # type: Union[azure.storage.blob.BlobServiceClient, azure.storage.blob.ContainerClient, azure.storage.blob.BlobClient]  # noqa
+            blob_kwargs=None,
             min_part_size=_DEFAULT_MIN_PART_SIZE,
     ):
         self._is_closed = False
         self._container_name = container
 
         self._blob = _get_blob_client(client, container, blob)
+        self._blob_kwargs = blob_kwargs or {}
         # type: azure.storage.blob.BlobClient
 
         self._min_part_size = min_part_size
@@ -419,7 +426,7 @@ class Writer(io.BufferedIOBase):
         if not self.closed:
             if self._current_part.tell() > 0:
                 self._upload_part()
-            self._blob.commit_block_list(self._block_list)
+            self._blob.commit_block_list(self._block_list, **self._blob_kwargs)
             self._block_list = []
             self._is_closed = True
         logger.debug("successfully closed")


Test patch:
diff --git a/smart_open/tests/test_azure.py b/smart_open/tests/test_azure.py
index d249748..3c668aa 100644
--- a/smart_open/tests/test_azure.py
+++ b/smart_open/tests/test_azure.py
@@ -51,10 +51,12 @@ class FakeBlobClient(object):
         self.__contents = io.BytesIO()
         self._staged_contents = {}
 
-    def commit_block_list(self, block_list):
+    def commit_block_list(self, block_list, metadata=None):
         data = b''.join([self._staged_contents[block_blob['id']] for block_blob in block_list])
         self.__contents = io.BytesIO(data)
-        self.set_blob_metadata(dict(size=len(data)))
+        metadata = metadata or {}
+        metadata.update({"size": len(data)})
+        self.set_blob_metadata(metadata)
         self._container_client.register_blob_client(self)
 
     def delete_blob(self):
@@ -590,6 +592,32 @@ class WriterTest(unittest.TestCase):
         ))
         assert output == [test_string]
 
+    def test_write_blob_client(self):
+        """Does writing into Azure Blob Storage work correctly?"""
+        test_string = u"žluťoučký koníček".encode('utf8')
+        blob_name = "test_write_blob_client_%s" % BLOB_NAME
+
+        container_client = CLIENT.get_container_client(CONTAINER_NAME)
+        blob_client = container_client.get_blob_client(blob_name)
+
+        with smart_open.open(
+            "azure://%s/%s" % (CONTAINER_NAME, blob_name),
+            "wb",
+            transport_params={
+                "client": blob_client, "blob_kwargs": {"metadata": {"name": blob_name}}
+            },
+        ) as fout:
+            fout.write(test_string)
+
+        self.assertEqual(blob_client.get_blob_properties()["name"], blob_name)
+
+        output = list(smart_open.open(
+            "azure://%s/%s" % (CONTAINER_NAME, blob_name),
+            "rb",
+            transport_params=dict(client=CLIENT),
+        ))
+        self.assertEqual(output, [test_string])
+
     def test_incorrect_input(self):
         """Does azure write fail on incorrect input?"""
         blob_name = "test_incorrect_input_%s" % BLOB_NAME