Add prefer_async param to TAPService constructor to control what mode search uses

Author: stvoutsin

CHANGES.rst

@@ -8,6 +8,9 @@ Enhancements and Fixes
 
 - Fix job result handling to prioritize standard URL structure [#684]
 
+- Add prefer_async param to TAPService constructor to control what mode uses [#686]
+
+
 Deprecations and Removals
 -------------------------
 

docs/dal/index.rst

@@ -222,6 +222,16 @@ other channels. Most commonly, you will discover them in the VO
 registry (cf. :ref:`pyvo.registry<pyvo-registry>`).
 
 To perform a query using ADQL, the ``search()`` method is used.
+For services that work better with asynchronous queries, you can configure
+the service to prefer async execution by default:
+
+.. doctest-remote-data::
+
+    >>> tap_service = vo.dal.TAPService("http://dc.g-vo.org/tap", prefer_async=True)
+
+This will make the ``search()`` method use asynchronous execution automatically,
+while still allowing you to force synchronous execution when needed.
+
 TAPService instances have several methods to inspect the metadata
 of the service - in particular, what tables with what columns are
 available - discussed below.
@@ -250,9 +260,43 @@ robust of long-running queries.  It also supports queuing queries,
 which allows service operators to be a lot more generous with
 resource limits.
 
-To specify the query mode, you can use either ``run_sync()`` for
+There are a few ways to control the query mode:
+
+**Explicit method calls:**
+To explicitly specify the query mode, you can use either ``run_sync()`` for
 synchronous query or ``run_async()`` for asynchronous query.
 
+**Service-level preference:**
+Alternatively, you can configure whether you want the service to prefer async execution 
+by passing the ``prefer_async`` parameter to the service constructor.
+Then the ``search()`` method will use async execution by default, but
+you can still force sync execution when needed by passing the
+``force_sync`` parameter to ``search()``.
+
+.. doctest-remote-data::
+
+    >>> # Service configured to prefer async
+    >>> service = vo.dal.TAPService("http://dc.g-vo.org/tap", prefer_async=True)
+    >>> result = service.search(ex_query)  # Uses async automatically
+    >>> # Force sync execution when needed
+    >>> result = service.search(ex_query, force_sync=True)  # Uses sync
+
+**Default behavior:**
+By default, ``search()`` will use synchronous execution:
+
+.. doctest-remote-data::
+
+    >>> # Default service (prefer_async=False)
+    >>> service = vo.dal.TAPService("http://dc.g-vo.org/tap")
+    >>> result = service.search(ex_query)  # Always uses sync
+
+**Running as AsyncTAPJobs:**
+The methods above (``search()`` with ``prefer_async=True`` and ``run_async()``)
+handle job lifecycle automatically. However, you may also want to work with
+job objects directly, which can give more control over the query execution
+and job information (for example some TAP services may display progress
+information in the job output).
+
 .. doctest-remote-data::
 
     >>> job = tap_service.submit_job(ex_query)
@@ -336,6 +380,19 @@ With ``run_async()`` you basically submit an asynchronous query and
 return its result. It is like running ``submit_job()`` first and then
 run the query manually.
 
+Choosing the right execution mode
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+**Use synchronous queries when:**
+- Running quick queries (typically < 1 minute)
+- The service has efficient sync implementation
+- Query results are relatively small
+
+**Use asynchronous queries when:**
+- Running long-running queries where query might hit sync timeout limits
+- Working with services that have more robust async execution
+- You want access to (UWS) jobs you have run and metadata or results for any of your async jobs
+
 Query limit
 ^^^^^^^^^^^
 

docs/io/uws.rst

@@ -382,7 +382,7 @@ While the examples above focus on individual jobs, you can also parse job lists:
     >>> # Iterate through jobs (each is a JobSummary object)
     >>> for job in jobs:  # doctest: +ELLIPSIS
     ...     print(f"Job {job.jobid}: {job.phase}")
-    Job ...: COMPLETED
+    Job ...: ...
 
 
 Error Handling

pyvo/dal/tap.py

@@ -107,7 +107,8 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
     _tables = None
     _examples = None
 
-    def __init__(self, baseurl, *, capability_description=None, session=None):
+    def __init__(self, baseurl, *, capability_description=None,
+                 session=None, prefer_async=False):
         """
         instantiate a Table Access Protocol service
 
@@ -117,6 +118,12 @@ def __init__(self, baseurl, *, capability_description=None, session=None):
            the base URL that should be used for forming queries to the service.
         session : object
            optional session to use for network requests
+        capability_description : str, optional
+              a capability description URL to use instead of the service
+              capabilities.
+        prefer_async : bool, optional
+            Whether to prefer async queries over sync queries. (Default:
+            false)
         """
         try:
             super().__init__(baseurl, session=session, capability_description=capability_description)
@@ -131,6 +138,8 @@ def __init__(self, baseurl, *, capability_description=None, session=None):
             raise DALServiceError(f"Cannot find TAP service at '"
                                   f"{baseurl}'.\n\n{str(e)}") from None
 
+        self._prefer_async = prefer_async
+
     def get_tap_capability(self):
         """
         returns the (first) TAP capability of this service.
@@ -281,8 +290,55 @@ def run_sync(
             query, language=language, maxrec=maxrec, uploads=uploads,
             **keywords).execute()
 
-    # alias for service discovery
-    search = run_sync
+    def search(self, query, *, language="ADQL", maxrec=None, uploads=None,
+               force_sync=False, **keywords):
+        """
+        submit a TAP query and return its result.
+        Checks the prefer_async setting and runs the query synchronously or
+        asynchronously depending on the setting.
+
+        Parameters
+        ----------
+        query : str
+            The query string / parameters
+        language : str
+            specifies the query language, default ADQL.
+            useful for services which allow to use the backend query language.
+        maxrec : int
+            the maximum records to return. defaults to the service default
+        uploads : dict
+            a mapping from table names to file like objects containing a votable
+        force_sync : bool
+            if True, forces synchronous execution regardless of prefer_async setting
+        **keywords
+            additional keyword arguments passed to the underlying query method
+
+        Returns
+        -------
+        TAPResults
+            the query result
+
+        Raises
+        ------
+        DALServiceError
+            for errors connecting to or communicating with the service
+        DALQueryError
+            for errors either in the input query syntax or
+            other user errors detected by the service
+        DALFormatError
+            for errors parsing the VOTable response
+
+        See Also
+        --------
+        TAPResults
+        AsyncTAPJob
+        """
+        if self._prefer_async and not force_sync:
+            return self.run_async(query, language=language, maxrec=maxrec,
+                                  uploads=uploads, **keywords)
+        else:
+            return self.run_sync(query, language=language, maxrec=maxrec,
+                                 uploads=uploads, **keywords)
 
     def run_async(
             self, query, *, language="ADQL", maxrec=None, uploads=None,

pyvo/dal/tests/test_tap.py

@@ -1099,6 +1099,107 @@ def test_get_udf(self, tapservice):
         func = tapservice.get_tap_capability().get_adql().get_udf("IVO_hasword")  # case insensitive!
         assert func.form == "ivo_hasword(haystack TEXT, needle TEXT) -> INTEGER"
 
+    def test_prefer_async_default_false(self):
+        service = TAPService('http://example.com/tap')
+        assert service._prefer_async is False
+
+    def test_prefer_async_parameter(self):
+        service = TAPService('http://example.com/tap', prefer_async=True)
+        assert service._prefer_async is True
+
+    @pytest.mark.usefixtures('sync_fixture')
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+    def test_search_default_behavior_unchanged(self):
+        service = TAPService('http://example.com/tap')
+
+        from unittest.mock import patch
+        with patch.object(service, 'run_sync') as mock_sync, \
+                patch.object(service, 'run_async') as mock_async:
+            mock_sync.return_value = "sync_result"
+            mock_async.return_value = "async_result"
+            result = service.search("SELECT * FROM ivoa.obscore")
+            mock_sync.assert_called_once_with(
+                "SELECT * FROM ivoa.obscore",
+                language="ADQL",
+                maxrec=None,
+                uploads=None
+            )
+            mock_async.assert_not_called()
+            assert result == "sync_result"
+
+    @pytest.mark.usefixtures('async_fixture')
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+    def test_search_prefer_async_true(self):
+        service = TAPService('http://example.com/tap', prefer_async=True)
+
+        from unittest.mock import patch
+        with patch.object(service, 'run_sync') as mock_sync, \
+                patch.object(service, 'run_async') as mock_async:
+            mock_sync.return_value = "sync_result"
+            mock_async.return_value = "async_result"
+            result = service.search("SELECT * FROM ivoa.obscore")
+            mock_async.assert_called_once_with(
+                "SELECT * FROM ivoa.obscore",
+                language="ADQL",
+                maxrec=None,
+                uploads=None
+            )
+            mock_sync.assert_not_called()
+            assert result == "async_result"
+
+    @pytest.mark.usefixtures('sync_fixture')
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+    def test_search_force_sync_overrides_prefer_async(self):
+        service = TAPService('http://example.com/tap', prefer_async=True)
+
+        from unittest.mock import patch
+        with patch.object(service, 'run_sync') as mock_sync, \
+                patch.object(service, 'run_async') as mock_async:
+
+            mock_sync.return_value = "sync_result"
+            mock_async.return_value = "async_result"
+            result = service.search("SELECT * FROM ivoa.obscore", force_sync=True)
+            mock_sync.assert_called_once_with(
+                "SELECT * FROM ivoa.obscore",
+                language="ADQL",
+                maxrec=None,
+                uploads=None
+            )
+            mock_async.assert_not_called()
+            assert result == "sync_result"
+
+    @pytest.mark.usefixtures('async_fixture')
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+    @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+    def test_search_parameters_passed_through(self):
+        service = TAPService('http://example.com/tap', prefer_async=True)
+
+        uploads = {'test_table': 'test_data'}
+        from unittest.mock import patch
+        with patch.object(service, 'run_async') as mock_async:
+            mock_async.return_value = "async_result"
+            service.search(
+                "SELECT * FROM ivoa.obscore",
+                language="SQL",
+                maxrec=100,
+                uploads=uploads,
+                extra_param="test_value"
+            )
+            mock_async.assert_called_once_with(
+                "SELECT * FROM ivoa.obscore",
+                language="SQL",
+                maxrec=100,
+                uploads=uploads,
+                extra_param="test_value"
+            )
+
 
 def test_get_endpoint_candidates():
     # Directly instantiate the TAPService with a known base URL