Merge branch 'master' into 474_non_json_errors

Author: ponyisi

docs/command_line.rst

@@ -76,3 +76,14 @@ clear
 Clear all of the transforms from the cache. Add ``-y`` to force the
 operation without confirming with the console.
 
+datasets
+~~~~~~~~
+
+These commands interact with datasets cached on the server
+
+list
+^^^^
+List all of the datasets cached on the server. Accepts a command line argument
+of ``--did-finder`` to filter the list of datasets by a specific DID finder such
+as ``rucio`` or ``user``.
+

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "servicex"
-version = "3.0.0-alpha.100"
+version = "3.0.1-alpha.1"
 description = "Python SDK and CLI Client for ServiceX"
 readme = "README.md"
 license = { text = "BSD-3-Clause" }  # SPDX short identifier

servicex/app/datasets.py

@@ -0,0 +1,79 @@
+# Copyright (c) 2024, IRIS-HEP
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# * Redistributions of source code must retain the above copyright notice, this
+#   list of conditions and the following disclaimer.
+#
+# * Redistributions in binary form must reproduce the above copyright notice,
+#   this list of conditions and the following disclaimer in the documentation
+#   and/or other materials provided with the distribution.
+#
+# * Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+import asyncio
+from typing import Optional
+
+import rich
+
+from servicex.app.cli_options import url_cli_option, backend_cli_option
+
+import typer
+
+from servicex.servicex_client import ServiceXClient
+from rich.table import Table
+
+datasets_app = typer.Typer(name="datasets", no_args_is_help=True)
+
+
+@datasets_app.command(no_args_is_help=True)
+def list(
+        url: Optional[str] = url_cli_option,
+        backend: Optional[str] = backend_cli_option,
+        did_finder: Optional[str] = typer.Option(
+            None,
+            help="Filter datasets by DID finder. Some useful values are 'rucio' or 'user'",
+            show_default=False,
+        ),
+):
+    """
+    List the datasets.
+    """
+    sx = ServiceXClient(url=url, backend=backend)
+    table = Table(title="ServiceX Datasets")
+    table.add_column("ID")
+    table.add_column("Name")
+    table.add_column("Files")
+    table.add_column("Size")
+    table.add_column("Status")
+    table.add_column("Created")
+    datasets = asyncio.run(sx.get_datasets(did_finder=did_finder))
+    for d in datasets:
+        # Format the CachedDataset object into a table row
+        # The last_updated field is what we should be displaying, but that is
+        # currently set to 1970-00-00 in the server and is never updated.
+        # Stick with the last_used field until
+        # https://github.com/ssl-hep/ServiceX/issues/906 is resolved
+        table.add_row(
+            str(d.id),
+            d.name if d.did_finder != "user" else "File list",
+            "%d" % d.n_files,
+            "{:,}MB".format(round(d.size / 1e6)),
+            d.lookup_status,
+            d.last_used.strftime('%Y-%m-%dT%H:%M:%S'),
+        )
+    rich.print(table)

servicex/app/main.py

@@ -31,6 +31,7 @@
 import typer
 
 from servicex._version import __version__
+from servicex.app.datasets import datasets_app
 from servicex.app.transforms import transforms_app
 from servicex.app.cache import cache_app
 from servicex.app.codegen import codegen_app
@@ -40,6 +41,7 @@
 app.add_typer(transforms_app)
 app.add_typer(cache_app)
 app.add_typer(codegen_app)
+app.add_typer(datasets_app)
 
 
 def show_version(show: bool):

servicex/expandable_progress.py

@@ -193,6 +193,10 @@ def advance(self, task_id, task_type):
         elif self.display_progress and not self.overall_progress:
             self.progress.advance(task_id=task_id)
 
+    def refresh(self):
+        if self.progress is not None:
+            self.progress.refresh()
+
 
 class TranformStatusProgress(Progress):
     def get_renderables(self):

servicex/models.py

@@ -170,3 +170,18 @@ class TransformedResults(BaseModel):
     files: int
     result_format: ResultFormat
     log_url: Optional[str] = None
+
+
+class CachedDataset(BaseModel):
+    """
+    Model for a cached dataset held by ServiceX server
+    """
+    id: int
+    name: str
+    did_finder: str
+    n_files: int
+    size: int
+    events: int
+    last_used: datetime
+    last_updated: datetime
+    lookup_status: str

servicex/query_core.py

@@ -79,6 +79,7 @@ def __init__(
         result_format: ResultFormat = ResultFormat.parquet,
         ignore_cache: bool = False,
         query_string_generator: Optional[QueryStringGenerator] = None,
+        fail_if_incomplete: bool = True,
     ):
         r"""
         This is the main class for constructing transform requests and receiving the
@@ -96,6 +97,7 @@ def __init__(
                                         for new files?
         :param result_format:
         :param ignore_cache:  If true, ignore the cache and always submit a new transform
+        :param fail_if_incomplete: If true, raise an exception if we don't have 100% completion
         """
         self.servicex = sx_adapter
         self.configuration = config
@@ -116,6 +118,7 @@ def __init__(
 
         self.request_id = None
         self.ignore_cache = ignore_cache
+        self.fail_if_incomplete = fail_if_incomplete
         self.query_string_generator = query_string_generator
 
         # Number of seconds in between ServiceX status polls
@@ -191,6 +194,7 @@ def transform_complete(task: Task):
             :param task:
             :return:
             """
+            expandable_progress.refresh()
             if task.exception():
                 logger.error(
                     f"ServiceX Exception for request ID {self.request_id} ({self.title})\"",
@@ -210,11 +214,16 @@ def transform_complete(task: Task):
                     self.cache.delete_record_by_request_id(self.request_id)
                     titlestr = (f'"{self.current_status.title}" '
                                 if self.current_status.title is not None else '')
-                    logger.error(
+                    errorstr = (
                         f"Transform {titlestr}completed with failures: "
                         f"{self.current_status.files_failed}/"
                         f"{self.current_status.files} files failed. Will not cache."
                     )
+                    failedfiles = (self.servicex.url + '/transformation-request/'
+                                   + f'/{self.request_id}/results?status=failure')
+                    errorstr2 = ("A list of failed files is at [bold red on white]"
+                                 f"[link={failedfiles}]this link[/link][/bold red on white]")
+                    logger.error(errorstr2)
                     logger.error(
                         f"Transform Request id: {self.current_status.request_id}"
                     )
@@ -225,6 +234,8 @@ def transform_complete(task: Task):
                                                           LogLevel.error,
                                                           TimeFrame.month)
                         logger.error(f"More information of '{self.title}' [bold red on white][link={kibana_link}]HERE[/link][/bold red on white]")  # NOQA: E501
+                    if self.fail_if_incomplete:
+                        raise ServiceXException(errorstr)
                 else:
                     logger.info("Transforms completed successfully")
             else:  # pragma: no cover
@@ -234,7 +245,8 @@ def transform_complete(task: Task):
         sx_request_hash = sx_request.compute_hash()
 
         # Invalidate the cache if the hash already present but if the user ignores cache
-        if self.ignore_cache and self.cache.contains_hash(sx_request_hash):
+        if self.ignore_cache and (self.cache.contains_hash(sx_request_hash)
+                                  or self.cache.is_transform_request_submitted(sx_request_hash)):
             self.cache.delete_record_by_hash(sx_request_hash)
 
         # Let's see if this is in the cache already, but respect the user's wishes

servicex/servicex_adapter.py

@@ -35,7 +35,7 @@
 from google.auth import jwt
 from tenacity import AsyncRetrying, stop_after_attempt, wait_fixed, retry_if_not_exception_type
 
-from servicex.models import TransformRequest, TransformStatus
+from servicex.models import TransformRequest, TransformStatus, CachedDataset
 
 
 class AuthorizationError(BaseException):
@@ -123,6 +123,22 @@ def get_code_generators(self):
                     f"Not authorized to access serviceX at {self.url}")
             return r.json()
 
+    async def get_datasets(self, did_finder=None) -> List[CachedDataset]:
+        headers = await self._get_authorization()
+
+        with httpx.Client() as client:
+            params = {"did-finder": did_finder} if did_finder else {}
+            r = client.get(headers=headers,
+                           url=f"{self.url}/servicex/datasets",
+                           params=params)
+
+            if r.status_code == 403:
+                raise AuthorizationError(
+                    f"Not authorized to access serviceX at {self.url}")
+
+            datasets = [CachedDataset(**d) for d in r.json()['datasets']]
+            return datasets
+
     async def submit_transform(self, transform_request: TransformRequest) -> str:
         headers = await self._get_authorization()
         retry_options = ExponentialRetry(attempts=3, start_timeout=30)

servicex/servicex_client.py

@@ -137,7 +137,7 @@ def _load_ServiceXSpec(
     return config
 
 
-def _build_datasets(config, config_path, servicex_name):
+def _build_datasets(config, config_path, servicex_name, fail_if_incomplete):
     def get_codegen(_sample: Sample, _general: General):
         if _sample.Codegen is not None:
             return _sample.Codegen
@@ -158,6 +158,7 @@ def get_codegen(_sample: Sample, _general: General):
             result_format=config.General.OutputFormat.to_ResultFormat(),
             ignore_cache=sample.IgnoreLocalCache,
             query=sample.Query,
+            fail_if_incomplete=fail_if_incomplete
         )
         logger.debug(f"Query string: {query.generate_selection_string()}")
         query.ignore_cache = sample.IgnoreLocalCache
@@ -197,11 +198,12 @@ def deliver(
     config: Union[ServiceXSpec, Mapping[str, Any], str, Path],
     config_path: Optional[str] = None,
     servicex_name: Optional[str] = None,
-    return_exceptions: bool = True
+    return_exceptions: bool = True,
+    fail_if_incomplete: bool = True
 ):
     config = _load_ServiceXSpec(config)
 
-    datasets = _build_datasets(config, config_path, servicex_name)
+    datasets = _build_datasets(config, config_path, servicex_name, fail_if_incomplete)
 
     group = DatasetGroup(datasets)
 
@@ -278,6 +280,13 @@ async def get_transform_status_async(self, transform_id) -> TransformStatus:
 
     get_transform_status = make_sync(get_transform_status_async)
 
+    def get_datasets(self, did_finder=None):
+        r"""
+        Retrieve all datasets you have run on the server
+        :return: List of Query objects
+        """
+        return self.servicex.get_datasets(did_finder)
+
     def get_code_generators(self, backend=None):
         r"""
         Retrieve the code generators deployed with the serviceX instance
@@ -302,6 +311,7 @@ def generic_query(
         title: str = "ServiceX Client",
         result_format: ResultFormat = ResultFormat.parquet,
         ignore_cache: bool = False,
+        fail_if_incomplete: bool = True,
     ) -> Query:
         r"""
         Generate a Query object for a generic codegen specification
@@ -345,6 +355,7 @@ def generic_query(
             query_cache=self.query_cache,
             result_format=result_format,
             ignore_cache=ignore_cache,
-            query_string_generator=query
+            query_string_generator=query,
+            fail_if_incomplete=fail_if_incomplete
         )
         return qobj

tests/test_expandable_progress.py

@@ -61,6 +61,14 @@ def test_overall_progress_mock(mock_progress):
     assert mock_progress.return_value.stop.call_count == 1
 
 
+@patch("servicex.expandable_progress.TranformStatusProgress",
+       return_value=MagicMock(TranformStatusProgress))
+def test_refresh_mock(mock_progress):
+    with ExpandableProgress(overall_progress=True) as progress:
+        progress.refresh()
+        mock_progress.return_value.refresh.assert_called_once()
+
+
 def test_provided_progress(mocker):
     class MockedProgress(TranformStatusProgress):
         def __init__(self):