mirror of
https://gitlab.com/MoonTestUse1/AdministrationItDepartmens.git
synced 2025-08-14 00:25:46 +02:00
Fix database
This commit is contained in:
@@ -14,7 +14,7 @@ class TestSettings(BaseSettings):
|
||||
POSTGRES_DB: str = "test_app"
|
||||
|
||||
# JWT
|
||||
SECRET_KEY: str = "test_secret_key"
|
||||
SECRET_KEY: str = "test_secret_key_super_secret_test_key_123"
|
||||
ALGORITHM: str = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
|
||||
|
||||
|
@@ -1,9 +1,8 @@
|
||||
"""Employee model"""
|
||||
from sqlalchemy import Column, Integer, String, Boolean, DateTime
|
||||
from sqlalchemy import Column, Integer, String, Boolean
|
||||
from sqlalchemy.orm import relationship
|
||||
from datetime import datetime
|
||||
|
||||
from ..database import Base
|
||||
from ..db.base_class import Base
|
||||
|
||||
class Employee(Base):
|
||||
"""Employee model"""
|
||||
@@ -17,6 +16,7 @@ class Employee(Base):
|
||||
hashed_password = Column(String, nullable=False)
|
||||
is_active = Column(Boolean, default=True)
|
||||
is_admin = Column(Boolean, default=False)
|
||||
created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
|
||||
|
||||
requests = relationship("Request", back_populates="employee")
|
||||
# Отношения
|
||||
requests = relationship("Request", back_populates="employee", cascade="all, delete-orphan")
|
||||
tokens = relationship("Token", back_populates="employee", cascade="all, delete-orphan")
|
@@ -1,17 +1,16 @@
|
||||
"""Token model"""
|
||||
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
|
||||
from sqlalchemy import Column, Integer, String, ForeignKey
|
||||
from sqlalchemy.orm import relationship
|
||||
from datetime import datetime
|
||||
|
||||
from ..database import Base
|
||||
from ..db.base_class import Base
|
||||
|
||||
class Token(Base):
|
||||
"""Token model"""
|
||||
__tablename__ = "tokens"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
token = Column(String, unique=True, index=True)
|
||||
employee_id = Column(Integer, ForeignKey("employees.id"), nullable=False)
|
||||
created_at = Column(DateTime, default=datetime.utcnow)
|
||||
|
||||
employee = relationship("Employee", backref="tokens")
|
||||
token = Column(String, unique=True, index=True, nullable=False)
|
||||
employee_id = Column(Integer, ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
|
||||
|
||||
# Отношения
|
||||
employee = relationship("Employee", back_populates="tokens")
|
@@ -5,22 +5,28 @@ from sqlalchemy.orm import Session
|
||||
from typing import Optional
|
||||
|
||||
from ..core.config import settings
|
||||
from ..core.test_config import test_settings
|
||||
from ..models.token import Token
|
||||
from ..schemas.auth import TokenData
|
||||
|
||||
def get_settings():
|
||||
"""Get settings based on environment"""
|
||||
return test_settings if test_settings.TESTING else settings
|
||||
|
||||
def create_access_token(data: dict) -> str:
|
||||
"""Create access token"""
|
||||
to_encode = data.copy()
|
||||
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
config = get_settings()
|
||||
expire = datetime.utcnow() + timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
|
||||
encoded_jwt = jwt.encode(to_encode, config.SECRET_KEY, algorithm=config.ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
def verify_token(token: str) -> Optional[int]:
|
||||
"""Verify token and return employee_id"""
|
||||
try:
|
||||
# Проверяем, что токен действителен
|
||||
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
|
||||
config = get_settings()
|
||||
payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
|
||||
employee_id = int(payload.get("sub"))
|
||||
if employee_id is None:
|
||||
return None
|
||||
@@ -46,7 +52,10 @@ def create_and_save_token(employee_id: int, db: Session) -> str:
|
||||
# Создаем токен
|
||||
access_token = create_access_token({"sub": str(employee_id)})
|
||||
|
||||
# Сохраняем токен в базу
|
||||
# Удаляем старые токены пользователя
|
||||
db.query(Token).filter(Token.employee_id == employee_id).delete()
|
||||
|
||||
# Сохраняем новый токен в базу
|
||||
db_token = Token(
|
||||
token=access_token,
|
||||
employee_id=employee_id
|
||||
|
Reference in New Issue
Block a user