Forward content_settings on write (open, pipe_file, put_file)

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 Unreleased
 ----------
-- .
+- Added a `content_settings` parameter to `open()` (write modes), `pipe_file()` and `put_file()` to set blob content settings (content type, content disposition, cache control, ...) on write. Accepts a `dict` (recommended) or an `azure.storage.blob.ContentSettings` instance. [#554](https://github.com/fsspec/adlfs/pull/554)
 
 2026.5.0
 --------

adlfs/spec.py

@@ -31,6 +31,7 @@
     BlobProperties,
     BlobSasPermissions,
     BlobType,
+    ContentSettings,
     generate_blob_sas,
 )
 from azure.storage.blob.aio import BlobPrefix
@@ -163,6 +164,19 @@ def _normalize_etag_quotes(etag: str) -> Optional[str]:
     return f'"{etag.strip(double_quote)}"'
 
 
+def _as_content_settings(
+    content_settings: Optional[Union[ContentSettings, dict]],
+) -> Optional[ContentSettings]:
+    """Coerce ``content_settings`` given as a dict into a ``ContentSettings``.
+
+    Accepts either a plain ``dict`` (recommended — keeps the Azure SDK out of the
+    caller's imports) or a ``ContentSettings`` instance.
+    """
+    if content_settings is None or isinstance(content_settings, ContentSettings):
+        return content_settings
+    return ContentSettings(**content_settings)
+
+
 class AzureBlobFileSystem(AsyncFileSystem):
     """
     Access Azure Datalake Gen2 and Azure Storage if it were a file system using Multiprotocol Access
@@ -1505,6 +1519,7 @@ async def _pipe_file(
         """Set the bytes of given file"""
         if kwargs.pop("mode", "") == "create":
             overwrite = False
+        content_settings = _as_content_settings(kwargs.pop("content_settings", None))
         container_name, path, _ = self.split_path(path)
         async with self.service_client.get_blob_client(
             container=container_name, blob=path
@@ -1514,6 +1529,7 @@ async def _pipe_file(
                     data=value,
                     overwrite=overwrite,
                     metadata={"is_directory": "false"},
+                    content_settings=content_settings,
                     max_concurrency=max_concurrency or self.max_concurrency,
                     **self._timeout_kwargs,
                     **kwargs,
@@ -1744,6 +1760,7 @@ async def _put_file(
 
         if kwargs.pop("mode", "") == "create":
             overwrite = False
+        content_settings = _as_content_settings(kwargs.pop("content_settings", None))
         container_name, path, _ = self.split_path(rpath, delimiter=delimiter)
 
         if os.path.isdir(lpath):
@@ -1758,6 +1775,7 @@ async def _put_file(
                             f1,
                             overwrite=overwrite,
                             metadata={"is_directory": "false"},
+                            content_settings=content_settings,
                             raw_response_hook=make_callback(
                                 "upload_stream_current", callback
                             ),
@@ -1899,6 +1917,7 @@ def _open(
         cache_type="readahead",
         metadata=None,
         version_id: Optional[str] = None,
+        content_settings: Optional[Union[ContentSettings, dict]] = None,
         **kwargs,
     ):
         """Open a file on the datalake, or a block blob
@@ -1927,6 +1946,11 @@ def _open(
         version_id: str
             Explicit version of the blob to open.  This requires that the abfs filesystem
             is versioning aware and blob versioning is enabled on the releveant container.
+
+        content_settings: dict
+            Optional content settings applied to the blob when writing, e.g.
+            ``{"content_type": "application/pdf", "content_disposition": "..."}``.
+            A ``ContentSettings`` instance is also accepted.
         """
         logger.debug(f"_open:  {path}")
         if block_size is None:
@@ -1946,6 +1970,7 @@ def _open(
             cache_type=cache_type,
             metadata=metadata,
             version_id=version_id,
+            content_settings=content_settings,
             **kwargs,
         )
 
@@ -1966,6 +1991,7 @@ def __init__(
         cache_options: dict = {},
         metadata=None,
         version_id: Optional[str] = None,
+        content_settings: Optional[Union[ContentSettings, dict]] = None,
         **kwargs,
     ):
         """
@@ -2003,6 +2029,11 @@ def __init__(
             attribute is populated with the version created by the upload when the
             filesystem has ``version_aware=True``.
 
+        content_settings: dict
+            Optional content settings applied to the blob when writing, e.g.
+            ``{"content_type": "application/pdf", "content_disposition": "..."}``.
+            A ``ContentSettings`` instance is also accepted.
+
         kwargs: dict
             Passed to AbstractBufferedFile
         """
@@ -2073,6 +2104,7 @@ def __init__(
 
         else:
             self._metadata = metadata or {"is_directory": "false"}
+            self._content_settings = _as_content_settings(content_settings)
             self.buffer = io.BytesIO()
             self.offset = None
             self.forced = False
@@ -2209,7 +2241,10 @@ async def _async_initiate_upload(self, **kwargs):
         if self.mode == "ab":
             if not await self.fs._exists(self.path):
                 async with self.container_client.get_blob_client(blob=self.blob) as bc:
-                    await bc.create_append_blob(metadata=self.metadata)
+                    await bc.create_append_blob(
+                        metadata=self.metadata,
+                        content_settings=self._content_settings,
+                    )
 
     _initiate_upload = sync_wrapper(_async_initiate_upload)
 
@@ -2269,7 +2304,10 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs):
                         blob=self.blob
                     ) as bc:
                         response = await bc.commit_block_list(
-                            block_list=block_list, metadata=self.metadata, **commit_kw
+                            block_list=block_list,
+                            metadata=self.metadata,
+                            content_settings=self._content_settings,
+                            **commit_kw,
                         )
                         if self.fs.version_aware:
                             self.version_id = response.get("version_id")
@@ -2287,6 +2325,7 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs):
                         response = await bc.upload_blob(
                             data=data,
                             metadata=self.metadata,
+                            content_settings=self._content_settings,
                             overwrite=(self.mode == "wb"),
                         )
                         if self.fs.version_aware:
@@ -2301,6 +2340,7 @@ async def _async_upload_chunk(self, final: bool = False, **kwargs):
                             response = await bc.commit_block_list(
                                 block_list=block_list,
                                 metadata=self.metadata,
+                                content_settings=self._content_settings,
                                 **commit_kw,
                             )
                             if self.fs.version_aware:

adlfs/tests/test_spec.py

@@ -1413,6 +1413,101 @@ def test_metadata_write(storage):
     fs.rmdir("test-metadata-write")
 
 
+CONTENT_SETTINGS = {
+    "content_type": "application/pdf",
+    "content_disposition": 'attachment; filename="f.pdf"',
+    "cache_control": "max-age=3600",
+}
+
+
+def test_content_settings_open(storage):
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name, connection_string=CONN_STR
+    )
+    fs.mkdir("test-cs-open")
+    with fs.open("test-cs-open/file.pdf", "wb", content_settings=CONTENT_SETTINGS) as f:
+        f.write(b"0123456789")
+    cs = fs.info("test-cs-open/file.pdf")["content_settings"]
+    assert cs["content_type"] == "application/pdf"
+    assert cs["content_disposition"] == 'attachment; filename="f.pdf"'
+    assert cs["cache_control"] == "max-age=3600"
+
+    # empty streaming write still applies content settings
+    with fs.open(
+        "test-cs-open/empty.pdf", "wb", content_settings=CONTENT_SETTINGS
+    ) as f:
+        f.write(b"")
+    assert (
+        fs.info("test-cs-open/empty.pdf")["content_settings"]["content_type"]
+        == "application/pdf"
+    )
+    fs.rmdir("test-cs-open")
+
+
+def test_content_settings_append(storage):
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name, connection_string=CONN_STR
+    )
+    fs.mkdir("test-cs-append")
+    with fs.open(
+        "test-cs-append/file.pdf", "ab", content_settings=CONTENT_SETTINGS
+    ) as f:
+        f.write(b"0123456789")
+    cs = fs.info("test-cs-append/file.pdf")["content_settings"]
+    assert cs["content_type"] == "application/pdf"
+    assert cs["content_disposition"] == 'attachment; filename="f.pdf"'
+    fs.rmdir("test-cs-append")
+
+
+def test_content_settings_pipe_file(storage):
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name, connection_string=CONN_STR
+    )
+    fs.mkdir("test-cs-pipe")
+    fs.pipe_file(
+        "test-cs-pipe/piped.pdf", b"0123456789", content_settings=CONTENT_SETTINGS
+    )
+    info = fs.info("test-cs-pipe/piped.pdf")
+    assert info["content_settings"]["content_disposition"] == (
+        'attachment; filename="f.pdf"'
+    )
+    assert fs.cat_file("test-cs-pipe/piped.pdf") == b"0123456789"
+    fs.rmdir("test-cs-pipe")
+
+
+def test_content_settings_put_file(storage, tmp_path):
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name, connection_string=CONN_STR
+    )
+    fs.mkdir("test-cs-put")
+    local = tmp_path / "f.pdf"
+    local.write_bytes(b"0123456789")
+    fs.put_file(str(local), "test-cs-put/put.pdf", content_settings=CONTENT_SETTINGS)
+    info = fs.info("test-cs-put/put.pdf")
+    assert info["content_settings"]["content_type"] == "application/pdf"
+    assert fs.cat_file("test-cs-put/put.pdf") == b"0123456789"
+    fs.rmdir("test-cs-put")
+
+
+def test_content_settings_accepts_object(storage):
+    # A ContentSettings instance is accepted for compatibility (dict is the
+    # documented form).
+    from azure.storage.blob import ContentSettings
+
+    fs = AzureBlobFileSystem(
+        account_name=storage.account_name, connection_string=CONN_STR
+    )
+    fs.mkdir("test-cs-object")
+    settings = ContentSettings(content_type="application/pdf")
+    with fs.open("test-cs-object/obj.pdf", "wb", content_settings=settings) as f:
+        f.write(b"x")
+    assert (
+        fs.info("test-cs-object/obj.pdf")["content_settings"]["content_type"]
+        == "application/pdf"
+    )
+    fs.rmdir("test-cs-object")
+
+
 def test_put_file(storage, tmp_path):
     fs = AzureBlobFileSystem(
         account_name=storage.account_name, connection_string=CONN_STR
@@ -2012,6 +2107,7 @@ async def test_pipe_file_timeout(storage, mocker):
     upload_blob.assert_called_once_with(
         data=b"data",
         metadata={"is_directory": "false"},
+        content_settings=None,
         overwrite=True,
         max_concurrency=fs.max_concurrency,
         timeout=11,
@@ -2040,6 +2136,7 @@ async def test_put_file_timeout(storage, mocker, tmp_path):
     upload_blob.assert_called_once_with(
         mocker.ANY,
         metadata={"is_directory": "false"},
+        content_settings=None,
         overwrite=True,
         raw_response_hook=None,
         max_concurrency=fs.max_concurrency,