Use dependency injection for key schema

Author: Andrew Brookins

app/main.py

@@ -1,21 +1,58 @@
 import functools
-import json
 import logging
 from datetime import datetime
+from typing import Dict
+from typing import Iterable
+from typing import List
+from typing import Tuple
+from typing import Union
 
 import aioredis
 import requests
 from aioredis.exceptions import ResponseError
+from fastapi import Depends
 from fastapi import FastAPI
 from pydantic import BaseSettings
 
 
-TIMESERIES_KEY = 'is-bitcoin-lit:sentiment:mean:{time}'
-SUMMARY_KEY = 'is-bitcoin-lit:summary:hourly:{time}'
-SENTIMENT_API_URL = 'https://api.senticrypt.com/v1/history/bitcoin-{time}.json'
+DEFAULT_KEY_PREFIX = 'is-bitcoin-lit'
+SENTIMENT_API_URL = 'HTTps://api.senticrypt.com/v1/history/bitcoin-{time}.json'
 TIME_FORMAT_STRING = '%Y-%m-%d_%H'
 
 
+def prefixed_key(f):
+    """
+    A method decorator that prefixes return values.
+    Prefixes any string that the decorated method `f` returns with the value of
+    the `prefix` attribute on the owner object `self`.
+    """
+
+    def prefixed_method(self, *args, **kwargs):
+        key = f(self, *args, **kwargs)
+        return f'{self.prefix}:{key}'
+
+    return prefixed_method
+
+
+class Keys:
+    """Methods to generate key names for Redis data structures."""
+
+    def __init__(self, prefix: str = DEFAULT_KEY_PREFIX):
+        self.prefix = prefix
+
+    @prefixed_key
+    def timeseries_sentiment_key(self) -> str:
+        return f'sentiment:mean'
+
+    @prefixed_key
+    def timeseries_price_key(self) -> str:
+        return f'price:mean'
+
+    @prefixed_key
+    def summary_key(self) -> str:
+        return f'summary:hourly'
+
+
 class Config(BaseSettings):
     redis_url: str = 'redis://redis:6379'
 
@@ -27,17 +64,44 @@ class Config(BaseSettings):
 
 
 def make_summary(data):
+    """Take a series of averages and summarize them as means of means."""
     return {
         'time': datetime.now().timestamp(),
-        'mean_sentiment': sum(d['mean'] for d in data) / len(data),
+        'mean_of_means_sentiment': sum(d['mean'] for d in data) / len(data),
+        'mean_of_means_price': sum(float(d['btc_price']) for d in data) / len(data),
     }
 
 
+async def add_many_to_timeseries(
+    key_pairs: Iterable[Tuple[str, str]],
+    data: List[Dict[str, Union[str, float]]]
+):
+    """
+    Add many samples to a single timeseries key.
+
+    `key_pairs` is an iteratble of tuples containing in the 0th position the
+    timestamp key into which to insert entries and the 1th position the name
+    of the key within th e`data` dict to find the sample.
+    """
+    partial = functools.partial(redis.execute_command, 'TS.MADD')
+    for datapoint in data:
+        for key, attr in key_pairs:
+            partial = functools.partial(
+                partial, key, datapoint['timestamp'], datapoint[attr],
+            )
+    return await partial()
+
+
+def make_keys():
+    return Keys()
+
+
 @app.get('/is-bitcoin-lit')
-async def bitcoin():
+async def bitcoin(keys: Keys = Depends(make_keys)):
     sentiment_time = datetime.now().strftime(TIME_FORMAT_STRING)
-    summary_key = SUMMARY_KEY.format(time=sentiment_time)
-    ts_key = TIMESERIES_KEY.format(time=sentiment_time)
+    summary_key = keys.summary_key()
+    ts_price_key = keys.timeseries_price_key()
+    ts_sentiment_key = keys.timeseries_sentiment_key()
     url = SENTIMENT_API_URL.format(time=sentiment_time)
 
     summary = await redis.hgetall(summary_key)
@@ -48,20 +112,21 @@ async def bitcoin():
         summary = make_summary(data)
         await redis.hset(summary_key, mapping=summary)
         await redis.expire(summary_key, 60)
-        partial = functools.partial(redis.execute_command, 'TS.MADD', ts_key)
-        for datapoint in data:
-            partial = functools.partial(
-                partial, datapoint['timestamp'], datapoint['mean'],
-            )
-        await partial()
+        await add_many_to_timeseries(
+            (
+                (ts_price_key, 'btc_price'),
+                (ts_sentiment_key, 'mean'),
+            ), data,
+        )
 
     return summary
 
 
 @app.on_event('startup')
-async def startup_event():
+async def startup_event(keys: Keys = Depends(make_keys)):
     try:
-        redis.execute_command('TS.CREATE', TIMESERIES_KEY)
+        redis.execute_command('TS.CREATE', keys.timeseries_sentiment_key())
+        redis.execute_command('TS.CREATE', keys.timeseries_price_key())
     except ResponseError:
         # Time series already exists
         pass

tests/conftest.py

@@ -7,12 +7,10 @@
 
 from app.main import app
 from app.main import config
+from app.main import Keys
+from app.main import make_keys
 
-
-@pytest.fixture(scope='function')
-async def client():
-    async with AsyncClient(app=app, base_url='http://test') as ac:
-        yield ac
+TEST_PREFIX = 'test:is-bitcoin-lit'
 
 
 @pytest.fixture(scope='module')
@@ -26,3 +24,24 @@ def event_loop(request):
     loop = asyncio.get_event_loop_policy().new_event_loop()
     yield loop
     loop.close()
+
+
+@pytest.fixture
+async def keys(redis: aioredis.Redis):
+    def make_test_keys():
+        return Keys(TEST_PREFIX)
+    app.dependency_overrides[make_keys] = make_test_keys
+    keys = make_test_keys()
+
+    yield keys
+
+    # Cleanup any test keys the test run created
+    keys = await redis.keys(f'{TEST_PREFIX}*')
+    if keys:
+        await redis.delete(*keys)
+
+
+@pytest.fixture(scope='function')
+async def client(keys):
+    async with AsyncClient(app=app, base_url='http://test') as ac:
+        yield ac

tests/test_bitcoin_api.py

@@ -1,20 +1,11 @@
-from datetime import datetime
-
 import pytest
 from aioredis import Redis
 from httpx import AsyncClient
 
-from app.main import SUMMARY_KEY
-from app.main import TIME_FORMAT_STRING
+from app.main import Keys
 
 URL = '/is-bitcoin-lit'
-SENTICRYPT_FIELDS = ('time', 'mean_sentiment')
-
-
-@pytest.fixture
-def summary_key():
-    time = datetime.now().strftime(TIME_FORMAT_STRING)
-    yield SUMMARY_KEY.format(time=time)
+SENTICRYPT_FIELDS = ('time', 'mean_of_means_sentiment', 'mean_of_means_price')
 
 
 @pytest.mark.asyncio
@@ -29,11 +20,11 @@ async def test_api(client: AsyncClient):
 
 
 @pytest.mark.asyncio
-async def test_api_cache(client: AsyncClient, redis: Redis, summary_key):
+async def test_api_cache(client: AsyncClient, redis: Redis, keys: Keys):
     # prime the cache
     await client.get(URL)
 
-    summary = await redis.hgetall(summary_key)
+    summary = await redis.hgetall(keys.summary_key())
     assert summary is not None
 
     for field in SENTICRYPT_FIELDS: