90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
import time
|
|
from typing import cast, Optional
|
|
|
|
from sqlalchemy import select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from defs import sqlite
|
|
from models.models.user import User
|
|
|
|
|
|
class UserAction:
|
|
@staticmethod
|
|
async def add_user(user: User):
|
|
async with sqlite.session() as session:
|
|
session = cast(AsyncSession, session)
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
@staticmethod
|
|
async def get_user_by_username(username: str) -> Optional[User]:
|
|
async with sqlite.session() as session:
|
|
session = cast(AsyncSession, session)
|
|
statement = select(User).where(User.username == username)
|
|
results = await session.exec(statement)
|
|
return user[0] if (user := results.first()) else None
|
|
|
|
@staticmethod
|
|
async def get_user_by_id(uid: int) -> Optional[User]:
|
|
async with sqlite.session() as session:
|
|
session = cast(AsyncSession, session)
|
|
statement = select(User).where(User.uid == uid)
|
|
results = await session.exec(statement)
|
|
return user[0] if (user := results.first()) else None
|
|
|
|
@staticmethod
|
|
async def check_admin(uid: int) -> bool:
|
|
user = await UserAction.get_user_by_id(uid)
|
|
return user.is_admin if user else False
|
|
|
|
@staticmethod
|
|
async def update_user(old_user: User, new_user: User = None):
|
|
if new_user:
|
|
old_user.username = new_user.username
|
|
old_user.password = new_user.password
|
|
old_user.is_admin = new_user.is_admin
|
|
old_user.register_time = new_user.register_time
|
|
old_user.last_login_time = new_user.last_login_time
|
|
old_user.session = new_user.session
|
|
async with sqlite.session() as session:
|
|
session = cast(AsyncSession, session)
|
|
session.add(old_user)
|
|
await session.commit()
|
|
await session.refresh(old_user)
|
|
|
|
@staticmethod
|
|
async def add_or_update_user(user: User):
|
|
if old_user := await UserAction.get_user_by_username(user.username):
|
|
await UserAction.update_user(old_user, user)
|
|
else:
|
|
await UserAction.add_user(user)
|
|
|
|
@staticmethod
|
|
async def change_user_password(username: str, password: str) -> bool:
|
|
user = await UserAction.get_user_by_username(username)
|
|
if not user:
|
|
return False
|
|
user.password = password
|
|
await UserAction.update_user(user)
|
|
return True
|
|
|
|
@staticmethod
|
|
def gen_new_user(
|
|
username: str,
|
|
password: str,
|
|
is_admin: bool = False,
|
|
register_time: int = 0,
|
|
last_login_time: int = 0,
|
|
session: str = "",
|
|
) -> User:
|
|
if not register_time:
|
|
register_time = int(time.time())
|
|
return User(
|
|
username=username,
|
|
password=password,
|
|
is_admin=is_admin,
|
|
register_time=register_time,
|
|
last_login_time=last_login_time,
|
|
session=session,
|
|
)
|