165 lines
4.6 KiB
Python
165 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from app.core.database import session_scope
|
|
from app.modules.users.models import User, UserProfile
|
|
|
|
ALLOWED_ROLES = {"user", "admin"}
|
|
ALLOWED_STATUSES = {"pending", "active", "blocked"}
|
|
|
|
|
|
def _detach(db, instance):
|
|
db.refresh(instance)
|
|
db.expunge(instance)
|
|
return instance
|
|
|
|
|
|
def create_user(
|
|
email: str,
|
|
password_hash: str,
|
|
role: str = "user",
|
|
is_superuser: bool = False,
|
|
status: str = "pending",
|
|
) -> User:
|
|
if role not in ALLOWED_ROLES:
|
|
raise ValueError("INVALID_ROLE")
|
|
if status not in ALLOWED_STATUSES:
|
|
raise ValueError("INVALID_STATUS")
|
|
if is_superuser and role != "admin":
|
|
raise ValueError("SUPERUSER_REQUIRES_ADMIN")
|
|
with session_scope() as db:
|
|
user = User(
|
|
id=str(uuid4()),
|
|
email=email.lower(),
|
|
password_hash=password_hash,
|
|
role=role,
|
|
is_superuser=is_superuser,
|
|
status=status,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
profile = UserProfile(user_id=user.id, display_name=email.split("@")[0])
|
|
db.add(profile)
|
|
db.flush()
|
|
return _detach(db, user)
|
|
|
|
|
|
def get_user_by_email(email: str) -> User | None:
|
|
with session_scope() as db:
|
|
user = db.scalar(select(User).where(User.email == email.lower()))
|
|
if not user:
|
|
return None
|
|
return _detach(db, user)
|
|
|
|
|
|
def get_user_by_id(user_id: str) -> User | None:
|
|
with session_scope() as db:
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
return None
|
|
return _detach(db, user)
|
|
|
|
|
|
def update_user(user: User) -> None:
|
|
if user.role not in ALLOWED_ROLES:
|
|
raise ValueError("INVALID_ROLE")
|
|
if user.status not in ALLOWED_STATUSES:
|
|
raise ValueError("INVALID_STATUS")
|
|
if user.is_superuser and user.role != "admin":
|
|
raise ValueError("SUPERUSER_REQUIRES_ADMIN")
|
|
with session_scope() as db:
|
|
db_user = db.get(User, user.id)
|
|
if not db_user:
|
|
return
|
|
db_user.email = user.email
|
|
db_user.password_hash = user.password_hash
|
|
db_user.role = user.role
|
|
db_user.is_superuser = user.is_superuser
|
|
db_user.status = user.status
|
|
db_user.failed_login_attempts = user.failed_login_attempts
|
|
db_user.locked_until = user.locked_until
|
|
db_user.email_verified_at = user.email_verified_at
|
|
db_user.updated_at = datetime.now(UTC)
|
|
|
|
|
|
def list_users(page: int, limit: int) -> tuple[list[User], int]:
|
|
with session_scope() as db:
|
|
total = db.scalar(select(func.count()).select_from(User)) or 0
|
|
users = db.scalars(
|
|
select(User).order_by(User.created_at.desc()).offset((page - 1) * limit).limit(limit)
|
|
).all()
|
|
return [_detach(db, user) for user in users], total
|
|
|
|
|
|
def count_users() -> int:
|
|
with session_scope() as db:
|
|
return db.scalar(select(func.count()).select_from(User)) or 0
|
|
|
|
|
|
def count_users_registered_today() -> int:
|
|
today = datetime.now(UTC).date()
|
|
with session_scope() as db:
|
|
return (
|
|
db.scalar(
|
|
select(func.count())
|
|
.select_from(User)
|
|
.where(func.date(User.created_at) == today)
|
|
)
|
|
or 0
|
|
)
|
|
|
|
|
|
def count_admins() -> int:
|
|
with session_scope() as db:
|
|
return db.scalar(select(func.count()).select_from(User).where(User.role == "admin")) or 0
|
|
|
|
|
|
def count_superusers() -> int:
|
|
with session_scope() as db:
|
|
return (
|
|
db.scalar(
|
|
select(func.count())
|
|
.select_from(User)
|
|
.where(User.role == "admin", User.is_superuser.is_(True))
|
|
)
|
|
or 0
|
|
)
|
|
|
|
|
|
def delete_user(user_id: str) -> bool:
|
|
with session_scope() as db:
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
return False
|
|
db.delete(user)
|
|
return True
|
|
|
|
|
|
def get_profile(user_id: str) -> UserProfile:
|
|
with session_scope() as db:
|
|
profile = db.get(UserProfile, user_id)
|
|
if not profile:
|
|
raise KeyError(user_id)
|
|
return _detach(db, profile)
|
|
|
|
|
|
def update_profile(
|
|
user_id: str,
|
|
display_name: str | None = None,
|
|
avatar_url: str | None = None,
|
|
) -> UserProfile:
|
|
with session_scope() as db:
|
|
profile = db.get(UserProfile, user_id)
|
|
if not profile:
|
|
raise KeyError(user_id)
|
|
if display_name is not None:
|
|
profile.display_name = display_name
|
|
if avatar_url is not None:
|
|
profile.avatar_url = avatar_url
|
|
db.flush()
|
|
return _detach(db, profile)
|