148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""User service for VoIdea."""
|
|
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.models.user import User
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
def _check_owner(target: User) -> None:
|
|
"""Protect owner from deletion, suspension, or role change."""
|
|
if target.is_owner:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Владелец системы защищён от изменений",
|
|
)
|
|
|
|
|
|
class UserService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_by_id(self, user_id: str) -> Optional[User]:
|
|
result = await self.db.execute(
|
|
select(User).where(User.id == user_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_email(self, email: str) -> Optional[User]:
|
|
result = await self.db.execute(
|
|
select(User).where(User.email == email)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def update_profile(
|
|
self, user_id: str, display_name: Optional[str] = None,
|
|
avatar_url: Optional[str] = None,
|
|
) -> Optional[User]:
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
if display_name is not None:
|
|
user.display_name = display_name
|
|
if avatar_url is not None:
|
|
user.avatar_url = avatar_url
|
|
|
|
await self.db.commit()
|
|
await self.db.refresh(user)
|
|
return user
|
|
|
|
async def delete(self, user_id: str) -> bool:
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return False
|
|
|
|
_check_owner(user)
|
|
user.is_active = False
|
|
await self.db.commit()
|
|
return True
|
|
|
|
async def hard_delete(self, user_id: str) -> bool:
|
|
"""Permanently delete user from DB (admin only, never for owner)."""
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return False
|
|
|
|
_check_owner(user)
|
|
await self.db.delete(user)
|
|
await self.db.commit()
|
|
return True
|
|
|
|
async def list_users(
|
|
self, skip: int = 0, limit: int = 100, search: Optional[str] = None
|
|
) -> list[User]:
|
|
query = select(User)
|
|
if search:
|
|
query = query.where(
|
|
or_(
|
|
User.display_name.ilike(f"%{search}%"),
|
|
User.email.ilike(f"%{search}%"),
|
|
)
|
|
)
|
|
result = await self.db.execute(
|
|
query.order_by(User.created_at.desc()).offset(skip).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
async def update_role(self, user_id: str, is_superuser: bool) -> Optional[User]:
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
_check_owner(user)
|
|
user.is_superuser = is_superuser
|
|
await self.db.commit()
|
|
await self.db.refresh(user)
|
|
return user
|
|
|
|
# ── v2.0 admin methods ──
|
|
|
|
async def admin_update_user(
|
|
self, user_id: str, updates: dict[str, Any]
|
|
) -> Optional[User]:
|
|
"""Admin: update user fields with owner protection."""
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
|
|
is_owner_change = any(k in updates for k in ("is_owner", "role"))
|
|
if is_owner_change or "is_active" in updates:
|
|
_check_owner(user)
|
|
|
|
for key, value in updates.items():
|
|
if hasattr(user, key) and key not in ("id", "password_hash", "created_at"):
|
|
if key == "permissions" and value is not None:
|
|
value = {k: bool(v) for k, v in value.items()}
|
|
setattr(user, key, value)
|
|
|
|
await self.db.commit()
|
|
await self.db.refresh(user)
|
|
return user
|
|
|
|
async def set_owner_by_email(self, email: str) -> Optional[User]:
|
|
"""Mark a user as system owner by email (called at startup / registration)."""
|
|
if not settings.system_owner_email:
|
|
return None
|
|
if email != settings.system_owner_email:
|
|
return None
|
|
user = await self.get_by_email(email)
|
|
if not user:
|
|
return None
|
|
if user.is_owner:
|
|
return user
|
|
|
|
user.is_owner = True
|
|
user.role = "owner"
|
|
user.is_superuser = True
|
|
user.is_active = True
|
|
await self.db.commit()
|
|
await self.db.refresh(user)
|
|
return user
|