diff --git a/src/lsst/rsp/catalog.py b/src/lsst/rsp/catalog.py index b592d44..13de5d4 100644 --- a/src/lsst/rsp/catalog.py +++ b/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 diff --git a/src/lsst/rsp/service.py b/src/lsst/rsp/service.py new file mode 100644 index 0000000..348b6c2 --- /dev/null +++ b/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()) diff --git a/src/lsst/rsp/utils.py b/src/lsst/rsp/utils.py index 3ffca45..591300e 100644 --- a/src/lsst/rsp/utils.py +++ b/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"""