feat: created functions to get multiple signed URLs. (#105)

* feat: created functions to get multiple signed URLs.

* feat: Fixed optional params. Handling auth token issue #73 in separate PR.

* feat: remove sync code as it will be generated by unasync.

* chore: generate sync client

---------

Co-authored-by: Alexander Leonov <[email protected]>
Co-authored-by: anand2312 <[email protected]>

Author: dishwad

storage3/_async/file_api.py

@@ -12,6 +12,7 @@
 from ..types import (
     BaseBucket,
     CreateSignedURLOptions,
+    CreateSignedURLsOptions,
     FileOptions,
     ListBucketFilesOptions,
     RequestMethod,
@@ -62,19 +63,56 @@ async def create_signed_url(
             file path to be downloaded, including the current file name.
         expires_in
             number of seconds until the signed URL expires.
+        options
+            options to be passed for downloading or transforming the file.
         """
+        json = {"expiresIn": str(expires_in)}
+        if options.get("download"):
+            json.update({"download": options["download"]})
+        if options.get("transform"):
+            json.update({"transform": options["transform"]})
+
         path = self._get_final_path(path)
         response = await self._request(
             "POST",
             f"/object/sign/{path}",
-            json={"expiresIn": str(expires_in)},
+            json=json,
         )
         data = response.json()
         data[
             "signedURL"
         ] = f"{self._client.base_url}{cast(str, data['signedURL']).lstrip('/')}"
         return data
 
+    async def create_signed_urls(
+        self, paths: list[str], expires_in: int, options: CreateSignedURLsOptions = {}
+    ) -> list[dict[str, str]]:
+        """
+        Parameters
+        ----------
+        path
+            file path to be downloaded, including the current file name.
+        expires_in
+            number of seconds until the signed URL expires.
+        options
+            options to be passed for downloading the file.
+        """
+        json = {"paths": paths, "expiresIn": str(expires_in)}
+        if options.get("download"):
+            json.update({"download": options["download"]})
+
+        response = await self._request(
+            "POST",
+            f"/object/sign/{self.id}",
+            json=json,
+        )
+        data = response.json()
+        for item in data:
+            item[
+                "signedURL"
+            ] = f"{self._client.base_url}{cast(str, item['signedURL']).lstrip('/')}"
+        return data
+
     async def get_public_url(self, path: str, options: TransformOptions = {}) -> str:
         """
         Parameters

storage3/_sync/file_api.py

@@ -12,6 +12,7 @@
 from ..types import (
     BaseBucket,
     CreateSignedURLOptions,
+    CreateSignedURLsOptions,
     FileOptions,
     ListBucketFilesOptions,
     RequestMethod,
@@ -62,19 +63,56 @@ def create_signed_url(
             file path to be downloaded, including the current file name.
         expires_in
             number of seconds until the signed URL expires.
+        options
+            options to be passed for downloading or transforming the file.
         """
+        json = {"expiresIn": str(expires_in)}
+        if options.get("download"):
+            json.update({"download": options["download"]})
+        if options.get("transform"):
+            json.update({"transform": options["transform"]})
+
         path = self._get_final_path(path)
         response = self._request(
             "POST",
             f"/object/sign/{path}",
-            json={"expiresIn": str(expires_in)},
+            json=json,
         )
         data = response.json()
         data[
             "signedURL"
         ] = f"{self._client.base_url}{cast(str, data['signedURL']).lstrip('/')}"
         return data
 
+    def create_signed_urls(
+        self, paths: list[str], expires_in: int, options: CreateSignedURLsOptions = {}
+    ) -> list[dict[str, str]]:
+        """
+        Parameters
+        ----------
+        path
+            file path to be downloaded, including the current file name.
+        expires_in
+            number of seconds until the signed URL expires.
+        options
+            options to be passed for downloading the file.
+        """
+        json = {"paths": paths, "expiresIn": str(expires_in)}
+        if options.get("download"):
+            json.update({"download": options["download"]})
+
+        response = self._request(
+            "POST",
+            f"/object/sign/{self.id}",
+            json=json,
+        )
+        data = response.json()
+        for item in data:
+            item[
+                "signedURL"
+            ] = f"{self._client.base_url}{cast(str, item['signedURL']).lstrip('/')}"
+        return data
+
     def get_public_url(self, path: str, options: TransformOptions = {}) -> str:
         """
         Parameters

storage3/types.py

@@ -48,15 +48,21 @@ class ListBucketFilesOptions(TypedDict):
     sortBy: _sortByType
 
 
-class TransformOptions(TypedDict):
-    height: Optional[float]
-    width: Optional[float]
-    resize: Optional[Union[Literal["cover"], Literal["contain"], Literal["fill"]]]
+class TransformOptions(TypedDict, total=False):
+    height: int
+    width: int
+    resize: Literal["cover", "contain", "fill"]
+    format: Literal["origin", "avif"]
+    quality: int
 
 
-class CreateSignedURLOptions(TypedDict):
-    download: Optional[Union[str, bool]]
-    transform: Optional[TransformOptions]
+class CreateSignedURLOptions(TypedDict, total=False):
+    download: Union[str, bool]
+    transform: TransformOptions
+
+
+class CreateSignedURLsOptions(TypedDict):
+    download: Union[str, bool]
 
 
 FileOptions = TypedDict(

tests/_async/test_client.py

@@ -156,6 +156,54 @@ def file(tmp_path: Path, uuid_factory: Callable[[], str]) -> FileForTesting:
     )
 
 
+@pytest.fixture
+def multi_file(tmp_path: Path, uuid_factory: Callable[[], str]) -> list[FileForTesting]:
+    """Creates multiple test files (same content, same bucket/folder path, different file names)"""
+    file_name_1 = "test_image_1.svg"
+    file_name_2 = "test_image_2.svg"
+    file_content = (
+        b'<svg width="109" height="113" viewBox="0 0 109 113" fill="none" xmlns="http://www.w3.org/2000/svg"> '
+        b'<path d="M63.7076 110.284C60.8481 113.885 55.0502 111.912 54.9813 107.314L53.9738 40.0627L99.1935 '
+        b'40.0627C107.384 40.0627 111.952 49.5228 106.859 55.9374L63.7076 110.284Z" fill="url(#paint0_linear)"/> '
+        b'<path d="M63.7076 110.284C60.8481 113.885 55.0502 111.912 54.9813 107.314L53.9738 40.0627L99.1935 '
+        b'40.0627C107.384 40.0627 111.952 49.5228 106.859 55.9374L63.7076 110.284Z" fill="url(#paint1_linear)" '
+        b'fill-opacity="0.2"/> <path d="M45.317 2.07103C48.1765 -1.53037 53.9745 0.442937 54.0434 5.041L54.4849 '
+        b'72.2922H9.83113C1.64038 72.2922 -2.92775 62.8321 2.1655 56.4175L45.317 2.07103Z" fill="#3ECF8E"/> <defs>'
+        b'<linearGradient id="paint0_linear" x1="53.9738" y1="54.974" x2="94.1635" y2="71.8295"'
+        b'gradientUnits="userSpaceOnUse"> <stop stop-color="#249361"/> <stop offset="1" stop-color="#3ECF8E"/> '
+        b'</linearGradient> <linearGradient id="paint1_linear" x1="36.1558" y1="30.578" x2="54.4844" y2="65.0806" '
+        b'gradientUnits="userSpaceOnUse"> <stop/> <stop offset="1" stop-opacity="0"/> </linearGradient> </defs> </svg>'
+    )
+    bucket_folder = uuid_factory()
+    bucket_path_1 = f"{bucket_folder}/{file_name_1}"
+    bucket_path_2 = f"{bucket_folder}/{file_name_2}"
+    file_path_1 = tmp_path / file_name_1
+    file_path_2 = tmp_path / file_name_2
+    with open(file_path_1, "wb") as f:
+        f.write(file_content)
+    with open(file_path_2, "wb") as f:
+        f.write(file_content)
+
+    return [
+        FileForTesting(
+            name=file_name_1,
+            local_path=str(file_path_1),
+            bucket_folder=bucket_folder,
+            bucket_path=bucket_path_1,
+            mime_type="image/svg+xml",
+            file_content=file_content,
+        ),
+        FileForTesting(
+            name=file_name_2,
+            local_path=str(file_path_2),
+            bucket_folder=bucket_folder,
+            bucket_path=bucket_path_2,
+            mime_type="image/svg+xml",
+            file_content=file_content,
+        ),
+    ]
+
+
 # TODO: Test create_bucket, delete_bucket, empty_bucket, list_buckets, fileAPI.list before upload test
 
 
@@ -194,6 +242,26 @@ async def test_client_create_signed_url(
     assert response.content == file.file_content
 
 
+async def test_client_create_signed_urls(
+    storage_file_client: AsyncBucketProxy, multi_file: list[FileForTesting]
+) -> None:
+    """Ensure we can create signed urls for files in a bucket"""
+    paths = []
+    for file in multi_file:
+        paths.append(file.bucket_path)
+        await storage_file_client.upload(
+            file.bucket_path, file.local_path, {"content-type": file.mime_type}
+        )
+
+    signed_urls = await storage_file_client.create_signed_urls(paths, 10)
+
+    async with HttpxClient() as client:
+        for url in signed_urls:
+            response = await client.get(url["signedURL"])
+            response.raise_for_status()
+            assert response.content == multi_file[0].file_content
+
+
 async def test_client_get_public_url(
     storage_file_client_public: AsyncBucketProxy, file: FileForTesting
 ) -> None:

tests/_sync/test_client.py

@@ -154,6 +154,54 @@ def file(tmp_path: Path, uuid_factory: Callable[[], str]) -> FileForTesting:
     )
 
 
+@pytest.fixture
+def multi_file(tmp_path: Path, uuid_factory: Callable[[], str]) -> list[FileForTesting]:
+    """Creates multiple test files (same content, same bucket/folder path, different file names)"""
+    file_name_1 = "test_image_1.svg"
+    file_name_2 = "test_image_2.svg"
+    file_content = (
+        b'<svg width="109" height="113" viewBox="0 0 109 113" fill="none" xmlns="http://www.w3.org/2000/svg"> '
+        b'<path d="M63.7076 110.284C60.8481 113.885 55.0502 111.912 54.9813 107.314L53.9738 40.0627L99.1935 '
+        b'40.0627C107.384 40.0627 111.952 49.5228 106.859 55.9374L63.7076 110.284Z" fill="url(#paint0_linear)"/> '
+        b'<path d="M63.7076 110.284C60.8481 113.885 55.0502 111.912 54.9813 107.314L53.9738 40.0627L99.1935 '
+        b'40.0627C107.384 40.0627 111.952 49.5228 106.859 55.9374L63.7076 110.284Z" fill="url(#paint1_linear)" '
+        b'fill-opacity="0.2"/> <path d="M45.317 2.07103C48.1765 -1.53037 53.9745 0.442937 54.0434 5.041L54.4849 '
+        b'72.2922H9.83113C1.64038 72.2922 -2.92775 62.8321 2.1655 56.4175L45.317 2.07103Z" fill="#3ECF8E"/> <defs>'
+        b'<linearGradient id="paint0_linear" x1="53.9738" y1="54.974" x2="94.1635" y2="71.8295"'
+        b'gradientUnits="userSpaceOnUse"> <stop stop-color="#249361"/> <stop offset="1" stop-color="#3ECF8E"/> '
+        b'</linearGradient> <linearGradient id="paint1_linear" x1="36.1558" y1="30.578" x2="54.4844" y2="65.0806" '
+        b'gradientUnits="userSpaceOnUse"> <stop/> <stop offset="1" stop-opacity="0"/> </linearGradient> </defs> </svg>'
+    )
+    bucket_folder = uuid_factory()
+    bucket_path_1 = f"{bucket_folder}/{file_name_1}"
+    bucket_path_2 = f"{bucket_folder}/{file_name_2}"
+    file_path_1 = tmp_path / file_name_1
+    file_path_2 = tmp_path / file_name_2
+    with open(file_path_1, "wb") as f:
+        f.write(file_content)
+    with open(file_path_2, "wb") as f:
+        f.write(file_content)
+
+    return [
+        FileForTesting(
+            name=file_name_1,
+            local_path=str(file_path_1),
+            bucket_folder=bucket_folder,
+            bucket_path=bucket_path_1,
+            mime_type="image/svg+xml",
+            file_content=file_content,
+        ),
+        FileForTesting(
+            name=file_name_2,
+            local_path=str(file_path_2),
+            bucket_folder=bucket_folder,
+            bucket_path=bucket_path_2,
+            mime_type="image/svg+xml",
+            file_content=file_content,
+        ),
+    ]
+
+
 # TODO: Test create_bucket, delete_bucket, empty_bucket, list_buckets, fileAPI.list before upload test
 
 
@@ -192,6 +240,26 @@ def test_client_create_signed_url(
     assert response.content == file.file_content
 
 
+def test_client_create_signed_urls(
+    storage_file_client: SyncBucketProxy, multi_file: list[FileForTesting]
+) -> None:
+    """Ensure we can create signed urls for files in a bucket"""
+    paths = []
+    for file in multi_file:
+        paths.append(file.bucket_path)
+        storage_file_client.upload(
+            file.bucket_path, file.local_path, {"content-type": file.mime_type}
+        )
+
+    signed_urls = storage_file_client.create_signed_urls(paths, 10)
+
+    with HttpxClient() as client:
+        for url in signed_urls:
+            response = client.get(url["signedURL"])
+            response.raise_for_status()
+            assert response.content == multi_file[0].file_content
+
+
 def test_client_get_public_url(
     storage_file_client_public: SyncBucketProxy, file: FileForTesting
 ) -> None: