50 lines
2.0 KiB
Python
50 lines
2.0 KiB
Python
from dataclasses import dataclass
|
||
from typing import List
|
||
|
||
from cachetools import TTLCache, cached
|
||
from sqlalchemy import String, func
|
||
from sqlalchemy.dialects.mssql import TINYINT
|
||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||
|
||
from src.entity import account
|
||
from src.entity.database import db
|
||
|
||
|
||
# 用户
|
||
@dataclass
|
||
class User(db.Base):
|
||
__tablename__ = 'ky_user'
|
||
|
||
id: Mapped[db.int_pk] = mapped_column(comment="唯一键")
|
||
username: Mapped[db.str_required_unique] = mapped_column(comment="账号")
|
||
password: Mapped[db.str_required] = mapped_column(comment="密码")
|
||
name: Mapped[str] = mapped_column(String(32), nullable=True, comment="别名")
|
||
email: Mapped[str] = mapped_column(String(32), nullable=False, comment="邮箱")
|
||
bot_token: Mapped[str] = mapped_column(String(64), nullable=True, comment="飞机机器人id")
|
||
group_id: Mapped[str] = mapped_column(String(32), nullable=True, comment="消息群组id")
|
||
chat_id: Mapped[str] = mapped_column(String(32), nullable=True, comment="消息个人id")
|
||
telegram_ids: Mapped[str] = mapped_column(String(128), nullable=True, comment="telegram_ids")
|
||
status: Mapped[int] = mapped_column(TINYINT, nullable=False, comment='账号状态,0禁用，1启用', default=0)
|
||
accounts: Mapped[List['account.Account']] = relationship('account.Account', back_populates='user', lazy=False)
|
||
|
||
|
||
def get_user_by_username_and_password(username: str, password: str) -> User:
|
||
with db.Session() as session:
|
||
user = session.query(User).filter(
|
||
User.username == username and User.password == password and User.status == 1).one()
|
||
return user
|
||
|
||
|
||
def get_all_users():
|
||
with db.Session() as session:
|
||
users = session.query(User).filter(User.status == 1).all()
|
||
return users
|
||
|
||
|
||
@cached(TTLCache(maxsize=100, ttl=3600))
|
||
def get_user_by_telegram_id(telegram_id: int) -> User:
|
||
with db.Session() as session:
|
||
user = session.query(User).filter(
|
||
User.status == 1 and func.find_in_set(telegram_id, User.telegram_ids) > 0).one()
|
||
return user
|