SG-41387 Add parameter created_at for sg.upload (#444)

* Add parameter created_at for sg.upload

Author: jerrycheng9

shotgun_api3/shotgun.py

@@ -2610,6 +2610,7 @@ def upload(
         field_name: Optional[str] = None,
         display_name: Optional[str] = None,
         tag_list: Optional[str] = None,
+        created_at: Optional[datetime.datetime] = None,
     ) -> int:
         """
         Upload a file to the specified entity.
@@ -2634,6 +2635,8 @@ def upload(
             This field must be a File/Link field type.
         :param str display_name: The display name to use for the file. Defaults to the file name.
         :param str tag_list: comma-separated string of tags to assign to the file.
+        :param datetime.datetime created_at: Optional datetime value to set as the created_at
+            time for the Attachment entity. If ``None``, the server will use the current time.
         :returns: Id of the Attachment entity that was created for the image.
         :rtype: int
         :raises: :class:`ShotgunError` on upload failure.
@@ -2662,6 +2665,13 @@ def upload(
         if os.path.getsize(path) == 0:
             raise ShotgunError("Path cannot be an empty file: '%s'" % path)
 
+        if created_at is not None:
+            if not isinstance(created_at, datetime.datetime):
+                raise ShotgunError(
+                    "created_at must be a datetime.datetime instance, got '%s'"
+                    % type(created_at)
+                )
+
         is_thumbnail = field_name in [
             "thumb_image",
             "filmstrip_thumb_image",
@@ -2679,6 +2689,7 @@ def upload(
                 display_name,
                 tag_list,
                 is_thumbnail,
+                created_at,
             )
         else:
             return self._upload_to_sg(
@@ -2689,6 +2700,7 @@ def upload(
                 display_name,
                 tag_list,
                 is_thumbnail,
+                created_at,
             )
 
     def _upload_to_storage(
@@ -2700,6 +2712,7 @@ def _upload_to_storage(
         display_name: Optional[str],
         tag_list: Optional[str],
         is_thumbnail: bool,
+        created_at: Optional[datetime.datetime] = None,
     ) -> int:
         """
         Internal function to upload a file to the Cloud storage and link it to the specified entity.
@@ -2712,6 +2725,7 @@ def _upload_to_storage(
         :param str display_name: The display name to use for the file. Defaults to the file name.
         :param str tag_list: comma-separated string of tags to assign to the file.
         :param bool is_thumbnail: indicates if the attachment is a thumbnail.
+        :param datetime created_at: The datetime to set for the attachment.
         :returns: Id of the Attachment entity that was created for the image.
         :rtype: int
         """
@@ -2723,7 +2737,7 @@ def _upload_to_storage(
         is_multipart_upload = os.path.getsize(path) > self._MULTIPART_UPLOAD_CHUNK_SIZE
 
         upload_info = self._get_attachment_upload_info(
-            is_thumbnail, filename, is_multipart_upload
+            is_thumbnail, filename, is_multipart_upload, created_at=created_at
         )
 
         # Step 2: upload the file
@@ -2768,6 +2782,8 @@ def _upload_to_storage(
             # None gets converted to a string and added as a tag...
             if tag_list:
                 params["tag_list"] = tag_list
+            if created_at is not None:
+                params["created_at"] = created_at
 
         result = self._send_form(url, params)
         if not result.startswith("1"):
@@ -2790,6 +2806,7 @@ def _upload_to_sg(
         display_name: Optional[str],
         tag_list: Optional[str],
         is_thumbnail: bool,
+        created_at: Optional[datetime.datetime] = None,
     ) -> int:
         """
         Internal function to upload a file to Shotgun and link it to the specified entity.
@@ -2802,7 +2819,7 @@ def _upload_to_sg(
         :param str display_name: The display name to use for the file. Defaults to the file name.
         :param str tag_list: comma-separated string of tags to assign to the file.
         :param bool is_thumbnail: indicates if the attachment is a thumbnail.
-
+        :param datetime created_at: The datetime to set for the attachment.
         :returns: Id of the Attachment entity that was created for the image.
         :rtype: int
         """
@@ -2848,6 +2865,8 @@ def _upload_to_sg(
             # None gets converted to a string and added as a tag...
             if tag_list:
                 params["tag_list"] = tag_list
+            if created_at is not None:
+                params["created_at"] = created_at.isoformat()
 
             params["file"] = open(path, "rb")
 
@@ -2863,14 +2882,21 @@ def _upload_to_sg(
         return attachment_id
 
     def _get_attachment_upload_info(
-        self, is_thumbnail: bool, filename: str, is_multipart_upload: bool
+        self,
+        is_thumbnail: bool,
+        filename: str,
+        is_multipart_upload: bool,
+        created_at: Optional[datetime.datetime] = None,
     ) -> Dict[str, Any]:
         """
         Internal function to get the information needed to upload a file to Cloud storage.
 
         :param bool is_thumbnail: indicates if the attachment is a thumbnail.
         :param str filename: name of the file that will be uploaded.
         :param bool is_multipart_upload: Indicates if we want multi-part upload information back.
+        :param datetime created_at: Optional datetime to use as the upload timestamp.
+            When provided, the server generates the S3 key with this timestamp so that
+            the Attachment's created_at and its storage path stay in sync.
 
         :returns: dictionary containing upload details from the server.
             These details are used throughout the upload process.
@@ -2886,6 +2912,9 @@ def _get_attachment_upload_info(
 
         params["multipart_upload"] = is_multipart_upload
 
+        if created_at is not None:
+            params["created_at"] = created_at.isoformat()
+
         upload_url = "/upload/api_get_upload_link_info"
         url = urllib.parse.urlunparse(
             (self.config.scheme, self.config.server, upload_url, None, None, None)

tests/test_api.py

@@ -325,6 +325,232 @@ def test_upload_to_sg(self, mock_send_form):
         mock_send_form.assert_called_once()
         self.sg.server_info["s3_direct_uploads_enabled"] = True
 
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    def test_upload_to_sg_with_created_at(self, mock_send_form):
+        """
+        Verify that created_at is passed (in ISO 8601 string form) as a form
+        parameter when uploading non-thumbnail attachments via _upload_to_sg().
+        """
+        self.sg.server_info["s3_direct_uploads_enabled"] = False
+        mock_send_form.return_value = "1\n:456\nasd"
+        this_dir, _ = os.path.split(__file__)
+        u_path = os.path.abspath(
+            os.path.expanduser(glob.glob(os.path.join(this_dir, "Noëlご.jpg"))[0])
+        )
+        custom_time = datetime.datetime(2026, 2, 15, 10, 30, 0)
+        self.sg.upload(
+            "Version",
+            self.version["id"],
+            u_path,
+            "attachments",
+            created_at=custom_time,
+        )
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertIn("created_at", params)
+        self.assertEqual(params["created_at"], custom_time.isoformat())
+        self.sg.server_info["s3_direct_uploads_enabled"] = True
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    def test_upload_to_sg_without_created_at(self, mock_send_form):
+        """
+        Verify that created_at is NOT included in form parameters when omitted.
+        """
+        self.sg.server_info["s3_direct_uploads_enabled"] = False
+        mock_send_form.return_value = "1\n:456\nasd"
+        this_dir, _ = os.path.split(__file__)
+        u_path = os.path.abspath(
+            os.path.expanduser(glob.glob(os.path.join(this_dir, "Noëlご.jpg"))[0])
+        )
+        self.sg.upload(
+            "Version",
+            self.version["id"],
+            u_path,
+            "attachments",
+        )
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertNotIn("created_at", params)
+        self.sg.server_info["s3_direct_uploads_enabled"] = True
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    @unittest.mock.patch("shotgun_api3.Shotgun._upload_file_to_storage")
+    @unittest.mock.patch("shotgun_api3.Shotgun._get_attachment_upload_info")
+    def test_upload_to_storage_with_created_at(
+        self, mock_get_info, mock_upload_file, mock_send_form
+    ):
+        """
+        Verify that created_at is passed as a form parameter when uploading
+        non-thumbnail attachments via _upload_to_storage() (S3/cloud path),
+        and that it is forwarded to _get_attachment_upload_info() so the server
+        can use it when generating the storage path.
+        """
+        self.sg.server_info["s3_direct_uploads_enabled"] = True
+        self.sg.server_info["s3_enabled_upload_types"] = {"Version": "*"}
+        mock_get_info.return_value = {
+            "upload_url": "https://example.com/upload",
+            "upload_info": {"upload_type": "s3"},
+        }
+        mock_send_form.return_value = "1\n:456\nasd"
+        this_dir, _ = os.path.split(__file__)
+        u_path = os.path.abspath(
+            os.path.expanduser(glob.glob(os.path.join(this_dir, "Noëlご.jpg"))[0])
+        )
+        custom_time = datetime.datetime(2026, 2, 15, 10, 30, 0)
+        self.sg.upload(
+            "Version",
+            self.version["id"],
+            u_path,
+            "attachments",
+            created_at=custom_time,
+        )
+        mock_get_info.assert_called_once()
+        _, get_info_kwargs = mock_get_info.call_args
+        self.assertEqual(get_info_kwargs.get("created_at"), custom_time)
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertIn("created_at", params)
+        self.assertEqual(params["created_at"], custom_time)
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    @unittest.mock.patch("shotgun_api3.Shotgun._upload_file_to_storage")
+    @unittest.mock.patch("shotgun_api3.Shotgun._get_attachment_upload_info")
+    def test_upload_to_storage_without_created_at(
+        self, mock_get_info, mock_upload_file, mock_send_form
+    ):
+        """
+        Verify that created_at is NOT included in form parameters when omitted
+        via _upload_to_storage() (S3/cloud path), and that None is forwarded to
+        _get_attachment_upload_info().
+        """
+        self.sg.server_info["s3_direct_uploads_enabled"] = True
+        self.sg.server_info["s3_enabled_upload_types"] = {"Version": "*"}
+        mock_get_info.return_value = {
+            "upload_url": "https://example.com/upload",
+            "upload_info": {"upload_type": "s3"},
+        }
+        mock_send_form.return_value = "1\n:456\nasd"
+        this_dir, _ = os.path.split(__file__)
+        u_path = os.path.abspath(
+            os.path.expanduser(glob.glob(os.path.join(this_dir, "Noëlご.jpg"))[0])
+        )
+        self.sg.upload(
+            "Version",
+            self.version["id"],
+            u_path,
+            "attachments",
+        )
+        mock_get_info.assert_called_once()
+        _, get_info_kwargs = mock_get_info.call_args
+        self.assertIsNone(get_info_kwargs.get("created_at"))
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertNotIn("created_at", params)
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    def test_get_attachment_upload_info_with_created_at(self, mock_send_form):
+        """
+        Verify that _get_attachment_upload_info() includes created_at in the
+        form params (as ISO 8601 string) when sending the request to
+        /upload/api_get_upload_link_info.
+        """
+        mock_send_form.return_value = (
+            "1\nhttps://example.com/upload\n2026-02-15T10:30:00\nAttachment\nupload-123"
+        )
+        custom_time = datetime.datetime(2026, 2, 15, 10, 30, 0)
+        self.sg._get_attachment_upload_info(
+            is_thumbnail=False,
+            filename="example.jpg",
+            is_multipart_upload=False,
+            created_at=custom_time,
+        )
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertIn("created_at", params)
+        self.assertEqual(params["created_at"], custom_time.isoformat())
+        self.assertEqual(params["upload_type"], "Attachment")
+        self.assertEqual(params["filename"], "example.jpg")
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    def test_get_attachment_upload_info_without_created_at(self, mock_send_form):
+        """
+        Verify that _get_attachment_upload_info() does NOT include created_at
+        in the form params when it is omitted (None).
+        """
+        mock_send_form.return_value = (
+            "1\nhttps://example.com/upload\n2026-02-15T10:30:00\nAttachment\nupload-123"
+        )
+        self.sg._get_attachment_upload_info(
+            is_thumbnail=False,
+            filename="example.jpg",
+            is_multipart_upload=False,
+        )
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertNotIn("created_at", params)
+
+    @unittest.mock.patch("shotgun_api3.Shotgun._send_form")
+    def test_get_attachment_upload_info_thumbnail_with_created_at(self, mock_send_form):
+        """
+        Verify that _get_attachment_upload_info() forwards created_at even when
+        is_thumbnail=True. The implementation does not gate created_at on the
+        thumbnail flag at this stage; the server is responsible for handling
+        thumbnail-specific behavior.
+        """
+        mock_send_form.return_value = (
+            "1\nhttps://example.com/upload\n2026-02-15T10:30:00\nThumbnail\nupload-123"
+        )
+        custom_time = datetime.datetime(2026, 2, 15, 10, 30, 0)
+        self.sg._get_attachment_upload_info(
+            is_thumbnail=True,
+            filename="thumb.jpg",
+            is_multipart_upload=False,
+            created_at=custom_time,
+        )
+        mock_send_form.assert_called_once()
+        mock_send_form_args, _ = mock_send_form.call_args
+        params = mock_send_form_args[1]
+        self.assertEqual(params["upload_type"], "Thumbnail")
+        self.assertIn("created_at", params)
+        self.assertEqual(params["created_at"], custom_time.isoformat())
+
+    def test_upload_created_at_invalid_type(self):
+        """
+        Verify that passing a non-datetime value for created_at raises ShotgunError.
+        """
+        this_dir, _ = os.path.split(__file__)
+        u_path = os.path.abspath(
+            os.path.expanduser(glob.glob(os.path.join(this_dir, "Noëlご.jpg"))[0])
+        )
+        with self.assertRaisesRegex(
+            shotgun_api3.ShotgunError,
+            "created_at must be a datetime.datetime instance",
+        ):
+            self.sg.upload(
+                "Version",
+                self.version["id"],
+                u_path,
+                "attachments",
+                created_at="2026-02-15T10:30:00Z",
+            )
+        with self.assertRaisesRegex(
+            shotgun_api3.ShotgunError,
+            "created_at must be a datetime.datetime instance",
+        ):
+            self.sg.upload(
+                "Version",
+                self.version["id"],
+                u_path,
+                "attachments",
+                created_at=1234567890,
+            )
+
     def test_upload_thumbnail_in_create(self):
         """Upload a thumbnail via the create method"""
         this_dir, _ = os.path.split(__file__)