Merge pull request #742 from stvoutsin/issue-708

ENH: Provide users with easy access to global variables

Author: bsipocz

CHANGES.rst

@@ -4,6 +4,9 @@
 Enhancements and Fixes
 ----------------------
 
+- Add ``pyvo.registry.get_RegTAP_service_url()`` to allow users to inspect the
+  currently configured registry endpoint. Expose other user-facing constants at their package level. [#742]
+
 - Improve handling of timeouts in tap wait and _update [#740]
 
 - Preserve text/plain error body in raise_if_error() after stream is consumed. [#736]

docs/registry/index.rst

@@ -604,6 +604,15 @@ Within a Python session, you can use the
 `pyvo.registry.choose_RegTAP_service` function, which also takes the
 TAP access URL.
 
+To inspect which endpoint is currently in use, call
+`pyvo.registry.get_RegTAP_service_url`:
+
+.. doctest::
+
+  >>> from pyvo import registry
+  >>> registry.get_RegTAP_service_url()
+  'http://reg.g-vo.org/tap'
+
 As long as you have on working registry endpoint, you can find the other
 RegTAP services using:
 

pyvo/auth/__init__.py

@@ -1,7 +1,9 @@
 # Licensed under a 3-clause BSD style license - see LICENSE.rst
 
-__all__ = ["AuthSession", "AuthURLs", "CredentialStore"]
+__all__ = ["AuthSession", "AuthURLs", "CredentialStore",
+           "ANONYMOUS", "BASIC", "CLIENT_CERTIFICATE", "COOKIE"]
 
 from .authsession import AuthSession
 from .authurls import AuthURLs
 from .credentialstore import CredentialStore
+from .securitymethods import ANONYMOUS, BASIC, CLIENT_CERTIFICATE, COOKIE

pyvo/dal/__init__.py

@@ -13,7 +13,8 @@
 from .ssa import SSAService, SSAQuery, SSAResults, SSARecord
 from .sla import SLAService, SLAQuery, SLAResults, SLARecord
 from .scs import SCSService, SCSQuery, SCSResults, SCSRecord
-from .tap import TAPService, TAPQuery, TAPResults, AsyncTAPJob
+from .tap import TAPService, TAPQuery, TAPResults, AsyncTAPJob, DEFAULT_JOB_POLL_TIMEOUT, DEFAULT_JOB_WAIT_TIMEOUT
+from .adhoc import DATALINK_BATCH_CALL_SIZE
 
 
 from .exceptions import (
@@ -31,4 +32,6 @@
     "SIARecord", "SSARecord", "SLARecord", "SCSRecord",
     "AsyncTAPJob",
     "DALAccessError", "DALProtocolError", "DALFormatError", "DALServiceError",
-    "DALQueryError", "DALOverflowWarning", "DALRateLimitError"]
+    "DALQueryError", "DALOverflowWarning", "DALRateLimitError",
+    "DEFAULT_JOB_POLL_TIMEOUT", "DEFAULT_JOB_WAIT_TIMEOUT",
+    "DATALINK_BATCH_CALL_SIZE"]

pyvo/dal/adhoc.py

@@ -83,7 +83,8 @@ def new_resource_unknown_tag(self, iterator, tag, data, config, pos):
 __all__ = [
     "AdhocServiceResultsMixin", "DatalinkResultsMixin", "DatalinkRecordMixin",
     "DatalinkService", "DatalinkQuery", "DatalinkResults", "DatalinkRecord",
-    "SodaRecordMixin", "SodaQuery"]
+    "SodaRecordMixin", "SodaQuery",
+    "DATALINK_BATCH_CALL_SIZE"]
 
 
 def _get_input_params_from_resource(resource):

pyvo/dal/tests/test_tap.py

@@ -16,9 +16,9 @@
 import requests
 import requests_mock
 
+from pyvo import dal
 from pyvo.dal.tap import escape, search, AsyncTAPJob, TAPService
 from pyvo.dal import DALQueryError, DALServiceError, DALOverflowWarning, DALRateLimitError
-
 from pyvo.io.uws import JobFile
 from pyvo.io.uws.tree import Parameter, Result, ErrorSummary, Message
 from pyvo.io.vosi.exceptions import VOSIError
@@ -1828,3 +1828,9 @@ def test_rate_limit_on_tables_endpoint(self):
             error = excinfo.value
             assert error.retry_after_seconds == 60
             assert error.code == 429
+
+
+def test_public_constants_accessible_from_dal():
+    assert isinstance(dal.DEFAULT_JOB_POLL_TIMEOUT, (int, float))
+    assert isinstance(dal.DEFAULT_JOB_WAIT_TIMEOUT, (int, float))
+    assert isinstance(dal.DATALINK_BATCH_CALL_SIZE, int)

pyvo/dam/__init__.py

@@ -1,4 +1,4 @@
 # Licensed under a 3-clause BSD style license - see LICENSE.rst
 from .obscore import *
 
-__all__ = ["ObsCoreMetadata"]
+__all__ = ["ObsCoreMetadata", "POLARIZATION_STATES", "CALIBRATION_LEVELS"]

pyvo/registry/__init__.py

@@ -7,6 +7,7 @@
 
 from .regtap import (search, ivoid2service,
     get_RegTAP_query,
+    get_RegTAP_service_url,
     choose_RegTAP_service,
     RegistryResults, RegistryResource)
 
@@ -15,7 +16,8 @@
                      UCD, UAT, Spatial, Spectral, Temporal,
                      RegTAPFeatureMissing)
 
-__all__ = ["search", "get_RegTAP_query", "Constraint", "SubqueriedConstraint",
+__all__ = ["search", "get_RegTAP_query", "get_RegTAP_service_url",
+           "Constraint", "SubqueriedConstraint",
            "Freetext", "Author",
            "Servicetype", "Waveband", "Datamodel", "Ivoid", "UCD",
            "UAT", "Spatial", "Spectral", "Temporal",

pyvo/registry/regtap.py

@@ -34,7 +34,7 @@
 from ..utils.formatting import para_format_desc
 
 
-__all__ = ["search", "get_RegTAP_query", "Interface",
+__all__ = ["search", "get_RegTAP_query", "get_RegTAP_service_url", "Interface",
            "RegistryResource", "RegistryResults", "ivoid2service"]
 
 REGISTRY_BASEURL = os.environ.get("IVOA_REGISTRY", "http://reg.g-vo.org/tap"
@@ -111,6 +111,22 @@ def get_RegTAP_service():
     return tap.TAPService(REGISTRY_BASEURL)
 
 
+def get_RegTAP_service_url():
+    """
+    Return the access URL of the currently configured RegTAP service.
+
+    By default, pyVO uses whatever is given in the environment variable
+    ``IVOA_REGISTRY``, defaulting to GAVO's TAP service.  To change this
+    use :py:func:`pyvo.registry.choose_RegTAP_service`.
+
+    Returns
+    -------
+    str
+        The TAP access URL of the current RegTAP endpoint.
+    """
+    return REGISTRY_BASEURL
+
+
 def choose_RegTAP_service(access_url):
     """
     changes the RegTAP service used by :py:func:`search`

pyvo/registry/tests/test_regtap.py

@@ -930,6 +930,20 @@ def test_sia2_service_operation():
     assert "s_dec" in res.to_table().columns
 
 
+def test_get_RegTAP_service_url_default():
+    assert regtap.get_RegTAP_service_url() == regtap.REGISTRY_BASEURL
+
+
+def test_get_RegTAP_service_url_reflects_changes():
+    alt_svc = "https://example.com/tap"
+    previous_url = regtap.REGISTRY_BASEURL
+    try:
+        regtap.choose_RegTAP_service(alt_svc)
+        assert regtap.get_RegTAP_service_url() == alt_svc
+    finally:
+        regtap.choose_RegTAP_service(previous_url)
+
+
 @pytest.mark.remote_data
 def test_endpoint_switching():
     alt_svc = "https://mast.stsci.edu/vo-tap/api/v0.1/registry"