[DM-41830] Major refactor

Pull the auth parts into utils along with other utility functions
Make catalog.py a lot shorter and break that off from siav2
Make a new service.py to hold the SIAv2 service

src/lsst/rsp/catalog.py

@@ -3,63 +3,9 @@
 from typing import Optional
 
 import pyvo
-import pyvo.auth.authsession
-import requests
 from deprecated import deprecated
 
-from .utils import get_access_token
-
-
-def _get_cutout_url() -> str:
-    return os.getenv("EXTERNAL_INSTANCE_URL", "") + "/api/cutout"
-
-
-def _get_datalink_url() -> str:
-    return os.getenv("EXTERNAL_INSTANCE_URL", "") + "/api/datalink"
-
-
-def _get_siav2_url() -> str:
-    return os.getenv("EXTERNAL_INSTANCE_URL", "") + "/api/siav2"
-
-
-def _get_tap_url(env_var: str, url: str) -> str:
-    tapurl = os.getenv("EXTERNAL_" + env_var + "_URL")
-    if not tapurl:
-        tapurl = (os.getenv("EXTERNAL_INSTANCE_URL") or "") + (
-            os.getenv(env_var + "_ROUTE") or "/api/" + url
-        )
-    return tapurl
-
-
-def _get_auth() -> Optional[pyvo.auth.authsession.AuthSession]:
-    tap_url = _get_tap_url("TAP", "tap")
-    obstap_url = _get_tap_url("OBSTAP", "obstap")
-    ssotap_url = _get_tap_url("SSOTAP", "ssotap")
-    s = requests.Session()
-    tok = get_access_token()
-    if not tok:
-        return None
-    s.headers["Authorization"] = "Bearer " + tok
-    auth = pyvo.auth.authsession.AuthSession()
-    auth.credentials.set("lsst-token", s)
-    auth.add_security_method_for_url(_get_cutout_url(), "lsst-token")
-    auth.add_security_method_for_url(_get_datalink_url(), "lsst-token")
-    auth.add_security_method_for_url(_get_siav2_url(), "lsst-token")
-    auth.add_security_method_for_url(_get_siav2_url() + "/query", "lsst-token")
-    auth.add_security_method_for_url(tap_url, "lsst-token")
-    auth.add_security_method_for_url(tap_url + "/sync", "lsst-token")
-    auth.add_security_method_for_url(tap_url + "/async", "lsst-token")
-    auth.add_security_method_for_url(tap_url + "/tables", "lsst-token")
-    auth.add_security_method_for_url(obstap_url, "lsst-token")
-    auth.add_security_method_for_url(obstap_url + "/sync", "lsst-token")
-    auth.add_security_method_for_url(obstap_url + "/async", "lsst-token")
-    auth.add_security_method_for_url(obstap_url + "/tables", "lsst-token")
-    auth.add_security_method_for_url(ssotap_url, "lsst-token")
-    auth.add_security_method_for_url(ssotap_url + "/sync", "lsst-token")
-    auth.add_security_method_for_url(ssotap_url + "/async", "lsst-token")
-    auth.add_security_method_for_url(ssotap_url + "/tables", "lsst-token")
-
-    return auth
+from .utils import get_pyvo_auth, get_service_url
 
 
 @deprecated(reason='Please use get_tap_service("tap")')
@@ -72,10 +18,6 @@ def get_obstap_service() -> pyvo.dal.TAPService:
     return get_tap_service("obstap")
 
 
-def get_siav2_service(*args: str) -> pyvo.dal.SIA2Service:
-    return SIA2Service(_get_siav2_url(), _get_auth())
-
-
 def get_tap_service(*args: str) -> pyvo.dal.TAPService:
     if len(args) == 0:
         warnings.warn(
@@ -87,12 +29,8 @@ def get_tap_service(*args: str) -> pyvo.dal.TAPService:
     else:
         database = args[0]
 
-    if database == "tap":
-        tap_url = _get_tap_url("TAP", "tap")
-    elif database == "obstap":
-        tap_url = _get_tap_url("OBSTAP", "obstap")
-    elif database == "ssotap":
-        tap_url = _get_tap_url("SSOTAP", "ssotap")
+    if database in ["tap", "obstap", "ssotap"]:
+        tap_url = get_service_url(database)
     else:
         raise Exception(database + " is not a valid tap service")
 
@@ -106,7 +44,7 @@ def get_tap_service(*args: str) -> pyvo.dal.TAPService:
     #
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", category=UserWarning)
-        ts = pyvo.dal.TAPService(tap_url, _get_auth())
+        ts = pyvo.dal.TAPService(tap_url, get_pyvo_auth())
     return ts
 
 
@@ -121,5 +59,5 @@ def retrieve_query(query_url: str) -> pyvo.dal.AsyncTAPJob:
     #
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", category=UserWarning)
-        atj = pyvo.dal.AsyncTAPJob(query_url, _get_auth())
+        atj = pyvo.dal.AsyncTAPJob(query_url, get_pyvo_auth())
     return atj

src/lsst/rsp/service.py

@@ -0,0 +1,12 @@
+import pyvo
+from pyvo.dal import SIA2Service
+
+from .utils import get_pyvo_auth, get_service_url
+
+def get_siav2_service(*args: str) -> pyvo.dal.SIA2Service:
+    if len(args) == 0:
+        ds = "dp0.2"
+    elif args == "latiss":
+        ds = "latiss"
+
+    return SIA2Service(get_service_url("siav2"), get_pyvo_auth())

src/lsst/rsp/utils.py

@@ -6,6 +6,9 @@
 from typing import Any, Optional, Union
 
 import bokeh.io
+import pyvo.auth.authsession
+import requests
+
 
 _NO_K8S = False
 
@@ -53,6 +56,51 @@ def get_hostname() -> str:
     return os.environ.get("HOSTNAME") or "localhost"
 
 
+def get_service_url(name: str, env_name=None) -> str:
+    if not env_name:
+        env_name = name.upper()
+
+    url = os.getenv(f"EXTERNAL_{env_name}_URL")
+    if url:
+        return url
+
+    fqdn = os.getenv("EXTERNAL_INSTANCE_URL") or ""
+    path = os.getenv(f"{env_name}_ROUTE") or f"/api/{name}"
+    return f"{fqdn}/{path}"
+
+
+def get_pyvo_auth() -> Optional[pyvo.auth.authsession.AuthSession]:
+    """Utility function to create a pyvo compatible auth object."""
+    tap_url = get_service_url("tap")
+    obstap_url = get_service_url("obstap")
+    ssotap_url = get_service_url("ssotap")
+    siav2_url = get_service_url("siav2")
+    s = requests.Session()
+    tok = get_access_token()
+    if not tok:
+        return None
+    s.headers["Authorization"] = "Bearer " + tok
+    auth = pyvo.auth.authsession.AuthSession()
+    auth.credentials.set("lsst-token", s)
+    auth.add_security_method_for_url(get_service_url("cutout"), "lsst-token")
+    auth.add_security_method_for_url(get_service_url("datalink"), "lsst-token")
+    auth.add_security_method_for_url(siav2_url, "lsst-token")
+    auth.add_security_method_for_url(siav2_url + "/query", "lsst-token")
+    auth.add_security_method_for_url(tap_url, "lsst-token")
+    auth.add_security_method_for_url(tap_url + "/sync", "lsst-token")
+    auth.add_security_method_for_url(tap_url + "/async", "lsst-token")
+    auth.add_security_method_for_url(tap_url + "/tables", "lsst-token")
+    auth.add_security_method_for_url(obstap_url, "lsst-token")
+    auth.add_security_method_for_url(obstap_url + "/sync", "lsst-token")
+    auth.add_security_method_for_url(obstap_url + "/async", "lsst-token")
+    auth.add_security_method_for_url(obstap_url + "/tables", "lsst-token")
+    auth.add_security_method_for_url(ssotap_url, "lsst-token")
+    auth.add_security_method_for_url(ssotap_url + "/sync", "lsst-token")
+    auth.add_security_method_for_url(ssotap_url + "/async", "lsst-token")
+    auth.add_security_method_for_url(ssotap_url + "/tables", "lsst-token")
+    return auth
+
+
 def show_with_bokeh_server(obj: Any) -> None:
     """Method to wrap bokeh with proxy URL"""