aiomysql refactor

This commit is contained in:
David Rodenkirchen
2024-09-03 14:30:32 +02:00
parent a9597b5c4f
commit 30b32a4c02
24 changed files with 901 additions and 755 deletions
+32 -7
View File
@@ -1,22 +1,47 @@
from rio import Column, Component, event, Text, TextStyle, Button, Color, Spacer, Revealer, Row
from asyncio import sleep
from typing import Optional
from rio import Column, Component, event, Text, TextStyle, Button, Color, Spacer, Revealer, Row, ProgressCircle
from src.ez_lan_manager import ConfigurationService, UserService, AccountingService
from src.ez_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ez_lan_manager.pages import BasePage
from src.ez_lan_manager.types.SessionStorage import SessionStorage
from src.ez_lan_manager.types.Transaction import Transaction
from src.ez_lan_manager.types.User import User
class AccountPage(Component):
user: Optional[User] = None
balance: Optional[int] = None
transaction_history: list[Transaction] = list()
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Guthabenkonto")
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
self.balance = await self.session[AccountingService].get_balance(self.user.user_id)
self.transaction_history = await self.session[AccountingService].get_transaction_history(self.user.user_id)
async def _on_banking_info_press(self):
self.banking_info_revealer.is_open = not self.banking_info_revealer.is_open
def build(self) -> Component:
user = self.session[UserService].get_user(self.session[SessionStorage].user_id)
a_s = self.session[AccountingService]
if not self.user and not self.balance:
return BasePage(
content=Column(
MainViewContentBox(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=2,
margin_bottom=2
)
),
align_y = 0,
)
)
self.banking_info_revealer = Revealer(
header=None,
content=Column(
@@ -45,7 +70,7 @@ class AccountPage(Component):
align_x=0.2
),
Text(
f"AUFLADUNG - {user.user_id} - {user.user_name}",
f"AUFLADUNG - {self.user.user_id} - {self.user.user_name}",
style=TextStyle(
fill=self.session.theme.neutral_color
),
@@ -73,7 +98,7 @@ class AccountPage(Component):
)
)
for transaction in sorted(self.session[AccountingService].get_transaction_history(user.user_id), key=lambda t: t.transaction_date, reverse=True):
for transaction in sorted(self.transaction_history, key=lambda t: t.transaction_date, reverse=True):
transaction_history.add(
Row(
Text(
@@ -89,7 +114,7 @@ class AccountPage(Component):
align_x=0
),
Text(
f"{'-' if transaction.is_debit else '+'}{a_s.make_euro_string_from_int(transaction.value)}",
f"{'-' if transaction.is_debit else '+'}{AccountingService.make_euro_string_from_int(transaction.value)}",
style=TextStyle(
fill=self.session.theme.danger_color if transaction.is_debit else self.session.theme.success_color,
font_size=0.8
@@ -106,7 +131,7 @@ class AccountPage(Component):
content=Column(
MainViewContentBox(
content=Text(
f"Kontostand: {a_s.make_euro_string_from_int(a_s.get_balance(user.user_id))}",
f"Kontostand: {AccountingService.make_euro_string_from_int(self.balance)}",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
+1 -6
View File
@@ -4,7 +4,7 @@ from typing import * # type: ignore
from rio import Component, event, Spacer, Card, Container, Column, Row, TextStyle, Color, Text
from src.ez_lan_manager import ConfigurationService, DatabaseService
from src.ez_lan_manager import ConfigurationService
from src.ez_lan_manager.components.DesktopNavigation import DesktopNavigation
class BasePage(Component):
@@ -14,11 +14,6 @@ class BasePage(Component):
async def on_window_size_change(self):
await self.force_refresh()
@event.on_populate
async def check_db_connection(self):
if not self.session[DatabaseService].is_connected:
self.session.navigate_to("./db-error")
def build(self) -> Component:
if self.content is None:
content = Spacer()
+37 -19
View File
@@ -1,16 +1,19 @@
from rio import Column, Component, event, TextStyle, Text, Spacer, Revealer, SwitcherBar, SwitcherBarChangeEvent
from typing import Optional
from rio import Column, Component, event, TextStyle, Text, Spacer, Revealer, SwitcherBar, SwitcherBarChangeEvent, ProgressCircle
from src.ez_lan_manager import ConfigurationService, CateringService
from src.ez_lan_manager.components.CateringSelectionItem import CateringSelectionItem
from src.ez_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ez_lan_manager.components.ShoppingCartAndOrders import ShoppingCartAndOrders
from src.ez_lan_manager.pages import BasePage
from src.ez_lan_manager.types.CateringMenuItem import CateringMenuItemCategory
from src.ez_lan_manager.types.CateringMenuItem import CateringMenuItemCategory, CateringMenuItem
from src.ez_lan_manager.types.SessionStorage import SessionStorage
class CateringPage(Component):
show_cart = True
all_menu_items: Optional[list[CateringMenuItem]] = None
def __post_init__(self) -> None:
self.session[SessionStorage].subscribe_to_logged_in_or_out_event(self.__class__.__name__, self.on_user_logged_in_status_changed)
@@ -18,6 +21,8 @@ class CateringPage(Component):
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Catering")
self.all_menu_items = await self.session[CateringService].get_menu()
async def on_user_logged_in_status_changed(self) -> None:
await self.force_refresh()
@@ -25,9 +30,13 @@ class CateringPage(Component):
async def on_switcher_bar_changed(self, _: SwitcherBarChangeEvent) -> None:
await self.shopping_cart_and_orders.switch()
@staticmethod
def get_menu_items_by_category(all_menu_items: list[CateringMenuItem], category: Optional[CateringMenuItemCategory]) -> list[CateringMenuItem]:
return list(filter(lambda item: item.category == category, all_menu_items))
def build(self) -> Component:
user_id = self.session[SessionStorage].user_id
catering_service = self.session[CateringService]
self.shopping_cart_and_orders = ShoppingCartAndOrders()
switcher_bar = SwitcherBar(
values=["cart", "orders"],
@@ -58,12 +67,14 @@ class CateringPage(Component):
)
) if user_id else Spacer()
return BasePage(
content=Column(
# SHOPPING CART
shopping_cart_and_orders_container,
# ITEM SELECTION
MainViewContentBox(
menu = [MainViewContentBox(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=2,
margin_bottom=2
)
)] if not self.all_menu_items else [MainViewContentBox(
Revealer(
header="Snacks",
content=Column(
@@ -75,7 +86,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.SNACK))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.SNACK))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -97,7 +108,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.BREAKFAST))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.BREAKFAST))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -119,7 +130,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.MAIN_COURSE))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.MAIN_COURSE))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -141,7 +152,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.DESSERT))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.DESSERT))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -163,7 +174,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.BEVERAGE_NON_ALCOHOLIC))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.BEVERAGE_NON_ALCOHOLIC))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -185,7 +196,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.BEVERAGE_ALCOHOLIC))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.BEVERAGE_ALCOHOLIC))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -207,7 +218,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.BEVERAGE_COCKTAIL))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.BEVERAGE_COCKTAIL))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -229,7 +240,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.BEVERAGE_SHOT))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.BEVERAGE_SHOT))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -251,7 +262,7 @@ class CateringPage(Component):
is_sensitive=(user_id is not None) and not catering_menu_item.is_disabled,
additional_info=catering_menu_item.additional_info,
is_grey=idx % 2 == 0
) for idx, catering_menu_item in enumerate(catering_service.get_menu(CateringMenuItemCategory.NON_FOOD))],
) for idx, catering_menu_item in enumerate(self.get_menu_items_by_category(self.all_menu_items, CateringMenuItemCategory.NON_FOOD))],
),
header_style=TextStyle(
fill=self.session.theme.background_color,
@@ -260,7 +271,14 @@ class CateringPage(Component):
margin=1,
align_y=0.5
)
),
)]
return BasePage(
content=Column(
# SHOPPING CART
shopping_cart_and_orders_container,
# ITEM SELECTION
*menu,
align_y=0
)
)
+8 -6
View File
@@ -1,4 +1,5 @@
from datetime import datetime, timedelta
from typing import Optional
from rio import Text, Column, TextStyle, Component, event, TextInput, MultiLineTextInput, Row, Button
@@ -7,6 +8,7 @@ from src.ez_lan_manager.components.AnimatedText import AnimatedText
from src.ez_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ez_lan_manager.pages import BasePage
from src.ez_lan_manager.types.SessionStorage import SessionStorage
from src.ez_lan_manager.types.User import User
class ContactPage(Component):
@@ -14,10 +16,15 @@ class ContactPage(Component):
# Using list to bypass this behavior
last_message_sent: list[datetime] = [datetime(day=1, month=1, year=2000)]
display_printing: list[bool] = [False]
user: Optional[User] = None
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Kontakt")
if self.session[SessionStorage].user_id is not None:
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
else:
self.user = None
async def on_send_pressed(self) -> None:
error_msg = ""
@@ -51,11 +58,6 @@ class ContactPage(Component):
await self.animated_text.display_text(True, "Nachricht erfolgreich gesendet!")
def build(self) -> Component:
if self.session[SessionStorage].user_id is not None:
user = self.session[UserService].get_user(self.session[SessionStorage].user_id)
else:
user = None
self.animated_text = AnimatedText(
margin_top = 2,
margin_bottom = 1,
@@ -64,7 +66,7 @@ class ContactPage(Component):
self.email_input = TextInput(
label="E-Mail Adresse",
text="" if not user else user.user_mail,
text="" if not self.user else self.user.user_mail,
margin_left=1,
margin_right=1,
margin_bottom=1,
+6 -6
View File
@@ -15,12 +15,12 @@ class DbErrorPage(Component):
async def on_window_size_change(self) -> None:
await self.force_refresh()
@event.on_mount
async def retry_db_connect(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Fehler")
while not self.session[DatabaseService].is_connected:
await sleep(2)
self.session.navigate_to("./")
# @event.on_mount
# async def retry_db_connect(self) -> None:
# await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Fehler")
# while not self.session[DatabaseService].is_connected:
# await sleep(2)
# self.session.navigate_to("./")
def build(self) -> Component:
content = Card(
+30 -13
View File
@@ -3,7 +3,8 @@ from hashlib import sha256
from typing import Optional
from from_root import from_root
from rio import Column, Component, event, Text, TextStyle, Button, Color, Row, TextInput, Image, TextInputChangeEvent, NoFileSelectedError
from rio import Column, Component, event, Text, TextStyle, Button, Color, Row, TextInput, Image, TextInputChangeEvent, NoFileSelectedError, \
ProgressCircle
from email_validator import validate_email, EmailNotValidError
from src.ez_lan_manager import ConfigurationService, UserService
@@ -15,6 +16,8 @@ from src.ez_lan_manager.types.User import User
class EditProfilePage(Component):
user: Optional[User] = None
pfp: Optional[bytes] = None
@staticmethod
def optional_date_to_str(d: Optional[date]) -> str:
if not d:
@@ -24,6 +27,8 @@ class EditProfilePage(Component):
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Profil bearbeiten")
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
self.pfp = await self.session[UserService].get_profile_picture(self.user.user_id)
def on_email_changed(self, change_event: TextInputChangeEvent) -> None:
try:
@@ -58,7 +63,7 @@ class EditProfilePage(Component):
return
image_data = await new_pfp.read_bytes()
self.session[UserService].set_profile_picture(self.session[SessionStorage].user_id, image_data)
await self.session[UserService].set_profile_picture(self.session[SessionStorage].user_id, image_data)
self.pfp_image_container.image = image_data
await self.animated_text.display_text(True, "Gespeichert!")
@@ -72,7 +77,7 @@ class EditProfilePage(Component):
await self.animated_text.display_text(False, "Passwörter nicht gleich!")
return
user: User = self.session[UserService].get_user(self.session[SessionStorage].user_id)
user: User = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
user.user_mail = self.email_input.text
if len(self.birthday_input.text) == 0:
@@ -86,12 +91,24 @@ class EditProfilePage(Component):
if len(self.new_pw_1_input.text.strip()) > 0:
user.user_password = sha256(self.new_pw_1_input.text.encode(encoding="utf-8")).hexdigest()
self.session[UserService].update_user(user)
await self.session[UserService].update_user(user)
await self.animated_text.display_text(True, "Gespeichert!")
def build(self) -> Component:
user = self.session[UserService].get_user(self.session[SessionStorage].user_id)
pfp = self.session[UserService].get_profile_picture(user.user_id)
if not self.user:
return BasePage(
content=Column(
MainViewContentBox(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=2,
margin_bottom=2
)
),
align_y = 0
)
)
self.animated_text = AnimatedText(
margin_top=2,
@@ -101,7 +118,7 @@ class EditProfilePage(Component):
self.email_input = TextInput(
label="E-Mail Adresse",
text=user.user_mail,
text=self.user.user_mail,
margin_left=1,
margin_right=1,
margin_bottom=1,
@@ -110,20 +127,20 @@ class EditProfilePage(Component):
)
self.first_name_input = TextInput(
label="Vorname",
text=user.user_first_name,
text=self.user.user_first_name,
margin_left=1,
margin_right=1,
grow_x=True
)
self.last_name_input = TextInput(
label="Nachname",
text=user.user_last_name,
text=self.user.user_last_name,
margin_right=1,
grow_x=True
)
self.birthday_input = TextInput(
label="Geburtstag (TT.MM.JJJJ)",
text=self.optional_date_to_str(user.user_birth_day),
text=self.optional_date_to_str(self.user.user_birth_day),
margin_left=1,
margin_right=1,
margin_bottom=1,
@@ -150,7 +167,7 @@ class EditProfilePage(Component):
)
self.pfp_image_container = Image(
from_root("src/ez_lan_manager/assets/img/anon_pfp.png") if pfp is None else pfp,
from_root("src/ez_lan_manager/assets/img/anon_pfp.png") if self.pfp is None else self.pfp,
align_x=0.5,
min_width=10,
min_height=10,
@@ -176,8 +193,8 @@ class EditProfilePage(Component):
on_press=self.upload_new_pfp
),
Row(
TextInput(label="Deine User-ID", text=user.user_id, is_sensitive=False, margin_left=1, grow_x=False),
TextInput(label="Dein Nickname", text=user.user_name, is_sensitive=False, margin_left=1, margin_right=1, grow_x=True),
TextInput(label="Deine User-ID", text=self.user.user_id, is_sensitive=False, margin_left=1, grow_x=False),
TextInput(label="Dein Nickname", text=self.user.user_name, is_sensitive=False, margin_left=1, margin_right=1, grow_x=True),
margin_bottom=1
),
self.email_input,
+2 -2
View File
@@ -24,11 +24,11 @@ class ForgotPasswordPage(Component):
lan_info = self.session[ConfigurationService].get_lan_info()
user_service = self.session[UserService]
mailing_service = self.session[MailingService]
user = user_service.get_user(self.email_input.text.strip())
user = await user_service.get_user(self.email_input.text.strip())
if user is not None:
new_password = "".join(choices(user_service.ALLOWED_USER_NAME_SYMBOLS, k=16))
user.user_password = sha256(new_password.encode(encoding="utf-8")).hexdigest()
user_service.update_user(user)
await user_service.update_user(user)
await mailing_service.send_email(
subject=f"Dein neues Passwort für {lan_info.name}",
body=f"Du hast für den EZ-LAN Manager der {lan_info.name} ein neues Passwort angefragt. "
+22 -9
View File
@@ -5,37 +5,50 @@ from rio import Column, Component, event, TextStyle, Text, Button, Row, TextInpu
from src.ez_lan_manager import ConfigurationService, UserService, TicketingService, SeatingService
from src.ez_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ez_lan_manager.pages import BasePage
from src.ez_lan_manager.types.Seat import Seat
from src.ez_lan_manager.types.User import User
class GuestsPage(Component):
table_elements: list[Button] = []
users_with_tickets: list[User] = []
users_with_seats: dict[User, Seat] = {}
user_filter: Optional[str] = None
def __post_init__(self) -> None:
user_service = self.session[UserService]
all_users = user_service.get_all_users()
ticketing_service = self.session[TicketingService]
self.users_with_tickets = list(filter(lambda user: ticketing_service.get_user_ticket(user.user_id) is not None, all_users))
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Teilnehmer")
user_service = self.session[UserService]
all_users = await user_service.get_all_users()
ticketing_service = self.session[TicketingService]
seating_service = self.session[SeatingService]
u_w_t = []
u_w_s = {}
for user in all_users:
ticket = await ticketing_service.get_user_ticket(user.user_id)
seat = await seating_service.get_user_seat(user.user_id)
if ticket is not None:
u_w_t.append(user)
if seat is not None:
u_w_s[user] = seat
self.users_with_tickets = u_w_t
self.users_with_seats = u_w_s
def on_searchbar_content_change(self, change_event: TextInputChangeEvent) -> None:
self.user_filter = change_event.text
def build(self) -> Component:
seating_service = self.session[SeatingService]
if self.user_filter:
users = [user for user in self.users_with_tickets if self.user_filter.lower() in user.user_name or self.user_filter.lower() in str(user.user_id)]
else:
users = self.users_with_tickets
self.table_elements.clear()
for idx, user in enumerate(users):
seat = seating_service.get_user_seat(user.user_id)
try:
seat = self.users_with_seats[user]
except KeyError:
seat = None
self.table_elements.append(
Button(
content=Row(Text(text=f"{user.user_id:0>4}", align_x=0, margin_right=1), Text(text=user.user_name, grow_x=True, wrap="ellipsize"), Text(text="-" if seat is None else seat.seat_id, align_x=1)),
+1 -1
View File
@@ -12,7 +12,7 @@ class NewsPage(Component):
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Neuigkeiten")
self.news_posts = self.session[NewsService].get_news()[:8]
self.news_posts = (await self.session[NewsService].get_news())[:8]
def build(self) -> Component:
posts = [NewsPost(
+2 -2
View File
@@ -62,13 +62,13 @@ class RegisterPage(Component):
mailing_service = self.session[MailingService]
lan_info = self.session[ConfigurationService].get_lan_info()
if user_service.get_user(self.email_input.text) is not None or user_service.get_user(self.user_name_input.text) is not None:
if await user_service.get_user(self.email_input.text) is not None or await user_service.get_user(self.user_name_input.text) is not None:
await self.animated_text.display_text(False, "Benutzername oder E-Mail bereits regestriert!")
self.submit_button.is_loading = False
return
try:
new_user = user_service.create_user(self.user_name_input.text, self.email_input.text, self.pw_1.text)
new_user = await user_service.create_user(self.user_name_input.text, self.email_input.text, self.pw_1.text)
if not new_user:
raise RuntimeError("User could not be created")
except Exception as e: