import datetime import os from typing import List, Dict, Tuple import boto3 from boto3.dynamodb.conditions import Key from wow_token.flavor import Flavor from wow_token.region import Region REGION_MAP = { 'us-west-1': 'us-west-1', 'us-west-2': 'us-west-2', 'us-east-1': 'us-east-1', 'us-east-2': 'us-east-2', 'ap-south-1': 'eu-north-1', 'ap-northeast-3': 'ap-northeast-1', 'ap-northeast-2': 'ap-northeast-1', 'ap-southeast-1': 'ap-southeast-1', 'ap-southeast-2': 'ap-southeast-2', 'ap-northeast-1': 'ap-northeast-1', 'ca-central-1': 'us-east-1', 'eu-central-1': 'eu-north-1', 'eu-west-1': 'eu-west-1', 'eu-west-2': 'eu-west-1', 'eu-west-3': 'eu-west-3', 'eu-north-1': 'eu-north-1', 'sa-east-1': 'sa-east-1', 'eu-south-1': 'eu-north-1' } def _region_selector(): if os.environ['AWS_REGION'] in REGION_MAP: local_region = REGION_MAP[os.environ['AWS_REGION']] else: local_region = 'eu-central-1' return local_region class Recent: def __init__(self): self._ddb = boto3.resource('dynamodb', region_name=_region_selector()) self._tables = { Flavor.RETAIL: self._ddb.Table('wow-token-price-recent'), Flavor.CLASSIC: self._ddb.Table('wow-token-classic-price-recent'), } def get_after_unix_timestamp(self, flavor: Flavor, region: Region, timestamp: int) -> List[Tuple[str, int]]: response = self._tables[flavor].query( KeyConditionExpression=( Key('region').eq(region.value) & Key('timestamp').gte(timestamp) ) ) data = [] last_price = 0 for item in response['Items']: price = int(int(item['price']) / 10_000) # the raw copper value is what is stored in DynamoDB if last_price != price: item_time = datetime.datetime.fromtimestamp(int(item['timestamp']), datetime.UTC).isoformat() data.append(( item_time, price )) last_price = price return data