wow-token-app-bot

An interactions.py Discord bot for getting and alerting on the WoW Token Price
git clone https://git.emily.sh/wowtoken-app/wow-token-app-bot.git
Log | Files | Refs | README | LICENSE

user_schema.py (3221B)

      1 from typing import List, Dict
      2 
      3 from aiodynamo.client import Table
      4 from aiodynamo.errors import ItemNotFound
      5 
      6 import token_bot.persistant_database as pdb
      7 from token_bot.token_database.region import Region
      8 
      9 
     10 class User:
     11     def __init__(
     12         self,
     13         user_id: int,
     14         region: Region = None,
     15         subscribed_alerts: List["pdb.Alert"] = None,
     16     ) -> None:
     17         self.user_id: int = user_id
     18         self._loaded: bool = False
     19         self.region: Region = region
     20         self.subscribed_alerts: List[pdb.Alert] = subscribed_alerts
     21 
     22     def __eq__(self, other):
     23         return self.user_id == other.user_id
     24 
     25     def __hash__(self):
     26         return hash(self.user_id)
     27 
     28     @classmethod
     29     def from_item(
     30         cls, primary_key: int, region: Region, subscribed_alerts: List[str]
     31     ) -> "User":
     32         alerts = [pdb.Alert.from_str(alert_str) for alert_str in subscribed_alerts]
     33         return cls(primary_key, region, alerts)
     34 
     35     @property
     36     def primary_key(self) -> str:
     37         return str(self.user_id)
     38 
     39     @property
     40     def primary_key_name(self) -> str:
     41         return "user_id"
     42 
     43     @property
     44     def key(self) -> Dict[str, str]:
     45         return {self.primary_key_name: self.primary_key}
     46 
     47     def _subscribed_alerts_as_trinity_list(self) -> List[str]:
     48         return (
     49             [str(alert) for alert in self.subscribed_alerts]
     50             if self.subscribed_alerts
     51             else []
     52         )
     53 
     54     async def _lazy_load(self, table: Table, consistent: bool = False) -> None:
     55         if consistent or not self._loaded:
     56             await self.get(table, consistent=consistent)
     57 
     58     async def put(self, table: Table) -> None:
     59         await table.put_item(
     60             item={
     61                 self.primary_key_name: self.primary_key,
     62                 "region": self.region,
     63                 "subscribed_alerts": self._subscribed_alerts_as_trinity_list(),
     64             }
     65         )
     66 
     67     async def delete(self, table: Table) -> None:
     68         if not self._loaded:
     69             await self._lazy_load(table, consistent=True)
     70         await table.delete_item(
     71             key={self.primary_key_name: self.primary_key},
     72         )
     73 
     74     async def get(self, table: Table, consistent: bool = False) -> bool:
     75         try:
     76             response = await table.get_item(key=self.key, consistent_read=consistent)
     77         except ItemNotFound:
     78             return False
     79 
     80         self.subscribed_alerts = []
     81         for string_trinity in response["subscribed_alerts"]:
     82             self.subscribed_alerts.append(pdb.Alert.from_str(string_trinity))
     83         self.region = Region(response["region"])
     84         return True
     85 
     86     async def add_alert(
     87         self, table: Table, alert: "pdb.Alert", consistent: bool = False
     88     ) -> None:
     89         await self._lazy_load(table, consistent=consistent)
     90         if alert not in self.subscribed_alerts:
     91             self.subscribed_alerts.append(alert)
     92             await self.put(table)
     93 
     94     async def remove_alert(
     95         self, table: Table, alert: "pdb.Alert", consistent: bool = True
     96     ) -> None:
     97         await self._lazy_load(table, consistent=consistent)
     98         if alert in self.subscribed_alerts:
     99             self.subscribed_alerts.remove(alert)
    100             await self.put(table)