Skip to content

Commit

Permalink
Create tranform url helper (#178)
Browse files Browse the repository at this point in the history
* Add basic unit test for files transform_url function

* Add test case with mutliple transform operations

* Format unit test code for file transformations

* Move url code to new transform_url function

* Fix wrong seperator for gravity coordinate values

* Clean up doc strings and comments for the file transform url
  • Loading branch information
sepal authored Oct 5, 2023
1 parent e2a38ec commit 2b7229d
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 7 deletions.
57 changes: 57 additions & 0 deletions tests/unit-tests/files_transformations_test.py
Original file line number Diff line number Diff line change
@@ -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
32 changes: 25 additions & 7 deletions xata/api/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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("/")
Expand All @@ -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:
Expand Down

0 comments on commit 2b7229d

Please sign in to comment.