78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
|
import logging
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Union
|
||
|
|
|
||
|
|
from modules.database import cursor
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class PyroUser:
|
||
|
|
"""Dataclass of DB entry of a user"""
|
||
|
|
|
||
|
|
__slots__ = ("id", "card", "locale")
|
||
|
|
|
||
|
|
id: int
|
||
|
|
card: Union[str, None]
|
||
|
|
locale: Union[str, None]
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def find(
|
||
|
|
cls, id: int, card: Union[str, None] = None, locale: Union[str, None] = None
|
||
|
|
):
|
||
|
|
"""Find user in database and create new record if user does not exist.
|
||
|
|
|
||
|
|
### Args:
|
||
|
|
* id (`int`): User's Telegram ID
|
||
|
|
* card (`Union[str, None]`, *optional*): User's card number. Defaults to `None`.
|
||
|
|
* locale (`Union[str, None]`, *optional*): User's locale. Defaults to `None`.
|
||
|
|
|
||
|
|
### Raises:
|
||
|
|
* `RuntimeError`: Raised when user entry after insertion could not be found.
|
||
|
|
|
||
|
|
### Returns:
|
||
|
|
* `PyroUser`: User with its database data.
|
||
|
|
"""
|
||
|
|
db_entry = cursor.execute(
|
||
|
|
"SELECT id, card, locale FROM users WHERE id = ?", (id,)
|
||
|
|
).fetchone()
|
||
|
|
|
||
|
|
if db_entry is None:
|
||
|
|
cursor.execute("INSERT INTO users VALUES (?, ?, ?)", (id, card, locale))
|
||
|
|
cursor.connection.commit()
|
||
|
|
db_entry = cursor.execute(
|
||
|
|
"SELECT id, card, locale FROM users WHERE id = ?", (id,)
|
||
|
|
).fetchone()
|
||
|
|
|
||
|
|
if db_entry is None:
|
||
|
|
raise RuntimeError("Could not find inserted user entry.")
|
||
|
|
|
||
|
|
return cls(*db_entry)
|
||
|
|
|
||
|
|
async def update_locale(self, locale: Union[str, None]) -> None:
|
||
|
|
"""Change user's locale stored in the database.
|
||
|
|
|
||
|
|
### Args:
|
||
|
|
* locale (`Union[str, None]`): New locale to be set.
|
||
|
|
"""
|
||
|
|
logger.debug("%s's locale has been set to %s", self.id, locale)
|
||
|
|
cursor.execute(
|
||
|
|
"UPDATE users SET locale = ? WHERE id = ?",
|
||
|
|
(locale, self.id),
|
||
|
|
)
|
||
|
|
cursor.connection.commit()
|
||
|
|
|
||
|
|
async def update_card(self, card: Union[str, None]) -> None:
|
||
|
|
"""Change user's card stored in the database.
|
||
|
|
|
||
|
|
### Args:
|
||
|
|
* card (`Union[str, None]`): New card to be set.
|
||
|
|
"""
|
||
|
|
logger.debug("%s's card has been set to %s", self.id, card)
|
||
|
|
cursor.execute(
|
||
|
|
"UPDATE users SET card = ? WHERE id = ?",
|
||
|
|
(card, self.id),
|
||
|
|
)
|
||
|
|
cursor.connection.commit()
|