57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
|
from sqlalchemy import Column, Integer, String, DateTime, create_engine
|
||
|
|
from sqlalchemy.orm import declarative_base, sessionmaker
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Optional, List
|
||
|
|
from bot.auth.crypto import encrypt_password
|
||
|
|
|
||
|
|
Base = declarative_base()
|
||
|
|
class User(Base):
|
||
|
|
__tablename__ = 'users'
|
||
|
|
id = Column(Integer, primary_key=True)
|
||
|
|
telegram_id = Column(Integer, unique=True, nullable=False)
|
||
|
|
name = Column(String, nullable=False)
|
||
|
|
token = Column(String, nullable=False)
|
||
|
|
refresh_token = Column(String, nullable=False)
|
||
|
|
expires_in = Column(String, nullable=False)
|
||
|
|
password = Column(String, nullable=False)
|
||
|
|
last_token_update = Column(DateTime, default=datetime.now())
|
||
|
|
|
||
|
|
# Подключение к базе данных SQLite
|
||
|
|
engine = create_engine('sqlite:///reminder_bot.db')
|
||
|
|
Base.metadata.create_all(engine)
|
||
|
|
Session = sessionmaker(bind=engine)
|
||
|
|
session = Session()
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
class UserManager:
|
||
|
|
def __init__(self):
|
||
|
|
self.session = Session()
|
||
|
|
|
||
|
|
def save_user(self, telegram_id: int, login: str, token_response, password: str):
|
||
|
|
"""Сохраняет нового пользователя в базу данных"""
|
||
|
|
new_user = User(
|
||
|
|
telegram_id=telegram_id,
|
||
|
|
name=login,
|
||
|
|
token=token_response['token'],
|
||
|
|
refresh_token=token_response['refresh_token'],
|
||
|
|
expires_in=token_response['expires_in'],
|
||
|
|
password=self.encrypt_password(password)
|
||
|
|
)
|
||
|
|
self.session.add(new_user)
|
||
|
|
self.session.commit()
|
||
|
|
|
||
|
|
def get_all_users(self) -> List[User]:
|
||
|
|
"""Получает всех пользователей из базы данных"""
|
||
|
|
return self.session.query(User).all()
|
||
|
|
|
||
|
|
def get_user_by_telegram_id(self, telegram_id: int) -> Optional[User]:
|
||
|
|
"""Получает пользователя по telegram_id"""
|
||
|
|
return self.session.query(User).filter_by(telegram_id=telegram_id).first()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def encrypt_password(password: str) -> str:
|
||
|
|
return encrypt_password(password)
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
self.session.close()
|