webapp/app/services/user_service.py

# app/services/user_service.py

import uuid
from datetime import datetime
from typing import List, Optional

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.utils.search import apply_keyword_search_stmt
from app.utils.pagination import paginate
from app.core.biz_exception import NotFoundError
from app.models.user import VasUser
from app.schemas.user import (
    VasUserCreate,
    VasUserUpdate,
    VasUserSetProfiles,
    VasUserOut,
)


class UserService:

    @staticmethod
    async def create(db: AsyncSession, data: VasUserCreate) -> VasUser:
        uid = f"usr-{uuid.uuid4().hex[:8]}"

        user = VasUser(
            id=uid,
            role=data.role,
            nickname=data.nickname,
            phone=data.phone,
            preferred_language="en",
            timezone="Asia/Shanghai",
            created_at=datetime.utcnow(),
        )

        db.add(user)
        await db.commit()
        await db.refresh(user)

        return user

    @staticmethod
    async def get(db: AsyncSession, id: str) -> VasUser:
        stmt = select(VasUser).where(VasUser.id == id)
        result = await db.execute(stmt)
        user = result.scalar_one_or_none()

        if not user:
            raise NotFoundError("User not exist")

        return user

    @staticmethod
    async def update(
        db: AsyncSession,
        id: str,
        payload: VasUserUpdate
    ) -> VasUser:
        stmt = select(VasUser).where(VasUser.id == id)
        result = await db.execute(stmt)
        obj = result.scalar_one_or_none()

        if not obj:
            raise NotFoundError("User not exist")

        data = payload.dict(exclude_unset=True)

        for key, value in data.items():
            setattr(obj, key, value)

        obj.updated_at = datetime.utcnow()

        await db.commit()
        await db.refresh(obj)

        return obj

    @staticmethod
    async def set_profiles(
        db: AsyncSession,
        user: VasUser,
        payload: VasUserSetProfiles
    ) -> VasUser:
        """
        更新用户资料（profile）
        """

        user.phone = payload.phone
        user.nickname = payload.nickname
        user.avatar_url = payload.avatar_url
        user.updated_at = datetime.utcnow()

        db.add(user)
        await db.commit()
        await db.refresh(user)

        return user

    @staticmethod
    async def list_all(
        db: AsyncSession,
        page: int = 1,
        size: int = 20,
        keyword: Optional[str] = None,
    ):
        stmt = select(VasUser)

        stmt = apply_keyword_search_stmt(
            stmt=stmt,
            model=VasUser,
            keyword=keyword,
            fields=["id", "email", "nickname", "phone"],
        )

        return await paginate(db, stmt, page, size)