diff --git a/tests/unit-tests/files_transformations_test.py b/tests/unit-tests/files_transformations_test.py new file mode 100644 index 0000000..7aa03c1 --- /dev/null +++ b/tests/unit-tests/files_transformations_test.py @@ -0,0 +1,57 @@ +# +# Licensed to Xatabase, Inc under one or more contributor +# license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright +# ownership. Xatabase, Inc licenses this file to you under the +# Apache License, Version 2.0 (the "License"); you may not +# use this file except in compliance with the License. You +# may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +import unittest + +from xata.client import XataClient + + +class TestFileTransformations(unittest.TestCase): + def test_single_transform_url(self): + client = XataClient(api_key="api_key", workspace_id="ws_id") + + url = client.files().transform_url( + "https://us-east-1.storage.xata.sh/4u1fh2o6p10blbutjnphcste94", + { + "height": 100, + } + ) + + expected = "https://us-east-1.storage.xata.sh/transform/height=100/4u1fh2o6p10blbutjnphcste94" + assert url == expected + + def test_multiple_transform_url(self): + client = XataClient(api_key="api_key", workspace_id="ws_id") + + url = client.files().transform_url( + "https://us-east-1.storage.xata.sh/4u1fh2o6p10blbutjnphcste94", + { + "width": 100, + "height": 100, + "fit": "cover", + "gravity": { + "x": 0, + "y": 1 + } + } + ) + + excepted = "https://us-east-1.storage.xata.sh/transform/width=100,height=100,fit=cover,gravity=0x1/4u1fh2o6p10blbutjnphcste94" + + assert url == excepted diff --git a/xata/api/files.py b/xata/api/files.py index df2d103..93ba1a8 100644 --- a/xata/api/files.py +++ b/xata/api/files.py @@ -238,15 +238,16 @@ def delete( url_path = f"/db/{db_branch_name}/tables/{table_name}/data/{record_id}/column/{column_name}/file" return self.request("DELETE", url_path) - def transform(self, url: str, operations: dict[str, any]) -> bytes: + def transform_url(self, url: str, operations: dict[str, any]) -> str: """ - Image transformations + Image transformations url + Returns the file url only for the given operations. All possible combinations: https://xata.io/docs/concepts/file-storage#image-transformations :param url: str Public or signed URL of the image :param operations: dict Image operations - :return Response + :return str """ # valid url ? url_parts = url.split("/") @@ -256,21 +257,38 @@ def transform(self, url: str, operations: dict[str, any]) -> bytes: # build operations - turn objects into lists ops = [] for k, v in operations.items(): - if type(v) is dict: + # Coordinates on the gravity operation use an "x" as separator + if type(v) is dict and k == "gravity": + v = "x".join([str(x) for x in v.values()]) + # All the other ones use a semicolon. + elif type(v) is dict: v = ";".join([str(x) for x in v.values()]) ops.append(f"{k}={v}") - # ops = ",".join(["%s=%s" % (k, ";".join(list(v.values()))) if type(v) is dict else f"{k}={v}" for k, v in operations.items()]) + if len(ops) == 0: raise Exception("missing image operations") region = url_parts[2].split(".")[0] file_id = url_parts[-1] + # signed url if len(url_parts) == 5: - endpoint = "https://%s.xata.sh/transform/%s/file/%s" % (region, ",".join(ops), file_id) + return "https://%s.xata.sh/transform/%s/file/%s" % (region, ",".join(ops), file_id) # public url else: - endpoint = "https://%s.storage.xata.sh/transform/%s/%s" % (region, ",".join(ops), file_id) + return "https://%s.storage.xata.sh/transform/%s/%s" % (region, ",".join(ops), file_id) + + def transform(self, url: str, operations: dict[str, any]) -> bytes: + """ + Image transformations + All possible combinations: https://xata.io/docs/concepts/file-storage#image-transformations + + :param url: str Public or signed URL of the image + :param operations: dict Image operations + + :return Response + """ + endpoint = self.transform_url(url, operations) resp = request("GET", endpoint) if resp.status_code != 200: