wowtoken-app/wow-token-app-bot
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Once and for all reformat.

Browse Source 
Will be using black code formatter
This commit is contained in:
Emily Doherty
2024-12-08 17:07:26 -08:00
parent 3a06464c29
commit 34badf17eb
25 changed files with 359 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
token_bot/persistant_database/__init__.py
View File

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

6
token_bot/persistant_database/alert_category.py
View File

@@ -7,7 +7,9 @@ class AlertCategory(Enum):
CUSTOM = 3
@staticmethod
def from_str(category: str): # It gets mad when I use the Type[AlertCategory] as a type hint
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
@@ -16,4 +18,4 @@ class AlertCategory(Enum):
case "sp_add_button":
return AlertCategory.CUSTOM
case _:
return AlertCategory[category.upper()]
return AlertCategory[category.upper()]

41
token_bot/persistant_database/alert_schema.py
View File

@@ -11,7 +11,9 @@ import token_bot.persistant_database as pdb
class Alert:
def __init__(self, alert: pdb.AlertType, flavor: Flavor, region: Region, price: int = 0) -> None:
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
@@ -22,18 +24,18 @@ class Alert:
self.users: List[pdb.User] = []
@classmethod
def from_item(cls, primary_key: int, sort_key: str, users: List[int]) -> 'Alert':
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_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:
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))
@@ -61,20 +63,24 @@ class Alert:
def key(self) -> dict[str, str | int]:
return {
self.primary_key_name: self.primary_key,
self.sort_key_name: self.sort_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
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 == AlertType.SPECIFIC_PRICE:
raise NotImplementedError
else:
alert_type_str = ' '.join(self.alert_type.name.split("_"))
alert_type_str = " ".join(self.alert_type.name.split("_"))
return f"{alert_type_str.title()}"
async def _lazy_load(self, table: Table, consistent: bool = False) -> None:
@@ -95,19 +101,16 @@ class Alert:
item={
self.primary_key_name: self.primary_key,
self.sort_key_name: self.sort_key,
'users': user_ids
"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
)
response = await table.get_item(key=self.key, consistent_read=consistent)
except ItemNotFound:
return False
self.users = [pdb.User(int(user_id)) for user_id in response['users']]
self.users = [pdb.User(int(user_id)) for user_id in response["users"]]
self._loaded = True
return True
@@ -116,13 +119,17 @@ class Alert:
return self.users
async def add_user(self, table: Table, user: pdb.User, consistent: bool = False) -> None:
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:
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:

3
token_bot/persistant_database/alert_type.py
View File

@@ -1,5 +1,6 @@
from enum import Enum
class AlertType(Enum):
ALL_TIME_HIGH = 1
ALL_TIME_LOW = 2
@@ -37,4 +38,4 @@ class AlertType(Enum):
case "All Time Low":
return AlertType.ALL_TIME_LOW
case _:
return AlertType.SPECIFIC_PRICE
return AlertType.SPECIFIC_PRICE

5
token_bot/persistant_database/database.py
View File

@@ -8,5 +8,6 @@ from aiodynamo.http.aiohttp import AIOHTTP
class Database:
def __init__(self, session: aiohttp.ClientSession):
self.client = Client(AIOHTTP(session), Credentials.auto(), os.getenv('AWS_REGION'))
self.client = Client(
AIOHTTP(session), Credentials.auto(), os.getenv("AWS_REGION")
)

45
token_bot/persistant_database/user_schema.py
View File

@@ -8,7 +8,12 @@ from token_bot.token_database.region import Region
class User:
def __init__(self, user_id: int, region: Region = None, subscribed_alerts: List['pdb.Alert'] = None) -> None:
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
@@ -21,7 +26,9 @@ class User:
return hash(self.user_id)
@classmethod
def from_item(cls, primary_key: int, region: Region, subscribed_alerts: List[str]) -> 'User':
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)
@@ -31,17 +38,18 @@ class User:
@property
def primary_key_name(self) -> str:
return 'user_id'
return "user_id"
@property
def key(self) -> Dict[str, str]:
return {
self.primary_key_name: self.primary_key
}
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 []
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:
@@ -51,8 +59,8 @@ class User:
await table.put_item(
item={
self.primary_key_name: self.primary_key,
'region': self.region,
'subscribed_alerts': self._subscribed_alerts_as_trinity_list()
"region": self.region,
"subscribed_alerts": self._subscribed_alerts_as_trinity_list(),
}
)
@@ -65,26 +73,27 @@ class User:
async def get(self, table: Table, consistent: bool = False) -> bool:
try:
response = await table.get_item(
key=self.key,
consistent_read=consistent
)
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']:
for string_trinity in response["subscribed_alerts"]:
self.subscribed_alerts.append(pdb.Alert.from_str(string_trinity))
self.region = Region(response['region'])
self.region = Region(response["region"])
return True
async def add_alert(self, table: Table, alert: 'pdb.Alert', consistent: bool = False) -> None:
async def add_alert(
self, table: Table, alert: "pdb.Alert", consistent: bool = False
) -> None:
await self._lazy_load(table, consistent=consistent)
if alert not in self.subscribed_alerts:
self.subscribed_alerts.append(alert)
await self.put(table)
async def remove_alert(self, table: Table, alert: 'pdb.Alert', consistent: bool = True) -> None:
async def remove_alert(
self, table: Table, alert: "pdb.Alert", consistent: bool = True
) -> None:
await self._lazy_load(table, consistent=consistent)
if alert in self.subscribed_alerts:
self.subscribed_alerts.remove(alert)