Make HttpEngine the default rather than AsyncHttpEngine

Due to buggy interaction with jupyter notebooks, which is how
most of our users work with the AllenSDK, make the synchronous
downloader the default for EcephysProjectCache.
See GH #1585.

Also resolves GH #1586 - passing 'asynchronous' to the
constructors now chooses the proper stream engine.

Author: kschelonka

allensdk/brain_observatory/ecephys/ecephys_project_api/ecephys_project_lims_api.py

@@ -568,7 +568,7 @@ def get_natural_scene_template(self, number: int) -> Iterable[bytes]:
 
     @classmethod
     def default(cls, lims_credentials: Optional[DbCredentials] = None, 
-                app_kwargs=None, asynchronous=True):
+                app_kwargs=None, asynchronous=False):
         """ Construct a "straightforward" lims api that can fetch data from 
         lims2.
 

allensdk/brain_observatory/ecephys/ecephys_project_api/ecephys_project_warehouse_api.py

@@ -307,7 +307,7 @@ def get_unit_analysis_metrics(self, unit_ids=None, ecephys_session_ids=None, ses
 
 
     @classmethod
-    def default(cls, asynchronous=True, **rma_kwargs):
+    def default(cls, asynchronous=False, **rma_kwargs):
         _rma_kwargs = {"scheme": "http", "host": "api.brain-map.org"}
         _rma_kwargs.update(rma_kwargs)
 

allensdk/brain_observatory/ecephys/ecephys_project_cache.py

@@ -167,7 +167,22 @@ class constructors rather than to initialize this class directly.
         self.fetch_tries = fetch_tries
         self.stream_writer = (stream_writer
                               or self.fetch_api.rma_engine.write_bytes)
-
+        if stream_writer is not None:
+            self.stream_writer = stream_writer
+        else:
+            if hasattr(self.fetch_api, "rma_engine"):    # EcephysProjectWarehouseApi    # noqa
+                self.stream_writer = self.fetch_api.rma_engine.write_bytes
+            # TODO: Make these names consistent in the different fetch apis
+            elif hasattr(self.fetch_api, "app_engine"):    # EcephysProjectLimsApi    # noqa
+                self.stream_writer = self.fetch_api.app_engine.write_bytes
+            else:
+                raise ValueError(
+                    "Must either set value for `stream_writer`, or use a "
+                    "`fetch_api` with an rma_engine or app_engine attribute "
+                    "that implements `write_bytes`. See `HttpEngine` and "
+                    "`AsyncHttpEngine` from "
+                    "allensdk.brain_observatory.ecephys.ecephys_project_api."
+                    "http_engine for examples.")
 
     def _get_sessions(self):
         path = self.get_cache_path(None, self.SESSIONS_KEY)
@@ -581,7 +596,7 @@ def _from_http_source_default(cls, fetch_api_cls, fetch_api_kwargs, **kwargs):
     def from_lims(cls, lims_credentials: Optional[DbCredentials] = None,
                   scheme: Optional[str] = None,
                   host: Optional[str] = None, 
-                  asynchronous: bool = True,
+                  asynchronous: bool = False,
                   manifest: Optional[Union[str, Path]] = None, 
                   version: Optional[str] = None, 
                   cache: bool = True,
@@ -607,7 +622,7 @@ def from_lims(cls, lims_credentials: Optional[DbCredentials] = None,
             value if unspecified. Will not be used unless `scheme` is
             also specified.
         asynchronous : bool
-            Whether to fetch file asynchronously. Defaults to True.
+            Whether to fetch file asynchronously. Defaults to False.
         manifest : str or Path
             full path at which manifest json will be stored
         version : str
@@ -636,7 +651,7 @@ def from_lims(cls, lims_credentials: Optional[DbCredentials] = None,
     def from_warehouse(cls, 
                        scheme: Optional[str] = None, 
                        host: Optional[str] = None,
-                       asynchronous: bool = True,
+                       asynchronous: bool = False,
                        manifest: Optional[Union[str, Path]] = None, 
                        version: Optional[str] = None,
                        cache: bool = True, 
@@ -657,7 +672,7 @@ def from_warehouse(cls,
             value if unspecified. Will not be used unless `scheme` is also
             specified.
         asynchronous : bool
-            Whether to fetch file asynchronously. Defaults to True.
+            Whether to fetch file asynchronously. Defaults to False.
         manifest : str or Path
             full path at which manifest json will be stored
         version : str
@@ -673,7 +688,7 @@ def from_warehouse(cls,
             app_kwargs = {"scheme": scheme, "host": host, 
                           "asynchronous": asynchronous}
         else:
-            app_kwargs = None
+            app_kwargs = {"asynchronous": asynchronous}
         return cls._from_http_source_default(
             EcephysProjectWarehouseApi, app_kwargs, manifest=manifest,
              version=version, cache=cache, fetch_tries=fetch_tries

allensdk/test/brain_observatory/ecephys/test_ecephys_project_cache.py

@@ -11,7 +11,7 @@
 import allensdk.brain_observatory.ecephys.ecephys_project_cache as epc
 import allensdk.brain_observatory.ecephys.write_nwb.__main__ as write_nwb
 from allensdk.brain_observatory.ecephys.ecephys_project_api.http_engine import (
-    write_from_stream, write_bytes_from_coroutine, AsyncHttpEngine
+    write_from_stream, write_bytes_from_coroutine, AsyncHttpEngine, HttpEngine
 )
 
 
@@ -195,14 +195,6 @@ def tmpdir_cache(shared_tmpdir, mock_api):
     )
 
 
-def test_stream_writer_method_default_correct():
-    """Checks that the stream_writer contained in the rma engine is used
-    when one is not supplied to the __init__ method.
-    """
-    cache = epc.EcephysProjectCache(stream_writer=None)
-    assert cache.stream_writer == cache.fetch_api.rma_engine.write_bytes
-
-
 def lazy_cache_test(cache, cache_name, api_name, expected, *args, **kwargs):
     obtained_one = getattr(cache, cache_name)(*args, **kwargs)
     obtained_two = getattr(cache, cache_name)(*args, **kwargs)
@@ -389,5 +381,83 @@ def test_from_lims_default(tmpdir_factory):
     cache = epc.EcephysProjectCache.from_lims(
         manifest=os.path.join(tmpdir, "manifest.json")
     )
-    assert isinstance(cache.fetch_api.app_engine, AsyncHttpEngine)
-    assert cache.stream_writer is epc.write_bytes_from_coroutine
+    assert isinstance(cache.fetch_api.app_engine, HttpEngine)
+    assert cache.stream_writer is write_from_stream
+    assert cache.fetch_api.app_engine.scheme == "http"
+    assert cache.fetch_api.app_engine.host == "lims2"
+
+
+def test_from_warehouse_default(tmpdir_factory):
+    tmpdir = str(tmpdir_factory.mktemp("test_from_warehouse_default"))
+
+    cache = epc.EcephysProjectCache.from_warehouse(
+        manifest=os.path.join(tmpdir, "manifest.json")
+    )
+    assert isinstance(cache.fetch_api.rma_engine, HttpEngine)
+    assert cache.stream_writer is write_from_stream
+    assert cache.fetch_api.rma_engine.scheme == "http"
+    assert cache.fetch_api.rma_engine.host == "api.brain-map.org"
+
+
+def test_init_default(tmpdir_factory):
+    tmpdir = str(tmpdir_factory.mktemp("test_init_default"))
+    cache = epc.EcephysProjectCache(
+        manifest=os.path.join(tmpdir, "manifest.json")
+    )
+    assert isinstance(cache.fetch_api.rma_engine, HttpEngine)
+    assert cache.stream_writer is cache.fetch_api.rma_engine.write_bytes
+    assert cache.fetch_api.rma_engine.scheme == "http"
+    assert cache.fetch_api.rma_engine.host == "api.brain-map.org"
+
+
+@pytest.mark.parametrize(
+    ("cache_constructor, asynchronous, engine_attr, expected_engine,"
+     "expected_scheme, expected_host, expected_stream_writer"), [
+        (
+            epc.EcephysProjectCache.from_lims, True,
+            "app_engine", AsyncHttpEngine, "http", "lims2",
+            write_bytes_from_coroutine
+        ),
+        (
+            epc.EcephysProjectCache.from_warehouse, True,
+            "rma_engine", AsyncHttpEngine, "http", "api.brain-map.org",
+            write_bytes_from_coroutine
+        ),
+        (
+            epc.EcephysProjectCache.from_warehouse, False,
+            "rma_engine", HttpEngine, "http", "api.brain-map.org",
+            write_from_stream
+        ),
+        (
+            epc.EcephysProjectCache.from_lims, False,
+            "app_engine", HttpEngine, "http", "lims2",
+            write_from_stream
+        ),
+    ])
+def test_stream_asynchronous_arg(
+        cache_constructor, asynchronous, engine_attr, expected_engine,
+        expected_scheme, expected_host, expected_stream_writer,
+        tmpdir_factory):
+    """ Ensure the proper stream engine is chosen from the `asynchronous`
+    argument in the EcephysProjectCache constructors (using other default
+    values)."""
+    tmpdir = str(tmpdir_factory.mktemp("test_stream_async_args"))
+    cache = cache_constructor(
+        asynchronous=asynchronous,
+        manifest=os.path.join(tmpdir, "manifest.json")
+    )
+    engine = getattr(cache.fetch_api, engine_attr)
+    assert isinstance(engine, expected_engine)
+    assert cache.stream_writer is expected_stream_writer
+    assert engine.scheme == expected_scheme
+    assert engine.host == expected_host
+
+
+def test_stream_writer_method_default_correct(tmpdir_factory):
+    """Checks that the stream_writer contained in the rma engine is used
+    when one is not supplied to the __init__ method.
+    """
+    tmpdir = str(tmpdir_factory.mktemp("test_from_warehouse_default"))
+    manifest = os.path.join(tmpdir, "manifest.json")
+    cache = epc.EcephysProjectCache(stream_writer=None, manifest=manifest)
+    assert cache.stream_writer == cache.fetch_api.rma_engine.write_bytes