diff --git a/storage3/_async/file_api.py b/storage3/_async/file_api.py index a3e8ca5a..701ba4c7 100644 --- a/storage3/_async/file_api.py +++ b/storage3/_async/file_api.py @@ -12,6 +12,7 @@ from ..types import ( BaseBucket, CreateSignedURLOptions, + CreateSignedURLsOptions, FileOptions, ListBucketFilesOptions, RequestMethod, @@ -62,12 +63,20 @@ 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[ @@ -75,6 +84,35 @@ async def create_signed_url( ] = 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 diff --git a/storage3/_sync/file_api.py b/storage3/_sync/file_api.py index f8a2f086..09ec2e43 100644 --- a/storage3/_sync/file_api.py +++ b/storage3/_sync/file_api.py @@ -12,6 +12,7 @@ from ..types import ( BaseBucket, CreateSignedURLOptions, + CreateSignedURLsOptions, FileOptions, ListBucketFilesOptions, RequestMethod, @@ -62,12 +63,20 @@ 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[ @@ -75,6 +84,35 @@ def create_signed_url( ] = 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 diff --git a/storage3/types.py b/storage3/types.py index 9f905b63..d69d0e54 100644 --- a/storage3/types.py +++ b/storage3/types.py @@ -40,15 +40,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( diff --git a/tests/_async/test_client.py b/tests/_async/test_client.py index 7b84880c..74925a13 100644 --- a/tests/_async/test_client.py +++ b/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' ' + b' ' + b' ' + b' ' + b' ' + ) + 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: diff --git a/tests/_sync/test_client.py b/tests/_sync/test_client.py index d3622128..9f273460 100644 --- a/tests/_sync/test_client.py +++ b/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' ' + b' ' + b' ' + b' ' + b' ' + ) + 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: