feat: multipart upload

pyproject.toml

@@ -31,6 +31,8 @@ pyparsing = ">=2.0,<3"
 clique = "==1.6.1"
 websocket-client = ">=0.40.0,<1"
 platformdirs = ">=4.0.0,<5"
+httpx = "^0.27.0"
+anyio = "^4.3.0"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.7.0"

source/ftrack_api/_http.py

@@ -0,0 +1,26 @@
+# :coding: utf-8
+# :copyright: Copyright (c) 2024 ftrack
+
+import httpx
+import os
+from pathlib import Path
+
+
+def _get_ssl_context():
+    ssl_context = httpx.create_ssl_context()
+
+    requests_ca_env = os.environ.get("REQUESTS_CA_BUNDLE")
+    if not requests_ca_env:
+        return
+
+    ca_path = Path(requests_ca_env)
+
+    if ca_path.is_file():
+        ssl_context.load_verify_locations(cafile=str(ca_path))
+    elif ca_path.is_dir():
+        ssl_context.load_verify_locations(capath=str(ca_path))
+
+    return ssl_context
+
+
+ssl_context = _get_ssl_context()

source/ftrack_api/accessor/server.py

@@ -4,13 +4,13 @@
 import os
 import hashlib
 import base64
-import json
 
 import requests
 
 from .base import Accessor
 from ..data import String
 import ftrack_api.exception
+from ftrack_api.uploader import Uploader
 import ftrack_api.symbol
 
 
@@ -72,7 +72,6 @@ def _read(self):
     def _write(self):
         """Write current data to remote key."""
         position = self.tell()
-        self.seek(0)
 
         # Retrieve component from cache to construct a filename.
         component = self._session.get("FileComponent", self.resource_identifier)
@@ -89,28 +88,16 @@ def _write(self):
             name = "{0}.{1}".format(name, component["file_type"].lstrip("."))
 
         try:
-            metadata = self._session.get_upload_metadata(
+            uploader = Uploader(
+                self._session,
                 component_id=self.resource_identifier,
                 file_name=name,
                 file_size=self._get_size(),
+                file=self.wrapped_file,
                 checksum=self._compute_checksum(),
             )
+            uploader.start()
         except Exception as error:
-            raise ftrack_api.exception.AccessorOperationFailedError(
-                "Failed to get put metadata: {0}.".format(error)
-            )
-
-        # Ensure at beginning of file before put.
-        self.seek(0)
-
-        # Put the file based on the metadata.
-        response = requests.put(
-            metadata["url"], data=self.wrapped_file, headers=metadata["headers"]
-        )
-
-        try:
-            response.raise_for_status()
-        except requests.exceptions.HTTPError as error:
             raise ftrack_api.exception.AccessorOperationFailedError(
                 "Failed to put file to server: {0}.".format(error)
             )
@@ -120,8 +107,7 @@ def _write(self):
     def _get_size(self):
         """Return size of file in bytes."""
         position = self.tell()
-        self.seek(0, os.SEEK_END)
-        length = self.tell()
+        length = self.seek(0, os.SEEK_END)
         self.seek(position)
         return length
 

source/ftrack_api/data.py

@@ -5,6 +5,7 @@
 import os
 from abc import ABCMeta, abstractmethod
 import tempfile
+import typing
 
 
 class Data(metaclass=ABCMeta):
@@ -25,14 +26,15 @@ def write(self, content):
     def flush(self):
         """Flush buffers ensuring data written."""
 
-    def seek(self, offset, whence=os.SEEK_SET):
+    def seek(self, offset, whence=os.SEEK_SET) -> int:
         """Move internal pointer by *offset*.
 
         The *whence* argument is optional and defaults to os.SEEK_SET or 0
         (absolute file positioning); other values are os.SEEK_CUR or 1
         (seek relative to the current position) and os.SEEK_END or 2
         (seek relative to the file's end).
 
+        Return the new absolute position.
         """
         raise NotImplementedError("Seek not supported.")
 
@@ -49,7 +51,7 @@ def close(self):
 class FileWrapper(Data):
     """Data wrapper for Python file objects."""
 
-    def __init__(self, wrapped_file):
+    def __init__(self, wrapped_file: typing.IO):
         """Initialise access to *wrapped_file*."""
         self.wrapped_file = wrapped_file
         self._read_since_last_write = False
@@ -81,7 +83,7 @@ def flush(self):
 
     def seek(self, offset, whence=os.SEEK_SET):
         """Move internal pointer by *offset*."""
-        self.wrapped_file.seek(offset, whence)
+        return self.wrapped_file.seek(offset, whence)
 
     def tell(self):
         """Return current position of internal pointer."""

source/ftrack_api/session.py

@@ -2249,7 +2249,9 @@ def encode_media(self, media, version_id=None, keep_original="auto"):
 
         return self.get("Job", result[0]["job_id"])
 
-    def get_upload_metadata(self, component_id, file_name, file_size, checksum=None):
+    def get_upload_metadata(
+        self, component_id, file_name, file_size, checksum=None, parts=None
+    ):
         """Return URL and headers used to upload data for *component_id*.
 
         *file_name* and *file_size* should match the components details.
@@ -2268,6 +2270,7 @@ def get_upload_metadata(self, component_id, file_name, file_size, checksum=None)
             "file_name": file_name,
             "file_size": file_size,
             "checksum": checksum,
+            "parts": parts,
         }
 
         try:
@@ -2286,6 +2289,29 @@ def get_upload_metadata(self, component_id, file_name, file_size, checksum=None)
 
         return result[0]
 
+    def complete_multipart_upload(self, component_id, upload_id, parts):
+        operation = {
+            "action": "complete_multipart_upload",
+            "component_id": component_id,
+            "upload_id": upload_id,
+            "parts": parts,
+        }
+
+        try:
+            result = self.call([operation])
+        except ftrack_api.exception.ServerError as error:
+            # Raise informative error if the action is not supported.
+            if "Invalid action u'complete_multipart_upload'" in error.message:
+                raise ftrack_api.exception.ServerCompatibilityError(
+                    "Server version {0!r} does not support "
+                    '"complete_multipart_upload", please update server and try '
+                    "again.".format(self.server_information.get("version"))
+                )
+            else:
+                raise
+
+        return result[0]
+
     def send_user_invite(self, user):
         """Send a invitation to the provided *user*.