wowtoken.app-backend

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

relative_path_handler.py (4037B)

      1 import datetime
      2 from typing import List, Tuple
      3 
      4 from wow_token.db.cached_range import CachedRange
      5 from wow_token.db.compacted import Compacted
      6 from wow_token.db.recent import Recent
      7 from wow_token.db.trinity import Trinity
      8 from wow_token.flavor import Flavor
      9 from wow_token.path_handler.relative_error import InvalidRelativePathError
     10 from wow_token.region import Region
     11 
     12 
     13 class RelativePathHandler:
     14     _cdb : Compacted
     15     _rdb : Recent
     16     def __init__(self, cdb: Compacted, rdb: Recent):
     17         self._cdb = cdb
     18         self._rdb = rdb
     19 
     20 
     21     def get_by_timedelta(self, flavor: Flavor, region: Region, timedelta: datetime.timedelta) -> List[Tuple[str, int]]:
     22         current_time = datetime.datetime.now(datetime.UTC)
     23         start_time = current_time - timedelta
     24 
     25         if timedelta.days < 61:
     26             return self._rdb.get_after_unix_timestamp(flavor, region, int(start_time.timestamp()))
     27         elif timedelta.days <= 90:
     28             trinity = Trinity(region, flavor, CachedRange('90d'))
     29         elif timedelta.days <= 183:
     30             trinity = Trinity(region, flavor, CachedRange('6m'))
     31         elif timedelta.days <= 365:
     32             trinity = Trinity(region, flavor, CachedRange('1y'))
     33         elif timedelta.days <= 730:
     34             trinity = Trinity(region, flavor, CachedRange('2y'))
     35         else:
     36             trinity = Trinity(region, flavor, CachedRange('all'))
     37 
     38         # If the data is exactly the size of the precomputed structure, go ahead and return it directly
     39         if timedelta.days == 90 or timedelta.days == 182 or timedelta.days == 365 or timedelta.days == 730:
     40             return self._cdb.get_precomputed_range(trinity, str)
     41 
     42         final_data = []
     43         data = self._cdb.get_precomputed_range(trinity)
     44         for timestamp, price in data:
     45             if timestamp >= start_time:
     46                 final_data.append((timestamp.isoformat(), price))
     47         return final_data
     48 
     49 
     50     def relative_time_handler(self, flavor: Flavor, region: Region, relative_range: str) -> List[Tuple[str, int]]:
     51         if relative_range == '30d':
     52             relative_range = '744h'
     53 
     54         relative_unit = relative_range[-1]
     55 
     56         match relative_unit:
     57             case 'h':
     58                 hours = int(relative_range[:-1])
     59                 if hours > 1488:
     60                     raise InvalidRelativePathError
     61                 start_time = datetime.datetime.now(datetime.UTC) - datetime.timedelta(hours=hours)
     62                 return self._rdb.get_after_unix_timestamp(flavor, region, int(start_time.timestamp()))
     63             case 'd':
     64                 days = int(relative_range[:-1])
     65                 if days > 730:
     66                     raise InvalidRelativePathError
     67                 delta = datetime.timedelta(days=days)
     68                 return self.get_by_timedelta(flavor, region, delta)
     69             case 'm':
     70                 months = int(relative_range[:-1])
     71                 if months > 48:
     72                     raise InvalidRelativePathError
     73                 delta = datetime.timedelta(days=int(30.437*months))
     74                 return self.get_by_timedelta(flavor, region, delta)
     75             case 'y':
     76                 years = int(relative_range[:-1])
     77                 if years > 10:
     78                     raise InvalidRelativePathError
     79                 delta = datetime.timedelta(days=int(365.25*years))
     80                 return self.get_by_timedelta(flavor, region, delta)
     81             case _:
     82                 if relative_range == 'all':
     83                     return self._cdb.get_precomputed_range(Trinity(region, flavor, CachedRange('all')), str)
     84                 raise InvalidRelativePathError
     85 
     86 
     87 
     88     def path_handler(self, uri) -> List[Tuple[str, int]]:
     89         # This URI takes the form of /v2/relative/{flavor}/{region}/{range}
     90         split_uri = uri.split('/')
     91         flavor = Flavor(split_uri[-3])
     92         region = Region(split_uri[-2])
     93         _range = split_uri[-1]
     94         if split_uri[-1].endswith('.json'):
     95             _range = split_uri[-1][:-5]
     96 
     97         return self.relative_time_handler(flavor, region, _range)
     98