Very Initial MVP

There is so much more to do, but I think it is time to commit this to VCS

token_bot/persistant_database/__init__.py

@@ -0,0 +1,4 @@
+from .alert_type import AlertType
+from .user_schema import User
+from .alert_schema import Alert
+

token_bot/persistant_database/alert_category.py

@@ -0,0 +1,19 @@
+from enum import Enum
+
+
+class AlertCategory(Enum):
+    HIGH = 1
+    LOW = 2
+    CUSTOM = 3
+
+    @staticmethod
+    def from_str(category: str):  # It gets mad when I use the Type[AlertCategory] as a type hint
+        match category:
+            case "high_alert_button":
+                return AlertCategory.HIGH
+            case "low_alert_button":
+                return AlertCategory.LOW
+            case "sp_add_button":
+                return AlertCategory.CUSTOM
+            case _:
+                return AlertCategory[category.upper()]

token_bot/persistant_database/alert_schema.py

@@ -0,0 +1,131 @@
+from typing import List
+
+from aiodynamo.client import Table
+from aiodynamo.errors import ItemNotFound
+from aiodynamo.expressions import F
+
+from token_bot.token_database.flavor import Flavor
+from token_bot.token_database.region import Region
+import token_bot.persistant_database as pdb
+
+
+class Alert:
+    def __init__(self, alert: pdb.AlertType, flavor: Flavor, region: Region, price: int = 0) -> None:
+        # AlertType is the Primary Key
+        self.alert_type: pdb.AlertType = alert
+        # Flavor (Retail, Classic) is the Sort Key
+        self.flavor: Flavor = flavor
+        self.region: Region = region
+        self.price: int = price
+        self._loaded: bool = False
+        self._users: List[pdb.User] = []
+
+    @classmethod
+    def from_item(cls, primary_key: int, sort_key: str, users: List[int]) -> 'Alert':
+        alert_type = pdb.AlertType(primary_key)
+        flavor_repr, region_repr, price_repr = sort_key.split('-')
+        flavor = Flavor(int(flavor_repr))
+        region = Region(region_repr)
+        price = int(price_repr)
+        return cls(alert_type, flavor, region, price)
+
+    @classmethod
+    def from_str(cls, string_trinity: str) -> 'Alert':
+        alert_repr, flavor_repr, region_repr, price_repr = string_trinity.split('-')
+        if len(string_trinity.split('-')) != 4:
+            raise ValueError
+        alert = pdb.AlertType(int(alert_repr))
+        flavor = Flavor(int(flavor_repr))
+        region = Region(region_repr)
+        price = int(price_repr)
+        return cls(alert, flavor, region, price)
+
+    @property
+    def primary_key(self) -> int:
+        return self.alert_type.value
+
+    @property
+    def primary_key_name(self) -> str:
+        return "alert"
+
+    @property
+    def sort_key(self) -> str:
+        return f"{self.flavor.value}-{self.region.value}-{self.price}"
+
+    @property
+    def sort_key_name(self) -> str:
+        return "flavor-region-price"
+
+    @property
+    def key(self) -> dict[str, str | int]:
+        return {
+            self.primary_key_name: self.primary_key,
+            self.sort_key_name: self.sort_key
+        }
+
+    def __str__(self):
+        return f"{self.alert_type.value}-{self.flavor.value}-{self.region.value}-{self.price}"
+
+    def __eq__(self, other):
+        return self.alert_type == other.alert_type and self.flavor == other.flavor and self.price == other.price
+
+    def to_human_string(self):
+        if self.alert_type.SPECIFIC_PRICE:
+            raise NotImplementedError
+        else:
+            return f"\n|Region: {self.region.value.upper()}\tFlavor: {self.flavor.value.upper()}\tAlert: {self.alert_type.name.lower()}|"
+
+    async def _lazy_load(self, table: Table, consistent: bool = False) -> None:
+        if consistent or not self._loaded:
+            await self.get(table, consistent=consistent)
+
+    async def _append_user(self, table: Table, user: pdb.User) -> None:
+        self._users.append(user)
+        await self.put(table)
+
+    async def _remove_user(self, table: Table, user: pdb.User) -> None:
+        update_expression = F("users").delete({user.user_id})
+        await table.update_item(
+            key=self.key,
+            update_expression=update_expression
+        )
+
+    async def put(self, table: Table) -> None:
+        user_ids = [user.user_id for user in self._users]
+        await table.put_item(
+            item={
+                self.primary_key_name: self.primary_key,
+                self.sort_key_name: self.sort_key,
+                'users': user_ids
+            }
+        )
+
+    async def get(self, table: Table, consistent: bool = False) -> bool:
+        try:
+            response = await table.get_item(
+                key=self.key,
+                consistent_read=consistent
+            )
+        except ItemNotFound:
+            return False
+        if 'Item' in response:
+            self._users = [pdb.User(int(user_id)) for user_id in response['Item']['users']['NS']]
+            self._loaded = True
+            return True
+
+    async def get_users(self, table: Table, consistent: bool = False) -> List[pdb.User]:
+        await self._lazy_load(table, consistent=consistent)
+
+        return self._users
+
+    async def add_user(self, table: Table, user: pdb.User, consistent: bool = False) -> None:
+        await self._lazy_load(table, consistent=consistent)
+
+        if user not in self._users:
+            await self._append_user(table=table, user=user)
+
+    async def remove_user(self, table: Table, user: pdb.User, consistent: bool = True) -> None:
+        await self._lazy_load(table, consistent=consistent)
+
+        if user in self._users:
+            await self._remove_user(table=table, user=user)

token_bot/persistant_database/alert_type.py

@@ -0,0 +1,36 @@
+from enum import Enum
+
+class AlertType(Enum):
+    ALL_TIME_HIGH = 1
+    ALL_TIME_LOW = 2
+    DAILY_HIGH = 3
+    DAILY_LOW = 4
+    WEEKLY_HIGH = 5
+    WEEKLY_LOW = 6
+    MONTHLY_HIGH = 7
+    MONTHLY_LOW = 8
+    YEARLY_HIGH = 9
+    YEARLY_LOW = 10
+    SPECIFIC_PRICE = 11
+
+    @staticmethod
+    def from_str(category: str):
+        match category:
+            case "Daily High":
+                return AlertType.DAILY_HIGH
+            case "Daily Low":
+                return AlertType.DAILY_LOW
+            case "Weekly High":
+                return AlertType.WEEKLY_HIGH
+            case "Weekly Low":
+                return AlertType.WEEKLY_LOW
+            case "Monthly High":
+                return AlertType.MONTHLY_HIGH
+            case "Monthly Low":
+                return AlertType.MONTHLY_LOW
+            case "Yearly High":
+                return AlertType.YEARLY_HIGH
+            case "Yearly Low":
+                return AlertType.YEARLY_LOW
+            case _:
+                return AlertType.SPECIFIC_PRICE

token_bot/persistant_database/database.py

@@ -0,0 +1,16 @@
+import os
+
+import aiohttp
+
+from aiodynamo.client import Client
+from aiodynamo.client import Table
+from aiodynamo.credentials import Credentials
+from aiodynamo.http.httpx import HTTPX
+from aiodynamo.http.aiohttp import AIOHTTP
+from httpx import AsyncClient
+
+
+class Database:
+    def __init__(self, session: aiohttp.ClientSession):
+        self.client = Client(AIOHTTP(session), Credentials.auto(), os.getenv('AWS_REGION'))
+

token_bot/persistant_database/user_schema.py

@@ -0,0 +1,80 @@
+from typing import List, Tuple, Dict
+
+from aiodynamo.client import Table
+from aiodynamo.errors import ItemNotFound
+
+
+from token_bot.token_database.flavor import Flavor
+from token_bot.token_database.region import Region
+import token_bot.persistant_database as pdb
+
+class User:
+    def __init__(self, user_id: int, region: Region = None, subscribed_alerts: List['pdb.Alert'] = None) -> None:
+        self.user_id: int = user_id
+        self._loaded: bool = False
+        self.region: Region = region
+        self.subscribed_alerts: List[pdb.Alert] = subscribed_alerts
+
+    def __eq__(self, other):
+        return self.user_id == other.user_id
+
+    @classmethod
+    def from_item(cls, primary_key: int, region: Region, subscribed_alerts: List[str]) -> 'User':
+        alerts = [pdb.Alert.from_str(alert_str) for alert_str in subscribed_alerts]
+        return cls(primary_key, region, alerts)
+
+    @property
+    def primary_key(self) -> int:
+        return self.user_id
+
+    @property
+    def primary_key_name(self) -> str:
+        return 'user_id'
+
+    @property
+    def key(self) -> Dict[str, int]:
+        return {
+            self.primary_key_name: self.primary_key
+        }
+
+
+    def _subscribed_alerts_as_trinity_list(self) -> List[str]:
+        return [str(alert) for alert in self.subscribed_alerts] if self.subscribed_alerts else []
+
+    async def _lazy_load(self, table: Table, consistent: bool = False) -> None:
+        if consistent or not self._loaded:
+            await self.get(table, consistent=consistent)
+
+    async def put(self, table: Table) -> None:
+        await table.put_item(
+            item={
+                self.primary_key_name: self.primary_key,
+                'region': self.region,
+                'subscribed_alerts': self._subscribed_alerts_as_trinity_list()
+            }
+        )
+
+    async def delete(self, table: Table) -> None:
+        if not self._loaded:
+            await self._lazy_load(table, consistent=True)
+        if self.subscribed_alerts:
+            for alert in self.subscribed_alerts:
+                await alert.remove_user(table, self)
+        await table.delete_item(
+            key={self.primary_key_name: self.primary_key},
+        )
+
+    async def get(self, table: Table, consistent: bool = False) -> bool:
+        try:
+            response = await table.get_item(
+                key=self.key,
+                consistent_read=consistent
+            )
+        except ItemNotFound:
+            return False
+
+        self.subscribed_alerts = []
+        for string_trinity in response['subscribed_alerts']:
+            self.subscribed_alerts.append(pdb.Alert.from_str(string_trinity))
+        self.region = Region(response['region'])
+        return True