1fbf9185a2
- Добавлена колонка 'Тип' во все таблицы истории сборок - Для push операций отображается registry вместо платформ - Сохранение пользователя при создании push лога - Исправлена ошибка с logger в push_docker_image endpoint - Улучшено отображение истории сборок с визуальными индикаторами
90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
"""
|
||
Сервис для работы с пользователями
|
||
Автор: Сергей Антропов
|
||
Сайт: https://devops.org.ru
|
||
"""
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from sqlalchemy import select, update
|
||
from app.models.user import User
|
||
from app.auth.security import get_password_hash, verify_password
|
||
from typing import Optional
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class UserService:
|
||
"""Сервис для работы с пользователями"""
|
||
|
||
@staticmethod
|
||
async def get_user_by_username(db: AsyncSession, username: str) -> Optional[User]:
|
||
"""Получение пользователя по имени"""
|
||
result = await db.execute(select(User).where(User.username == username))
|
||
return result.scalar_one_or_none()
|
||
|
||
@staticmethod
|
||
async def get_user_by_id(db: AsyncSession, user_id: int) -> Optional[User]:
|
||
"""Получение пользователя по ID"""
|
||
result = await db.execute(select(User).where(User.id == user_id))
|
||
return result.scalar_one_or_none()
|
||
|
||
@staticmethod
|
||
async def create_user(
|
||
db: AsyncSession,
|
||
username: str,
|
||
password: str,
|
||
is_superuser: bool = False
|
||
) -> User:
|
||
"""Создание нового пользователя"""
|
||
hashed_password = get_password_hash(password)
|
||
user = User(
|
||
username=username,
|
||
hashed_password=hashed_password,
|
||
is_active=True,
|
||
is_superuser=is_superuser
|
||
)
|
||
db.add(user)
|
||
await db.commit()
|
||
await db.refresh(user)
|
||
return user
|
||
|
||
@staticmethod
|
||
async def update_password(
|
||
db: AsyncSession,
|
||
user: User,
|
||
new_password: str
|
||
) -> User:
|
||
"""Обновление пароля пользователя"""
|
||
user.hashed_password = get_password_hash(new_password)
|
||
await db.commit()
|
||
await db.refresh(user)
|
||
return user
|
||
|
||
@staticmethod
|
||
async def verify_user_password(user: User, password: str) -> bool:
|
||
"""Проверка пароля пользователя"""
|
||
return verify_password(password, user.hashed_password)
|
||
|
||
@staticmethod
|
||
async def ensure_admin_user(db: AsyncSession) -> User:
|
||
"""Создание пользователя admin по умолчанию, если его нет"""
|
||
admin_user = await UserService.get_user_by_username(db, "admin")
|
||
|
||
if not admin_user:
|
||
logger.info("Создание пользователя admin по умолчанию")
|
||
admin_user = await UserService.create_user(
|
||
db,
|
||
username="admin",
|
||
password="admin",
|
||
is_superuser=True
|
||
)
|
||
else:
|
||
# Проверяем, что пароль правильно хеширован (bcrypt хеш имеет длину 60 символов)
|
||
if not admin_user.hashed_password or len(admin_user.hashed_password) < 50:
|
||
# Если пароль не хеширован или хеш неправильный, обновляем его
|
||
logger.info("Обновление пароля пользователя admin")
|
||
admin_user = await UserService.update_password(db, admin_user, "admin")
|
||
|
||
return admin_user
|