Add get_query_history

changelog.d/20250117_120000_athornton_DM_48465.md

@@ -0,0 +1,6 @@
+<!-- Delete the sections that don't apply -->
+
+### New features
+
+- added `get_query_history()` function
+

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
     "httpx<0.28",
     "structlog",  # Uses CalVer, not SemVer
     "symbolicmode<3",
+    "xmltodict"
 ]
 dynamic = ["version"]
 

src/lsst/rsp/__init__.py

@@ -5,6 +5,7 @@
 from .catalog import (
     get_catalog,
     get_obstap_service,
+    get_query_history,
     get_tap_service,
     retrieve_query,
 )
@@ -41,6 +42,7 @@
     "get_datalink_result",
     "get_digest",
     "get_node",
+    "get_query_history",
     "get_pod",
     "get_tap_service",
     "get_siav2_service",

src/lsst/rsp/catalog.py

@@ -3,8 +3,10 @@
 import warnings
 
 import pyvo
+import xmltodict
 from deprecated import deprecated
 
+from .client import RSPClient
 from .utils import get_pyvo_auth, get_service_url
 
 
@@ -66,3 +68,23 @@ def retrieve_query(query_url: str) -> pyvo.dal.AsyncTAPJob:
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", category=UserWarning)
         return pyvo.dal.AsyncTAPJob(query_url, get_pyvo_auth())
+
+
+async def get_query_history(n: int | None = None) -> list[str]:
+    """Retrieve last n query jobref ids.  If n is not specified, or n<1,
+    retrieve all query jobref ids.
+    """
+    client = RSPClient("/api/tap")
+    params = {}
+    if n and n > 0:
+        params = {"last": f"{n}"}
+    full_history_xml = await client.get("async", params=params)
+    history_dict = xmltodict.parse(full_history_xml.text)
+    joblist = history_dict["uws:jobs"]["uws:jobref"]
+    jobs: list[str] = []
+    for job in joblist:
+        for k, v in job.items():
+            if k == "@id":
+                jobs.append(v)
+                break
+    return jobs

tests/catalog_test.py

@@ -0,0 +1,48 @@
+"""Test the RSPClient."""
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from lsst.rsp import get_query_history
+
+
+@pytest.mark.usefixtures("_rsp_env")
+@pytest.mark.asyncio
+async def test_get_query_history(httpx_mock: HTTPXMock) -> None:
+    """Ensure that get_query_history() works, which in turn will ensure
+    that the RSPClient has the right headers and assembles its URL correctly.
+    """
+    httpx_mock.add_response(
+        url="https://rsp.example.com/api/tap/async",
+        match_headers={
+            "Authorization": "Bearer gf-dummytoken",
+            "Content-Type": "application/json",
+        },
+        text=(
+            """<?xml version="1.0" encoding="UTF-8"?>
+<uws:jobs xmlns:uws="http://www.ivoa.net/xml/UWS/v1.0"
+            xmlns:xlink="http://www.w3.org/1999/xlink" version="1.1">
+  <uws:jobref id="phdl67i3tmklfdbz">
+    <uws:phase>COMPLETED</uws:phase>
+    <uws:runId>dp02_dc2_catalogs.Object - data-dev</uws:runId>
+    <uws:ownerId>adam</uws:ownerId>
+    <uws:creationTime>2025-01-15T23:36:17.931Z</uws:creationTime>
+  </uws:jobref>
+  <uws:jobref id="r4qyb04xesh7mbz3">
+    <uws:phase>COMPLETED</uws:phase>
+    <uws:ownerId>adam</uws:ownerId>
+    <uws:creationTime>2024-12-05T17:49:27.518Z</uws:creationTime>
+  </uws:jobref>
+  <uws:jobref id="yk16agxjefl6gly6">
+    <uws:phase>COMPLETED</uws:phase>
+    <uws:runId>ivoa.ObsCore - data-dev</uws:runId>
+    <uws:ownerId>adam</uws:ownerId>
+    <uws:creationTime>2025-01-15T23:37:03.089Z</uws:creationTime>
+  </uws:jobref>
+</uws:jobs>"""
+        ),
+    )
+    jobs = await get_query_history()
+    assert jobs == ["phdl67i3tmklfdbz", "r4qyb04xesh7mbz3", "yk16agxjefl6gly6"]
+    # The httpx mock will throw an error at teardown if we did not exercise
+    # the mock, so we know the request matched both the URL and the headers.