wowtoken.app-backend

git clone https://git.emily.sh/wowtoken-app/wowtoken.app-backend.git
Log | Files | Refs | LICENSE

compacted.py (3084B)

      1 import datetime
      2 import os
      3 from typing import List, Dict, Tuple, Union, Type
      4 
      5 import boto3
      6 from boto3.dynamodb.conditions import Key
      7 
      8 from wow_token.db.trinity import Trinity
      9 from wow_token.db.year_month import YearMonth
     10 from wow_token.db.cache import Cache
     11 from wow_token.region import Region
     12 
     13 # TODO: Reduce Compacted Table Sprawl
     14 
     15 REGION_MAP = {
     16     'us-west-1': 'us-west-1',
     17     'us-west-2': 'us-west-2',
     18     'us-east-1': 'us-east-1',
     19     'us-east-2': 'us-east-2',
     20     'ap-south-1': 'eu-north-1',
     21     'ap-northeast-3': 'ap-northeast-1',
     22     'ap-northeast-2': 'ap-northeast-1',
     23     'ap-southeast-1': 'ap-southeast-1',
     24     'ap-southeast-2': 'ap-southeast-2',
     25     'ap-northeast-1': 'ap-northeast-1',
     26     'ca-central-1': 'us-east-1',
     27     'eu-central-1': 'eu-north-1',
     28     'eu-west-1': 'eu-west-1',
     29     'eu-west-2': 'eu-west-1',
     30     'eu-west-3': 'eu-west-3',
     31     'eu-north-1': 'eu-north-1',
     32     'sa-east-1': 'sa-east-1',
     33     'eu-south-1': 'eu-north-1'
     34 }
     35 
     36 
     37 def _region_selector():
     38     if os.environ['AWS_REGION'] in REGION_MAP:
     39         local_region = REGION_MAP[os.environ['AWS_REGION']]
     40     else:
     41         local_region = 'eu-central-1'
     42     return local_region
     43 
     44 
     45 def _data_as_str(data: List[Tuple[datetime.datetime, int]]) -> List[Tuple[str, int]]:
     46     data_as_str = []
     47     for timestamp, price in data:
     48         data_as_str.append((timestamp.isoformat(), price))
     49     return data_as_str
     50 
     51 
     52 class Compacted:
     53     _cache : Cache
     54     def __init__(self):
     55         self._ddb = boto3.resource('dynamodb', region_name=_region_selector())
     56         self._table = self._ddb.Table('wow-token-compacted')
     57         self._cache = Cache(self)
     58 
     59     def ddb_get_data(self, trinity: Trinity, _type: Union[Type[str], Type[datetime.datetime]] = datetime.datetime) -> Union[List[Tuple[datetime.datetime, int]], List[Tuple[str, int]]]:
     60         data = []
     61         response = self._table.query(
     62             KeyConditionExpression=Key('region-flavor-timestamp').eq(str(trinity))
     63         )
     64         if response['Items']:
     65             for timestamp, price in response['Items'][0]['data'].items():
     66                 date_time = datetime.datetime.fromtimestamp(int(timestamp), datetime.UTC)
     67                 if _type == str:
     68                     date_time = date_time.isoformat()
     69                 data.append((
     70                     date_time,
     71                     int(price)
     72                 ))
     73         return sorted(data, key=lambda x: x[0])
     74 
     75     def get_month(self, trinity: Trinity, _type: Union[Type[str], Type[datetime.datetime]] = datetime.datetime) -> Union[List[Tuple[datetime.datetime, int]], List[Tuple[str, int]]]:
     76         if _type == str:
     77             return _data_as_str(self._cache.get_month(trinity))
     78         return self._cache.get_month(trinity)
     79 
     80     def get_precomputed_range(self, trinity: Trinity, _type: Union[Type[str], Type[datetime.datetime]] = datetime.datetime) -> Union[List[Tuple[datetime.datetime, int]], List[Tuple[str, int]]]:
     81         if isinstance(trinity.range, YearMonth):
     82             return self.get_month(trinity, _type=_type)
     83         else:
     84             return self.ddb_get_data(trinity, _type=_type)