Add filename to generate s3 download url

clean_python/s3/sync_s3_gateway.py

@@ -137,18 +137,17 @@ def remove_multiple(self, ids: list[Id]) -> None:
         )
 
     def _create_presigned_url(
-        self,
-        id: Id,
-        client_method: str,
+        self, id: Id, client_method: str, filename: str | None = None
     ) -> AnyHttpUrl:
+        params = {"Bucket": self.provider.bucket, "Key": self._id_to_key(id)}
+        if filename:
+            params["ResponseContentDisposition"] = f"attachment; filename={filename}"
         return self.provider.client.generate_presigned_url(
-            client_method,
-            Params={"Bucket": self.provider.bucket, "Key": self._id_to_key(id)},
-            ExpiresIn=DEFAULT_EXPIRY,
+            client_method, Params=params, ExpiresIn=DEFAULT_EXPIRY
         )
 
-    def create_download_url(self, id: Id) -> AnyHttpUrl:
-        return self._create_presigned_url(id, "get_object")
+    def create_download_url(self, id: Id, filename: str | None = None) -> AnyHttpUrl:
+        return self._create_presigned_url(id, "get_object", filename)
 
     def create_upload_url(self, id: Id) -> AnyHttpUrl:
         return self._create_presigned_url(id, "put_object")

integration_tests/test_sync_s3_gateway.py

@@ -117,6 +117,33 @@ def test_remove_multiple_empty_list(s3_gateway: SyncS3Gateway, s3_bucket):
     s3_gateway.remove_multiple([])
 
 
+def test_create_download_url(s3_gateway: SyncS3Gateway, object_in_s3):
+    actual = s3_gateway.create_download_url(object_in_s3)
+
+    assert object_in_s3 in actual
+    assert "response-content-disposition=attachment%3B%20filename%3D" not in actual
+    assert "X-Amz-Expires=3600" in actual
+    assert "X-Amz-SignedHeaders=host" in actual
+
+
+def test_create_download_url_with_filename(s3_gateway: SyncS3Gateway, object_in_s3):
+    actual = s3_gateway.create_download_url(object_in_s3, "file.txt")
+
+    assert "file.txt" in actual
+    assert "response-content-disposition=attachment%3B%20filename%3Dfile.txt" in actual
+    assert object_in_s3 in actual
+    assert "X-Amz-Expires=3600" in actual
+    assert "X-Amz-SignedHeaders=host" in actual
+
+
+def test_create_upload_url(s3_gateway: SyncS3Gateway, object_in_s3):
+    actual = s3_gateway.create_upload_url(object_in_s3)
+
+    assert object_in_s3 in actual
+    assert "X-Amz-Expires=3600" in actual
+    assert "X-Amz-SignedHeaders=host" in actual
+
+
 def test_remove_filtered_all(s3_gateway: SyncS3Gateway, multiple_objects):
     s3_gateway.remove_filtered([])
 

tests/s3/test_sync_s3_gateway.py

@@ -0,0 +1,60 @@
+from unittest.mock import Mock
+import pytest
+
+from clean_python.s3 import SyncS3BucketProvider, SyncS3Gateway
+from clean_python.s3.sync_s3_gateway import DEFAULT_EXPIRY
+
+
+@pytest.fixture
+def provider() -> SyncS3BucketProvider:
+    return Mock(SyncS3BucketProvider, bucket="S3Bucket")
+
+
+@pytest.fixture
+def gateway(provider: Mock) -> SyncS3Gateway:
+    return SyncS3Gateway(provider)
+
+
+def test_create_download_url(gateway: SyncS3Gateway, provider: Mock):
+    provider.client.generate_presigned_url.return_value = (
+        "https://s3.amazonaws.com/object_in_s3"
+    )
+
+    actual = gateway.create_download_url("object-in-s3")
+
+    assert actual == provider.client.generate_presigned_url.return_value
+    provider.client.generate_presigned_url.assert_called_once_with(
+        "get_object",
+        Params={"Bucket": "S3Bucket", "Key": "object-in-s3"},
+        ExpiresIn=DEFAULT_EXPIRY,
+    )
+
+
+def test_create_download_url_with_filename(gateway: SyncS3Gateway, provider: Mock):
+    provider.client.generate_presigned_url.return_value = "https://s3.com/S3Object"
+
+    actual = gateway.create_download_url("S3Object", "file.txt")
+
+    assert actual == provider.client.generate_presigned_url.return_value
+    provider.client.generate_presigned_url.assert_called_once_with(
+        "get_object",
+        Params={
+            "Bucket": "S3Bucket",
+            "Key": "S3Object",
+            "ResponseContentDisposition": "attachment; filename=file.txt",
+        },
+        ExpiresIn=DEFAULT_EXPIRY,
+    )
+
+
+def test_create_upload_url(gateway: SyncS3Gateway, provider: Mock):
+    provider.client.generate_presigned_url.return_value = "https://s3.com/S3Object"
+
+    actual = gateway.create_upload_url("S3Object")
+
+    assert actual == provider.client.generate_presigned_url.return_value
+    provider.client.generate_presigned_url.assert_called_once_with(
+        "put_object",
+        Params={"Bucket": "S3Bucket", "Key": "S3Object"},
+        ExpiresIn=DEFAULT_EXPIRY,
+    )