recent.py (2141B)
1 import datetime 2 import os 3 from typing import List, Dict, Tuple 4 5 import boto3 6 from boto3.dynamodb.conditions import Key 7 8 from wow_token.flavor import Flavor 9 from wow_token.region import Region 10 11 REGION_MAP = { 12 'us-west-1': 'us-west-1', 13 'us-west-2': 'us-west-2', 14 'us-east-1': 'us-east-1', 15 'us-east-2': 'us-east-2', 16 'ap-south-1': 'eu-north-1', 17 'ap-northeast-3': 'ap-northeast-1', 18 'ap-northeast-2': 'ap-northeast-1', 19 'ap-southeast-1': 'ap-southeast-1', 20 'ap-southeast-2': 'ap-southeast-2', 21 'ap-northeast-1': 'ap-northeast-1', 22 'ca-central-1': 'us-east-1', 23 'eu-central-1': 'eu-north-1', 24 'eu-west-1': 'eu-west-1', 25 'eu-west-2': 'eu-west-1', 26 'eu-west-3': 'eu-west-3', 27 'eu-north-1': 'eu-north-1', 28 'sa-east-1': 'sa-east-1', 29 'eu-south-1': 'eu-north-1' 30 } 31 32 33 def _region_selector(): 34 if os.environ['AWS_REGION'] in REGION_MAP: 35 local_region = REGION_MAP[os.environ['AWS_REGION']] 36 else: 37 local_region = 'eu-central-1' 38 return local_region 39 40 41 class Recent: 42 def __init__(self): 43 self._ddb = boto3.resource('dynamodb', region_name=_region_selector()) 44 self._tables = { 45 Flavor.RETAIL: self._ddb.Table('wow-token-price-recent'), 46 Flavor.CLASSIC: self._ddb.Table('wow-token-classic-price-recent'), 47 } 48 49 def get_after_unix_timestamp(self, flavor: Flavor, region: Region, timestamp: int) -> List[Tuple[str, int]]: 50 response = self._tables[flavor].query( 51 KeyConditionExpression=( 52 Key('region').eq(region.value) & 53 Key('timestamp').gte(timestamp) 54 ) 55 ) 56 data = [] 57 last_price = 0 58 for item in response['Items']: 59 price = int(int(item['price']) / 10_000) # the raw copper value is what is stored in DynamoDB 60 if last_price != price: 61 item_time = datetime.datetime.fromtimestamp(int(item['timestamp']), datetime.UTC).isoformat() 62 data.append(( 63 item_time, 64 price 65 )) 66 last_price = price 67 return data