add supported server resources checks

servicex/query_core.py

@@ -560,23 +560,32 @@ async def get_signed_url(
             if progress:
                 progress.advance(task_id=download_progress, task_type="Download")
 
+        transformation_results_enabled = (
+            "transformationresults" in await self.servicex.get_resources()
+        )
+
         while True:
             if not cached_record:
                 await asyncio.sleep(self.minio_polling_interval)
             if self.minio:
                 # if self.minio exists, self.current_status will too
                 if self.current_status.files_completed > len(files_seen):
-                    new_begin_at = datetime.datetime.now(tz=datetime.timezone.utc)
-                    files = await self.servicex.get_transformation_results(
-                        self.current_status.request_id, begin_at
-                    )
-                    begin_at = new_begin_at
+                    if transformation_results_enabled:
+                        new_begin_at = datetime.datetime.now(tz=datetime.timezone.utc)
+                        files = await self.servicex.get_transformation_results(
+                            self.current_status.request_id, begin_at
+                        )
+                        begin_at = new_begin_at
+                    else:
+                        files = await self.minio.list_bucket()
 
                     for file in files:
-                        if "file-path" not in file:
-                            continue
-
-                        file_path = file["file-path"].replace("/", ":")
+                        if transformation_results_enabled:
+                            if "file-path" not in file:
+                                continue
+                            file_path = file.get("file-path", "").replace("/", ":")
+                        else:
+                            file_path = file.filename
 
                         if file_path not in files_seen:
                             if signed_urls_only:

servicex/servicex_adapter.py

@@ -28,7 +28,9 @@
 import os
 import time
 import datetime
-from typing import Optional, Dict, List
+import asyncio
+from typing import Optional, Dict, List, Any, TypeVar, Callable, cast
+from functools import wraps
 
 from aiohttp import ClientSession
 import httpx
@@ -44,6 +46,85 @@
 
 from servicex.models import TransformRequest, TransformStatus, CachedDataset
 
+T = TypeVar("T")
+
+
+def requires_resource(
+    resource_name: str,
+) -> Callable[[Callable[..., T]], Callable[..., T]]:
+    """
+    Decorator to check if a specific API resource is available on the server.
+
+    Args:
+        resource_name: The name of the resource that needs to be available
+
+    Returns:
+        A decorator function that wraps around class methods
+
+    Raises:
+        ResourceNotAvailableError: If the required resource is not available on the server
+    """
+
+    def decorator(func: Callable[..., T]) -> Callable[..., T]:
+        # Determine if function is async at decoration time (not runtime)
+        is_async = asyncio.iscoroutinefunction(func)
+        func_name = func.__name__
+
+        # Class-level cache for sync method resources
+        sync_cache_key = f"_sync_resources_for_{resource_name}"
+
+        if is_async:
+
+            @wraps(func)
+            async def async_wrapper(self, *args: Any, **kwargs: Any) -> T:
+                # Get resources and check availability in one operation
+                if resource_name not in await self.get_resources():
+                    raise ResourceNotAvailableError(
+                        f"Resource '{resource_name}' required for '{func_name}' is unavailable"
+                    )
+                return await func(self, *args, **kwargs)
+
+            return cast(Callable[..., T], async_wrapper)
+        else:
+
+            @wraps(func)
+            def sync_wrapper(self, *args: Any, **kwargs: Any) -> T:
+                # Initialize class-level cache attributes if needed
+                cls = self.__class__
+                if not hasattr(cls, sync_cache_key):
+                    setattr(cls, sync_cache_key, (None, 0))  # (resources, timestamp)
+
+                cache_ttl = getattr(self, "_resources_cache_ttl", 300)
+                cached_resources, timestamp = getattr(cls, sync_cache_key)
+                current_time = time.time()
+
+                # Check if cache needs refresh
+                if cached_resources is None or (current_time - timestamp) >= cache_ttl:
+                    loop = asyncio.new_event_loop()
+                    try:
+                        cached_resources = loop.run_until_complete(self.get_resources())
+                        setattr(cls, sync_cache_key, (cached_resources, current_time))
+                    finally:
+                        loop.close()
+
+                # Check resource availability
+                if resource_name not in cached_resources:
+                    raise ResourceNotAvailableError(
+                        f"Resource '{resource_name}' required for '{func_name}' is unavailable"
+                    )
+
+                return func(self, *args, **kwargs)
+
+            return cast(Callable[..., T], sync_wrapper)
+
+    return decorator
+
+
+class ResourceNotAvailableError(Exception):
+    """Exception raised when a required resource is not available on the server."""
+
+    pass
+
 
 class AuthorizationError(BaseException):
     pass
@@ -64,6 +145,47 @@ def __init__(self, url: str, refresh_token: Optional[str] = None):
         self.refresh_token = refresh_token
         self.token = None
 
+        self._available_resources: Optional[Dict[str, Any]] = None
+        self._resources_last_updated: Optional[float] = None
+        self._resources_cache_ttl = 60 * 5
+
+    async def get_resources(self) -> Dict[str, Any]:
+        """
+        Fetches the list of available resources from the server.
+        Caches the result for 5 minutes to avoid excessive API calls.
+
+        Returns:
+            A dictionary of available resources with their properties
+        """
+        current_time = time.time()
+
+        # Return cached resources if they exist and are not expired
+        if (
+            self._available_resources is not None
+            and self._resources_last_updated is not None
+            and current_time - self._resources_last_updated < self._resources_cache_ttl
+        ):
+            return self._available_resources
+
+        # Fetch resources from server
+        headers = await self._get_authorization()
+        async with ClientSession() as session:
+            async with session.get(
+                headers=headers, url=f"{self.url}/servicex/resources"
+            ) as r:
+                if r.status == 403:
+                    raise AuthorizationError(
+                        f"Not authorized to access serviceX at {self.url}"
+                    )
+                elif r.status != 200:
+                    msg = await _extract_message(r)
+                    raise RuntimeError(f"Failed to get resources: {r.status} - {msg}")
+
+                self._available_resources = await r.json()
+                self._resources_last_updated = current_time
+
+                return self._available_resources
+
     async def _get_token(self):
         url = f"{self.url}/token/refresh"
         headers = {"Authorization": f"Bearer {self.refresh_token}"}
@@ -229,6 +351,7 @@ async def delete_transform(self, transform_id=None):
                         f"Failed to delete transform {transform_id} - {msg}"
                     )
 
+    @requires_resource("transformationresults")
     async def get_transformation_results(
         self, request_id: str, begin_at: datetime.datetime
     ):

tests/test_dataset.py

@@ -136,6 +136,10 @@ async def test_download_files(python_dataset):
 
     minio_mock.download_file.return_value = Path("/path/to/downloaded_file")
     minio_mock.get_signed_url.return_value = Path("http://example.com/signed_url")
+    minio_mock.list_bucket.return_value = [
+        Mock(filename="file1.txt"),
+        Mock(filename="file2.txt"),
+    ]
 
     progress_mock = Mock()
     python_dataset.minio_polling_interval = 0
@@ -161,6 +165,10 @@ async def test_download_files_with_signed_urls(python_dataset):
     python_dataset.configuration = config
     minio_mock.download_file.return_value = "/path/to/downloaded_file"
     minio_mock.get_signed_url.return_value = "http://example.com/signed_url"
+    minio_mock.list_bucket.return_value = [
+        Mock(filename="file1.txt"),
+        Mock(filename="file2.txt"),
+    ]
     progress_mock = Mock()
 
     python_dataset.servicex = AsyncMock()

tests/test_servicex_dataset.py

@@ -226,6 +226,7 @@ async def test_submit(mocker):
     ]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1, file2]])
     mock_minio.download_file = AsyncMock(
         side_effect=lambda a, _, shorten_filename: PurePath(a)
     )
@@ -278,6 +279,7 @@ async def test_submit_partial_success(mocker):
     ]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.download_file = AsyncMock(
         side_effect=lambda a, _, shorten_filename: PurePath(a)
     )
@@ -323,6 +325,7 @@ async def test_use_of_cache(mocker):
         transform_status3,
     ]
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(return_value=[file1, file2])
     mock_minio.download_file = AsyncMock(
         side_effect=lambda a, _, shorten_filename: PurePath(a)
     )
@@ -357,6 +360,7 @@ async def test_use_of_cache(mocker):
         # second round, should hit the cache (and not call the sx_adapter, minio, or update_record)
         with ExpandableProgress(display_progress=False) as progress:
             servicex2 = AsyncMock()
+            mock_minio.list_bucket.reset_mock()
             servicex.get_transformation_results.reset_mock()
             mock_minio.get_signed_url.reset_mock()
             datasource2 = Query(
@@ -375,6 +379,7 @@ async def test_use_of_cache(mocker):
                 signed_urls_only=True, expandable_progress=progress
             )
             servicex2.assert_not_awaited()
+            mock_minio.list_bucket.assert_not_awaited()
             servicex.get_transformation_results.assert_not_awaited()
             mock_minio.get_signed_url.assert_not_awaited()
         upd.assert_not_called()
@@ -399,6 +404,7 @@ async def test_use_of_cache(mocker):
                 signed_urls_only=False, expandable_progress=progress
             )
         servicex.assert_not_awaited()
+        mock_minio.list_bucket.assert_not_awaited()
         servicex.get_transformation_results.assert_not_awaited()
         mock_minio.download_file.assert_not_awaited()
         upd.assert_called_once()
@@ -417,6 +423,7 @@ async def test_submit_cancel(mocker):
     ]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.download_file = AsyncMock(
         side_effect=lambda a, _, shorten_filename: PurePath(a)
     )
@@ -457,6 +464,7 @@ async def test_submit_fatal(mocker):
     ]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.download_file = AsyncMock(
         side_effect=lambda a, _, shorten_filename: PurePath(a)
     )
@@ -501,6 +509,7 @@ async def test_submit_generic(mocker, codegen_list):
     ]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.download_file = AsyncMock()
 
     mock_cache = mocker.MagicMock(QueryCache)
@@ -549,6 +558,7 @@ async def test_submit_cancelled(mocker, codegen_list):
     sx.get_transform_status.side_effect = [transform_status4]
 
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.download_file = AsyncMock()
 
     mock_cache = mocker.MagicMock(QueryCache)
@@ -623,6 +633,7 @@ async def test_use_of_ignore_cache(mocker, servicex):
     )
     # Prepare Minio
     mock_minio = AsyncMock()
+    mock_minio.list_bucket = AsyncMock(side_effect=[[file1], [file1]])
     mock_minio.get_signed_url = AsyncMock(side_effect=["http://file1", "http://file2"])
     mocker.patch("servicex.minio_adapter.MinioAdapter", return_value=mock_minio)
     did = FileListDataset("/foo/bar/baz.root")
@@ -698,6 +709,7 @@ async def test_use_of_ignore_cache(mocker, servicex):
             res = await datasource_without_ignore_cache.submit_and_download(
                 signed_urls_only=True, expandable_progress=progress
             )  # noqa
+        mock_minio.list_bucket.assert_not_awaited()
         servicex.get_transformation_results.assert_not_awaited()
         mock_minio.download_file.assert_not_awaited()
         assert len(res.signed_url_list) == 2