Always write through a temporary file

Author: aabmass

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_upload/completion_hook.py

@@ -18,6 +18,9 @@
 import json
 import logging
 import posixpath
+import shutil
+from tempfile import SpooledTemporaryFile
+import tempfile
 import threading
 from base64 import b64encode
 from concurrent.futures import Future, ThreadPoolExecutor
@@ -190,15 +193,18 @@ def _do_upload(
             else "application/jsonl"
         )
 
-        with self._fs.open(path, "w", content_type=content_type) as file:
+        with tempfile.NamedTemporaryFile(mode="w+") as temp_file:
             for message in message_lines:
                 json.dump(
                     message,
-                    file,
+                    temp_file,
                     separators=(",", ":"),
                     cls=Base64JsonEncoder,
                 )
-                file.write("\n")
+                temp_file.write("\n")
+
+            temp_file.flush()
+            self._fs.put_file(temp_file.name, path, content_type=content_type)
 
     def on_completion(
         self,

util/opentelemetry-util-genai/tests/test_upload.py

@@ -139,7 +139,7 @@ def blocked_upload(*args: Any, **kwargs: Any):
             return MagicMock()
 
         try:
-            self.mock_fs.open.side_effect = blocked_upload
+            self.mock_fs.put_file.side_effect = blocked_upload
             yield
         finally:
             unblock_upload.set()
@@ -157,7 +157,7 @@ def test_upload_then_shutdown(self):
         self.hook.shutdown()
 
         self.assertEqual(
-            self.mock_fs.open.call_count,
+            self.mock_fs.put_file.call_count,
             3,
             "should have uploaded 3 files",
         )
@@ -173,7 +173,7 @@ def test_upload_blocked(self):
                 )
 
             self.assertLessEqual(
-                self.mock_fs.open.call_count,
+                self.mock_fs.put_file.call_count,
                 MAXSIZE,
                 f"uploader should only be called {MAXSIZE=} times",
             )
@@ -201,7 +201,7 @@ def test_shutdown_timeout(self):
             self.hook.shutdown(timeout_sec=0.01)
 
     def test_failed_upload_logs(self):
-        self.mock_fs.open.side_effect = RuntimeError("failed to upload")
+        self.mock_fs.put_file.side_effect = RuntimeError("failed to upload")
 
         with self.assertLogs(level=logging.ERROR) as logs:
             self.hook.on_completion(
@@ -234,8 +234,8 @@ def test_upload_format_sets_content_type(self):
             )
             hook.shutdown()
 
-            self.mock_fs.open.assert_called_with(
-                ANY, "w", content_type=expect_content_type
+            self.mock_fs.put_file.assert_called_with(
+                ANY, ANY, content_type=expect_content_type
             )
 
     def test_parse_upload_format_envvar(self):