Have async raise 4xx/5xx status codes and add simultaneous connection limit

Author: chinandrew

src/client/delphi_epidata.py

@@ -13,7 +13,7 @@
 import asyncio
 import warnings
 
-from aiohttp import ClientSession
+from aiohttp import ClientSession, TCPConnector
 from pkg_resources import get_distribution, DistributionNotFound
 
 # Obtain package version for the user-agent. Uses the installed version by
@@ -709,27 +709,26 @@ def covidcast_nowcast(
     return Epidata._request(params)
 
   @staticmethod
-  def async_epidata(param_list, batch_size=100):
+  def async_epidata(param_list, batch_size=50):
     """Make asynchronous Epidata calls for a list of parameters."""
     async def async_get(params, session):
       """Helper function to make Epidata GET requests."""
       async with session.get(Epidata.BASE_URL, params=params) as response:
+        response.raise_for_status()
         return await response.json(), params
 
     async def async_make_calls(param_combos):
       """Helper function to asynchronously make and aggregate Epidata GET requests."""
       tasks = []
-      async with ClientSession() as session:
+      connector = TCPConnector(limit=batch_size)
+      async with ClientSession(connector=connector) as session:
         for param in param_combos:
           task = asyncio.ensure_future(async_get(param, session))
           tasks.append(task)
         responses = await asyncio.gather(*tasks)
         return responses
 
-    batches = [param_list[i:i+batch_size] for i in range(0, len(param_list), batch_size)]
-    responses = []
-    for batch in batches:
-      loop = asyncio.get_event_loop()
-      future = asyncio.ensure_future(async_make_calls(batch))
-      responses += loop.run_until_complete(future)
+    loop = asyncio.get_event_loop()
+    future = asyncio.ensure_future(async_make_calls(param_list))
+    responses = loop.run_until_complete(future)
     return responses