remove s3 bucket polling when waiting for transformation results

Author: MattShirley

servicex/query_core.py

@@ -522,6 +522,7 @@ async def download_files(
         Task to monitor the list of files in the transform output's bucket. Any new files
         will be downloaded.
         """
+
         files_seen = set()
         result_uris = []
         download_tasks = []
@@ -557,15 +558,19 @@ async def get_signed_url(
             if self.minio:
                 # if self.minio exists, self.current_status will too
                 if self.current_status.files_completed > len(files_seen):
-                    files = await self.minio.list_bucket()
+                    files = await self.servicex.get_transformation_results(self.current_status.request_id)
+
                     for file in files:
-                        if file.filename not in files_seen:
+                        if 'file-path' not in file:
+                            continue
+
+                        file_path = file['file-path'].replace('/', ':')
+                        if file_path not in files_seen:
                             if signed_urls_only:
                                 download_tasks.append(
                                     loop.create_task(
                                         get_signed_url(
-                                            self.minio,
-                                            file.filename,
+                                            file_path,
                                             progress,
                                             download_progress,
                                         )
@@ -576,14 +581,14 @@ async def get_signed_url(
                                     loop.create_task(
                                         download_file(
                                             self.minio,
-                                            file.filename,
+                                            file_path,
                                             progress,
                                             download_progress,
                                             shorten_filename=self.configuration.shortened_downloaded_filename,  # NOQA: E501
                                         )
                                     )
                                 )  # NOQA 501
-                            files_seen.add(file.filename)
+                            files_seen.add(file_path)
 
             # Once the transform is complete and all files are seen we can stop polling.
             # Also, if we are just downloading or signing urls for a previous transform

servicex/servicex_adapter.py

@@ -228,14 +228,35 @@ async def delete_transform(self, transform_id=None):
                         f"Failed to delete transform {transform_id} - {msg}"
                     )
 
+    async def get_transformation_results(self, request_id: str):
+        headers = await self._get_authorization()
+        url = self.url + f'/servicex/internal/transformation/{request_id}/results'
+
+        async with ClientSession() as session:
+            async with session.get(headers=headers, url=url) as r:
+                if r.status == 403:
+                    raise AuthorizationError(
+                        f"Not authorized to access serviceX at {self.url}"
+                    )
+
+                if r.status == 404:
+                    raise ValueError(f"Request {request_id} not found")
+
+                if r.status != 200:
+                    msg = await _extract_message(r)
+                    raise RuntimeError(
+                        f"Failed with message: {msg}"
+                    )
+
+                return (await r.json())['results']
+
     async def cancel_transform(self, transform_id=None):
         headers = await self._get_authorization()
         path_template = f"/servicex/transformation/{transform_id}/cancel"
         url = self.url + path_template.format(transform_id=transform_id)
 
         async with ClientSession() as session:
             async with session.get(headers=headers, url=url) as r:
-
                 if r.status == 403:
                     raise AuthorizationError(
                         f"Not authorized to access serviceX at {self.url}"