Migrating to Quart

gentle_gnomes/pyproject.toml

@@ -12,6 +12,7 @@ flask = "^1.0"
 python-dotenv = "^0.10.3"
 requests = "^2.22"
 scipy = "^1.3"
+aiohttp = "^3.5"
 
 [tool.poetry.dev-dependencies]
 pytest = "^4.6"

gentle_gnomes/src/__init__.py

@@ -7,10 +7,9 @@
 def create_app(test_config=None):
     app = Flask(__name__, instance_relative_config=True)
     app.config.from_mapping(SECRET_KEY='dev')
+    app.config.from_pyfile('config.py', silent=True)
 
-    if test_config is None:
-        app.config.from_pyfile('config.py', silent=True)
-    else:
+    if test_config is not None:
         app.config.from_mapping(test_config)
 
     app.azavea = azavea.Client(app.config['AZAVEA_TOKEN'])

gentle_gnomes/src/azavea.py

@@ -1,11 +1,15 @@
-from typing import Dict, Iterator, List, NamedTuple, Optional, Union
+from functools import partial
+from datetime import datetime
 
 import requests
+import aiohttp
+import typing as t
+import asyncio as aio
 
 BASE_URL = 'https://app.climate.azavea.com/api'
 
 
-class City(NamedTuple):
+class City(t.NamedTuple):
     name: str
     admin: str
     id: int
@@ -18,16 +22,17 @@ class Client:
     """Client for interacting with the Azavea Climate API."""
 
     def __init__(self, token: str):
-        self.session = requests.Session()
-        self.session.headers = {'Authorization': f'Token {token}'}
+        self.headers = {'Authorization': f'Token {token}'}
+        self.proxy = AsyncProxy()
 
-    def _get(self, endpoint: str, **kwargs) -> Union[Dict, List]:
-        response = self.session.get(BASE_URL + endpoint, ** kwargs)
-        response.raise_for_status()
+    def _get(self, endpoint: str, **kwargs) -> t.Union[t.Dict, t.List]:
+        # Update headers with default
+        kwargs['headers'] = {**self.headers, **kwargs.get('headers', {})}
 
-        return response.json()
+        return self.proxy.get(BASE_URL + endpoint, **kwargs)
 
-    def get_cities(self, **kwargs) -> Iterator[City]:
+
+    def get_cities(self, **kwargs) -> t.Iterator[City]:
         """Return all available cities."""
         params = {'page': 1}
         params.update(kwargs.get('params', {}))
@@ -43,7 +48,7 @@ def get_cities(self, **kwargs) -> Iterator[City]:
             for city in cities['features']:
                 yield City(city['properties']['name'], city['properties']['admin'], city['id'])
 
-    def get_nearest_city(self, lat: float, lon: float, limit: int = 1, **kwargs) -> Optional[City]:
+    def get_nearest_city(self, lat: float, lon: float, limit: int = 1, **kwargs) -> t.Optional[City]:
         """Return the nearest city to the provided lat/lon or None if not found."""
         params = {
             'lat': lat,
@@ -57,18 +62,65 @@ def get_nearest_city(self, lat: float, lon: float, limit: int = 1, **kwargs) ->
             city = cities['features'][0]
             return City(city['properties']['name'], city['properties']['admin'], city['id'])
 
-    def get_scenarios(self, **kwargs) -> List:
+    def get_scenarios(self, **kwargs) -> t.List:
         """Return all available scenarios."""
         return self._get('/scenario', **kwargs)
 
-    def get_indicators(self, **kwargs) -> Dict:
+    def get_indicators(self, **kwargs) -> t.Dict:
         """Return the full list of indicators."""
         return self._get('/indicator', **kwargs)
 
-    def get_indicator_details(self, indicator: str, **kwargs) -> Dict:
+    def get_indicator_details(self, indicator: str, **kwargs) -> t.Dict:
         """Return the description and parameters of a specified indicator."""
         return self._get(f'/indicator/{indicator}', **kwargs)
 
-    def get_indicator_data(self, city: int, scenario: str, indicator: str, **kwargs) -> Dict:
+    def get_indicator_data(self, city: int, scenario: str, indicator: str, **kwargs) -> t.Dict:
         """Return derived climate indicator data for the requested indicator."""
         return self._get(f'/climate-data/{city}/{scenario}/indicator/{indicator}', **kwargs)
+
+
+class AsyncProxy:
+
+    def __init__(self):
+        self._pending = []  # [(url, kwargs), ...]
+
+    async def _process_requests(self, *requests: t.Tuple[str, dict]) -> t.Tuple[t.Union[dict, list]]:
+        session = aiohttp.ClientSession()
+
+        async def make_request(request):
+            url, kwargs = request
+            async with session.get(url, **kwargs) as response:
+                response.raise_for_status()
+                return await response.json()
+
+        async with session:
+            return await aio.gather(*map(make_request, requests))
+
+    def get(self, url: str, now=True, **kwargs) -> t.Optional[t.Union[dict, list]]:
+        """
+        Make a get request for the `url` using `kwargs`.
+
+        If `now` is False, store the request for a later. This is to be used with `.collect`, which
+        executes the requests concurrently. The response data is returned in the order it was
+        recieved.
+
+            ...
+            proxy.get(url0, now=False)
+            proxy.get(url1, now=False)
+
+            url0_data, url1_data = proxy.collect()
+        """
+        if now:
+            return aio.run(
+                self._process_requests((url, kwargs))
+            )
+        else:
+            self._pending.append((url, kwargs))
+
+    def collect(self) -> t.Tuple[t.Union[dict, list]]:
+        """Return pending requests in the order they were recieved."""
+        result = aio.run(
+            self._process_requests(*self._pending)
+        )
+        self._pending.clear()
+        return result

gentle_gnomes/tests/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 
 from src import create_app
+from src.azavea import Client
 
 
 @pytest.fixture
@@ -17,3 +18,8 @@ def client(app):
 @pytest.fixture
 def runner(app):
     return app.test_cli_runner()
+
+
+@pytest.fixture
+def azavea(app):
+    return Client(app.config['AZAVEA_TOKEN'])

gentle_gnomes/tests/test_azavea.py

@@ -0,0 +1,37 @@
+from time import time
+
+
+def test_request_data_exists(azavea):
+    assert azavea.get_indicators()
+
+
+def test_async_request_adds_to_pending(azavea):
+    azavea.get_indicators(now=False)
+    assert azavea.proxy._pending
+
+
+def test_async_requests_perserve_order(azavea):
+    indicators = azavea.get_indicators()
+    scenarios = azavea.get_scenarios()
+
+    azavea.get_indicators(now=False)
+    azavea.get_scenarios(now=False)
+
+    assert indicators, scenarios == azavea.proxy.collect()
+
+
+def test_async_proxy_is_actually_doing_something(azavea):
+    synct = time()
+    azavea.get_indicators()
+    azavea.get_indicators()
+    azavea.get_indicators()
+    synct = time() - synct
+
+    asynct = time()
+    azavea.get_indicators(now=False)
+    azavea.get_indicators(now=False)
+    azavea.get_indicators(now=False)
+    azavea.proxy.collect()
+    asynct = time() - asynct
+
+    assert asynct < synct