ServiceX DID finder

.github/workflows/deploy-config.json

@@ -29,6 +29,11 @@
     "image_name": "servicex-did-finder-xrootd",
     "test_required": true
   },
+  {
+    "dir_name": "did_finder_servicex",
+    "image_name": "servicex-did-finder-servicex",
+    "test_required": true
+  },
   {
     "dir_name": "code_generator_funcadl_uproot",
     "image_name": "servicex_code_gen_func_adl_uproot",

did_finder_servicex/Dockerfile

@@ -0,0 +1,44 @@
+FROM python:3.13-trixie
+
+LABEL maintainer="Peter Onyisi <ponyisi@utexas.edu>"
+
+# Create app directory
+WORKDIR /opt/servicex
+
+# Create celery user. Assign them to group zero since that is the group OpenShift will run the container as
+RUN useradd -g 0 -ms /bin/bash celery
+
+ENV  POETRY_VERSION=2.4.3
+
+RUN apt-get update && \
+    apt-get install cmake osg-ca-certs --no-install-recommends --assume-yes
+
+RUN pip install --no-cache-dir poetry==$POETRY_VERSION
+RUN mkdir -p /opt/servicex/pypoetry
+COPY pyproject.toml pyproject.toml
+COPY poetry.lock poetry.lock
+
+ENV XDG_CONFIG_HOME=/opt/servicex
+RUN poetry config virtualenvs.in-project true
+RUN poetry install --no-root --no-interaction --no-ansi
+
+# Bring over the main scripts
+COPY . .
+
+# Change ownership of the app directory to celery user. Also set the group to zero since
+# that is the group OpenShift will run the container as
+RUN chown -R celery:0 /opt/servicex
+RUN chmod -R g=u /opt/servicex
+
+# Switch to celery user
+USER celery
+
+# Make sure python isn't buffered
+ENV PYTHONUNBUFFERED=1
+
+ENV PYTHONPATH=/opt/servicex/src
+
+ENV BROKER_URL="amqp://guest:guest@localhost:5672//"
+
+
+ENTRYPOINT [ "scripts/start_celery_worker.sh"]

did_finder_servicex/README.md

@@ -0,0 +1,5 @@
+# ServiceX_DID_finder_ServiceX
+Use a ServiceX transformation ID to look up the outputs of that transform.
+
+## Finding files
+Provide a ServiceX transform ID. The code will look up URLs for accessing the results of that transformation, assuming they are still available.

did_finder_servicex/pyproject.toml

@@ -0,0 +1,28 @@
+[tool.poetry]
+name = "did-finder-servicex"
+version = "0.1.0"
+description = ""
+authors = ["Peter Onyisi <ponyisi@utexas.edu>"]
+readme = "README.md"
+packages = [{include = "servicex_did_finder_servicex", from="src"}]
+
+[tool.poetry.dependencies]
+python = ">=3.10,<4.0"
+servicex-did-finder-lib = "^3.2.0-rc1"
+httpx = ">=0.24"
+httpx_retries = ">=0.3.2"
+
+[tool.poetry.group.test]
+optional = true
+
+[tool.poetry.group.test.dependencies]
+pytest = ">=7.1.3"
+flake8 = ">=5.0.4"
+pytest-asyncio = ">=0.19.0"
+coverage = ">=6.5.0"
+pytest-mock = "^3.10.0"
+respx = "^0.23.1"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

did_finder_servicex/scripts/start_celery_worker.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+poetry run celery --broker="$BROKER_URL" -A servicex_did_finder_servicex worker \
+                  --loglevel=info -Q did_finder_servicex \
+                  --concurrency=1 --hostname=did_finder_servicex@%h

did_finder_servicex/src/servicex_did_finder_servicex/__init__.py

@@ -0,0 +1,27 @@
+# Copyright (c) 2026, 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.

did_finder_servicex/src/servicex_did_finder_servicex/celery.py

@@ -0,0 +1,128 @@
+# Copyright (c) 2026, 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 os
+from servicex_did_finder_lib.logstash_logging import initialize_logging
+from typing import Any, Dict, Generator
+
+from httpx import Client, Timeout
+from httpx_retries import RetryTransport, Retry
+
+from servicex_did_finder_lib import DIDFinderApp
+from servicex_did_finder_lib.exceptions import (
+    NoSuchDatasetException,
+    LookupFailureException,
+)
+
+__log = initialize_logging(component_name="servicex_did_finder")
+
+cache_prefix = os.environ.get("CACHE_PREFIX", "")
+servicex_endpoint = os.environ["SERVICEX_ENDPOINT"]
+
+app = DIDFinderApp("servicex")
+
+
+@app.did_lookup_task(name="did_finder_servicex.lookup_dataset")
+def lookup_dataset(self, did: str, dataset_id: int, endpoint: str) -> None:
+    self.do_lookup(
+        did=did, dataset_id=dataset_id, endpoint=endpoint, user_did_finder=find_files
+    )
+
+
+def find_files(
+    did_name: str, info: Dict[str, Any], did_finder_args: Dict | None = None
+) -> Generator[Dict[str, Any], None, None]:
+    """For each incoming ServiceX transform ID, get the URLs
+
+    Notes:
+
+    Args:
+        did_name (str): Dataset name
+        info (Dict[str, Any]): Information bag, mainly has the `request-id` which is
+                               used to track error mesages accross logs.
+        did_finder_args (dict): Additional arguments passed to the finder
+
+    Returns:
+        Generator[Dict[str, any], None]: yield each file
+    """
+    __log.info(
+        "DID Lookup request received.",
+        extra={"dataset_id": info["dataset-id"], "dataset_name": did_name},
+    )
+
+    try:
+        retry_options = Retry(total=3, backoff_factor=10)
+        with Client(
+            transport=RetryTransport(retry=retry_options),
+            timeout=Timeout(10, read=300),
+        ) as session:
+            service_url = (
+                servicex_endpoint
+                + f"/servicex/internal/transformation/{did_name}/results"
+            )
+            r = session.get(url=service_url)
+            if r.status_code != 200:
+                raise LookupFailureException(
+                    f"Failure searching for {did_name}: {r.status_code}"
+                )
+            transform_data = r.json()["results"]
+            file_list = [
+                (_["s3-object-name"], _["total-bytes"])
+                for _ in transform_data
+                if _["transform_status"] == "success"
+            ]
+
+            service_url = (
+                servicex_endpoint + "/servicex/internal/transformation/file-urls"
+            )
+            r = session.post(
+                service_url,
+                json={"request_id": did_name, "file_list": [_[0] for _ in file_list]},
+            )
+            if r.status_code == 404:
+                raise NoSuchDatasetException(f"Results for {did_name} not found")
+
+            url_data = r.json()["uris"]
+            res = []
+            for f, size in file_list:
+                _file = (
+                    url_data[f][0],
+                    size,
+                )
+                res.append(_file)
+    except NoSuchDatasetException as e:
+        raise e
+    except Exception as e:
+        raise LookupFailureException(f"Failure searching for {did_name}: {e}")
+
+    for data in res:
+        yield {
+            "paths": [data[0]],
+            "adler32": 0,  # No clue
+            "file_size": data[1],  # We could look up the size but that would be slow
+            "file_events": 0,  # And this we do not know
+        }