chore(closes OPEN-8832): optionally upload data from external URLs to our storage

Author: gustavocidornelas

src/openlayer/lib/data/_upload.py

@@ -67,7 +67,7 @@ def upload_bytes(
             data=data,
             object_name=object_name,
             content_type=content_type,
-            fields=fields,
+            fields=fields or {},
         )
     elif storage == StorageType.GCP:
         return upload_bytes_put(

src/openlayer/lib/tracing/attachment_uploader.py

@@ -59,15 +59,22 @@ class AttachmentUploader:
     S3, GCS, Azure, and local storage backends.
     """
 
-    def __init__(self, client: "Openlayer", storage: StorageType = STORAGE):
+    def __init__(
+        self,
+        client: "Openlayer",
+        storage: StorageType = STORAGE,
+        url_upload_enabled: bool = False,
+    ):
         """Initialize the attachment uploader.
 
         Args:
             client: The Openlayer client instance.
             storage: Storage type override. Defaults to the global STORAGE setting.
+            url_upload_enabled: Whether to download and re-upload URL attachments.
         """
         self._client = client
         self._storage = storage
+        self._url_upload_enabled = url_upload_enabled
         self._storage_uri_cache: Dict[str, str] = {}  # checksum -> storage_uri
 
     def upload_attachment(self, attachment: "Attachment") -> "Attachment":
@@ -88,12 +95,25 @@ def upload_attachment(self, attachment: "Attachment") -> "Attachment":
             logger.debug("Attachment %s already uploaded", attachment.name)
             return attachment
 
-        # Skip if it has an external URL (no upload needed)
+        # Handle external URL attachments
         if attachment.url:
-            logger.debug(
-                "Attachment %s has external URL, skipping upload", attachment.name
-            )
-            return attachment
+            if self._url_upload_enabled:
+                logger.debug(
+                    "Downloading attachment %s from external URL for upload",
+                    attachment.name,
+                )
+                if not attachment.download_url():
+                    logger.warning(
+                        "Failed to download attachment %s from URL, skipping upload",
+                        attachment.name,
+                    )
+                    return attachment
+            else:
+                logger.debug(
+                    "Attachment %s has external URL, skipping upload",
+                    attachment.name,
+                )
+                return attachment
 
         # Check if we have data to upload
         if not attachment.has_data():
@@ -235,7 +255,11 @@ def process_step(step: "Step") -> int:
                     continue
                 seen_ids.add(attachment.id)
 
-                if not attachment.is_uploaded() and attachment.has_data():
+                needs_upload = not attachment.is_uploaded() and (
+                    attachment.has_data()
+                    or (self._url_upload_enabled and attachment.url)
+                )
+                if needs_upload:
                     self.upload_attachment(attachment)
                     if attachment.is_uploaded():
                         step_upload_count += 1
@@ -273,7 +297,10 @@ def get_uploader() -> Optional[AttachmentUploader]:
     if _uploader is None:
         client = tracer._get_client()
         if client:
-            _uploader = AttachmentUploader(client)
+            _uploader = AttachmentUploader(
+                client,
+                url_upload_enabled=tracer._configured_url_upload_enabled,
+            )
 
     return _uploader
 

src/openlayer/lib/tracing/attachments.py

@@ -326,6 +326,31 @@ def from_base64(
             checksum_md5=checksum_md5,
         )
 
+    def download_url(self) -> bool:
+        """Download data from self.url and populate _pending_bytes, size_bytes, checksum_md5.
+
+        Returns:
+            True if download succeeded, False otherwise.
+        """
+        if not self.url:
+            return False
+
+        try:
+            import httpx
+
+            with httpx.Client(follow_redirects=True) as http_client:
+                response = http_client.get(self.url)
+                response.raise_for_status()
+
+            data = response.content
+            self._pending_bytes = data
+            self.size_bytes = len(data)
+            self.checksum_md5 = hashlib.md5(data).hexdigest()
+            return True
+        except Exception as e:
+            logger.error("Failed to download attachment from URL %s: %s", self.url, e)
+            return False
+
     def get_bytes(self) -> Optional[bytes]:
         """Get the binary data for this attachment.
 

src/openlayer/lib/tracing/tracer.py

@@ -67,6 +67,7 @@
 
 # Attachment upload configuration
 _configured_attachment_upload_enabled: bool = False
+_configured_url_upload_enabled: bool = False
 
 # Background publishing configuration
 _configured_background_publish_enabled: bool = True
@@ -108,6 +109,7 @@ def configure(
     offline_buffer_path: Optional[str] = None,
     max_buffer_size: Optional[int] = None,
     attachment_upload_enabled: bool = False,
+    url_upload_enabled: bool = False,
     background_publish_enabled: bool = True,
 ) -> None:
     """Configure the Openlayer tracer with custom settings.
@@ -132,6 +134,10 @@ def configure(
         attachment_upload_enabled: Enable uploading of attachments (images, audio, etc.) to
             Openlayer storage. When enabled, attachments on steps will be uploaded during
             trace completion. Defaults to False.
+        url_upload_enabled: Enable downloading and re-uploading of external URL
+            attachments to Openlayer storage. When enabled, attachments that reference
+            external URLs will be fetched and uploaded so the platform has a durable copy.
+            Requires attachment_upload_enabled to also be True. Defaults to False.
         background_publish_enabled: Enable background publishing of traces. When enabled,
             attachment uploads and trace publishing happen in a background thread, allowing
             the main thread to return immediately. When disabled, tracing is synchronous.
@@ -166,7 +172,7 @@ def configure(
     """
     global _configured_api_key, _configured_pipeline_id, _configured_base_url, _configured_timeout, _configured_max_retries, _client
     global _configured_on_flush_failure, _configured_offline_buffer_enabled, _configured_offline_buffer_path, _configured_max_buffer_size, _offline_buffer
-    global _configured_attachment_upload_enabled, _configured_background_publish_enabled
+    global _configured_attachment_upload_enabled, _configured_url_upload_enabled, _configured_background_publish_enabled
 
     _configured_api_key = api_key
     _configured_pipeline_id = inference_pipeline_id
@@ -178,6 +184,7 @@ def configure(
     _configured_offline_buffer_path = offline_buffer_path
     _configured_max_buffer_size = max_buffer_size
     _configured_attachment_upload_enabled = attachment_upload_enabled
+    _configured_url_upload_enabled = url_upload_enabled
     _configured_background_publish_enabled = background_publish_enabled
 
     # Reset the client and buffer so they get recreated with new configuration