⚡️ perf: concurrent fetch all

dsws_client/client.py

@@ -1,5 +1,6 @@
 """The DSWS client."""
 
+import concurrent.futures
 import itertools
 import logging
 import sys
@@ -52,6 +53,7 @@ def __init__(self, username: str, password: str, **kwargs: Any) -> None:
             verify=config.ssl_cert or True,
             headers={"Content-Type": "application/json"},
         )
+        self._max_concurrency = config.max_concurrency
         self._app_id = config.app_id
         self._data_source = config.data_source
         self._debug = config.debug
@@ -83,7 +85,7 @@ def fetch_snapshot_data(
             return_symbol_names=True,
             return_field_names=True,
         )
-        responses = self.fetch_all(request_bundles)
+        responses = self.fetch_all(request_bundles, threaded=self._max_concurrency > 1)
         data_responses = itertools.chain.from_iterable(
             response.data_responses for response in responses
         )
@@ -110,7 +112,7 @@ def fetch_timeseries_data(  # noqa: PLR0913
             return_symbol_names=True,
             return_field_names=True,
         )
-        responses = self.fetch_all(request_bundles)
+        responses = self.fetch_all(request_bundles, threaded=self._max_concurrency > 1)
         data_responses = itertools.chain.from_iterable(
             response.data_responses for response in responses
         )
@@ -171,11 +173,31 @@ def fetch_bundle(
     def fetch_all(
         self,
         request_bundles: List[List[DSDataRequest]],
+        *,
+        threaded: bool = False,
     ) -> Iterator[DSGetDataBundleResponse]:
         """Fetch as many bundles as needed to get all items."""
+        if threaded:
+            yield from self.fetch_all_threaded(request_bundles)
         for bundle in request_bundles:
             yield self.fetch_bundle(bundle)
 
+    def fetch_all_threaded(
+        self,
+        request_bundles: List[List[DSDataRequest]],
+    ) -> Iterator[DSGetDataBundleResponse]:
+        """Fetch as many bundles as needed to get all items (concurrently)."""
+        with concurrent.futures.ThreadPoolExecutor(
+            max_workers=self._max_concurrency
+        ) as executor:
+            logger.debug("fetching bundles in parallel")
+            futures = [
+                executor.submit(self.fetch_bundle, bundle) for bundle in request_bundles
+            ]
+            return (
+                future.result() for future in concurrent.futures.as_completed(futures)
+            )
+
     def fetch_token(self, **kwargs: object) -> Token:
         """
         Fetch a new token.

dsws_client/config.py

@@ -20,6 +20,7 @@ class DSWSConfig:
     ssl_cert: Optional[str] = None
     app_id: str = f"dsws-client-{__version__}"
     data_source: Optional[str] = None
+    max_concurrency: int = 1
     debug: bool = attrs.field(default=False, converter=attrs.converters.to_bool)
 
     def __init__(self, **kwargs: Any) -> None: