feat: add transformation bindings

storage3/_async/file_api.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import urllib.parse
 from dataclasses import dataclass, field
 from io import BufferedReader, FileIO
 from pathlib import Path
@@ -8,7 +9,13 @@
 from httpx import HTTPError, Response
 
 from ..constants import DEFAULT_FILE_OPTIONS, DEFAULT_SEARCH_OPTIONS
-from ..types import BaseBucket, ListBucketFilesOptions, RequestMethod
+from ..types import (
+    BaseBucket,
+    CreateSignedURLOptions,
+    ListBucketFilesOptions,
+    RequestMethod,
+    TransformOptions,
+)
 from ..utils import AsyncClient, StorageException
 
 __all__ = ["AsyncBucket"]
@@ -44,7 +51,9 @@ async def _request(
 
         return response
 
-    async def create_signed_url(self, path: str, expires_in: int) -> dict[str, str]:
+    async def create_signed_url(
+        self, path: str, expires_in: int, options: CreateSignedURLOptions = {}
+    ) -> dict[str, str]:
         """
         Parameters
         ----------
@@ -65,15 +74,18 @@ async def create_signed_url(self, path: str, expires_in: int) -> dict[str, str]:
         ] = f"{self._client.base_url}{cast(str, data['signedURL']).lstrip('/')}"
         return data
 
-    async def get_public_url(self, path: str) -> str:
+    async def get_public_url(self, path: str, options: TransformOptions = {}) -> str:
         """
         Parameters
         ----------
         path
             file path, including the path and file name. For example `folder/image.png`.
         """
+        render_path = "render/image" if options.get("transform") else "object"
+        transformation_query = urllib.parse.urlencode(options)
+        query_string = f"?{transformation_query}" if transformation_query else ""
         _path = self._get_final_path(path)
-        return f"{self._client.base_url}object/public/{_path}"
+        return f"{self._client.base_url}{render_path}/public/{_path}{query_string}"
 
     async def move(self, from_path: str, to_path: str) -> dict[str, str]:
         """
@@ -97,6 +109,28 @@ async def move(self, from_path: str, to_path: str) -> dict[str, str]:
         )
         return res.json()
 
+    async def copy(self, from_path: str, to_path: str) -> dict[str, str]:
+        """
+        Copies an existing file to a new path in the same bucket.
+
+        Parameters
+        ----------
+        from_path
+            The original file path, including the current file name. For example `folder/image.png`.
+        to_path
+            The new file path, including the new file name. For example `folder/image-copy.png`.
+        """
+        res = await self._request(
+            "POST",
+            "/object/copy",
+            json={
+                "bucketId": self.id,
+                "sourceKey": from_path,
+                "destinationKey": to_path,
+            },
+        )
+        return res.json()
+
     async def remove(self, paths: list) -> dict[str, str]:
         """
         Deletes files within the same bucket
@@ -139,7 +173,7 @@ async def list(
         )
         return response.json()
 
-    async def download(self, path: str) -> bytes:
+    async def download(self, path: str, options: TransformOptions = {}) -> bytes:
         """
         Downloads a file.
 
@@ -148,10 +182,16 @@ async def download(self, path: str) -> bytes:
         path
             The file path to be downloaded, including the path and file name. For example `folder/image.png`.
         """
+        render_path = (
+            "render/image/authenticated" if options.get("transform") else "object"
+        )
+        transformation_query = urllib.parse.urlencode(options)
+        query_string = f"?{transformation_query}" if transformation_query else ""
+
         _path = self._get_final_path(path)
         response = await self._request(
             "GET",
-            f"/object/{_path}",
+            f"{render_path}/{_path}{query_string}",
         )
         return response.content
 

storage3/_sync/file_api.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import urllib.parse
 from dataclasses import dataclass, field
 from io import BufferedReader, FileIO
 from pathlib import Path
@@ -8,7 +9,13 @@
 from httpx import HTTPError, Response
 
 from ..constants import DEFAULT_FILE_OPTIONS, DEFAULT_SEARCH_OPTIONS
-from ..types import BaseBucket, ListBucketFilesOptions, RequestMethod
+from ..types import (
+    BaseBucket,
+    CreateSignedURLOptions,
+    ListBucketFilesOptions,
+    RequestMethod,
+    TransformOptions,
+)
 from ..utils import StorageException, SyncClient
 
 __all__ = ["SyncBucket"]
@@ -44,7 +51,9 @@ def _request(
 
         return response
 
-    def create_signed_url(self, path: str, expires_in: int) -> dict[str, str]:
+    def create_signed_url(
+        self, path: str, expires_in: int, options: CreateSignedURLOptions = {}
+    ) -> dict[str, str]:
         """
         Parameters
         ----------
@@ -65,15 +74,18 @@ def create_signed_url(self, path: str, expires_in: int) -> dict[str, str]:
         ] = f"{self._client.base_url}{cast(str, data['signedURL']).lstrip('/')}"
         return data
 
-    def get_public_url(self, path: str) -> str:
+    def get_public_url(self, path: str, options: TransformOptions = {}) -> str:
         """
         Parameters
         ----------
         path
             file path, including the path and file name. For example `folder/image.png`.
         """
+        render_path = "render/image" if options.get("transform") else "object"
+        transformation_query = urllib.parse.urlencode(options)
+        query_string = f"?{transformation_query}" if transformation_query else ""
         _path = self._get_final_path(path)
-        return f"{self._client.base_url}object/public/{_path}"
+        return f"{self._client.base_url}{render_path}/public/{_path}{query_string}"
 
     def move(self, from_path: str, to_path: str) -> dict[str, str]:
         """
@@ -97,6 +109,28 @@ def move(self, from_path: str, to_path: str) -> dict[str, str]:
         )
         return res.json()
 
+    def copy(self, from_path: str, to_path: str) -> dict[str, str]:
+        """
+        Copies an existing file to a new path in the same bucket.
+
+        Parameters
+        ----------
+        from_path
+            The original file path, including the current file name. For example `folder/image.png`.
+        to_path
+            The new file path, including the new file name. For example `folder/image-copy.png`.
+        """
+        res = self._request(
+            "POST",
+            "/object/copy",
+            json={
+                "bucketId": self.id,
+                "sourceKey": from_path,
+                "destinationKey": to_path,
+            },
+        )
+        return res.json()
+
     def remove(self, paths: list) -> dict[str, str]:
         """
         Deletes files within the same bucket
@@ -139,7 +173,7 @@ def list(
         )
         return response.json()
 
-    def download(self, path: str) -> bytes:
+    def download(self, path: str, options: TransformOptions = {}) -> bytes:
         """
         Downloads a file.
 
@@ -148,10 +182,16 @@ def download(self, path: str) -> bytes:
         path
             The file path to be downloaded, including the path and file name. For example `folder/image.png`.
         """
+        render_path = (
+            "render/image/authenticated" if options.get("transform") else "object"
+        )
+        transformation_query = urllib.parse.urlencode(options)
+        query_string = f"?{transformation_query}" if transformation_query else ""
+
         _path = self._get_final_path(path)
         response = self._request(
             "GET",
-            f"/object/{_path}",
+            f"{render_path}/{_path}{query_string}",
         )
         return response.content
 
@@ -188,6 +228,7 @@ def upload(
             files = {"file": (filename, open(file, "rb"), headers.pop("content-type"))}
 
         _path = self._get_final_path(path)
+
         return self._request(
             "POST",
             f"/object/{_path}",

storage3/types.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime
+from typing import Optional, Union
 
 import dateutil.parser
 from typing_extensions import Literal, TypedDict
@@ -35,3 +36,14 @@ class ListBucketFilesOptions(TypedDict):
     limit: int
     offset: int
     sortBy: _sortByType
+
+
+class TransformOptions(TypedDict):
+    height: Optional[float]
+    width: Optional[float]
+    resize: Optional[Union[Literal["cover"], Literal["contain"], Literal["fill"]]]
+
+
+class CreateSignedURLOptions(TypedDict):
+    download: Optional[Union[str, bool]]
+    transform: Optional[TransformOptions]