webapp/app/services/user_service.py

# app/services/user_service.py
from datetime import datetime
from typing import List
from sqlalchemy.orm import Session
from app.utils.search import apply_keyword_search
from app.utils.pagination import paginate
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.user import VasUser
from app.schemas.user import VasUserCreate, VasUserUpdate, VasUserSetProfiles, VasUserOut


class UserService:
    
    @staticmethod
    def create(db: Session, data: VasUserCreate):
        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",
        )
        db.add(user)
        db.commit()
        return rec
    
    @staticmethod
    def get(db: Session, id: str):
        user = db.query(VasUser).filter_by(id=id).first()
        if not user:
            raise NotFoundError("User not exist")
        return user
    
    @staticmethod
    def update(db: Session, id: str, payload: VasUserUpdate):
        obj = db.query(VasUser).filter(VasUser.id == id).first()
        if not obj:
            raise NotFoundError("User not exist")
        data = payload.dict(exclude_unset=True)  # ⭐ 关键

        for key, value in data.items():
            setattr(obj, key, value)
        db.commit()
        db.refresh(obj)
        return obj
    
    @staticmethod
    def set_profiles(db: Session, user: VasUser, payload: VasUserSetProfiles):
        """
        更新用户资料（profile）
        """

        # 1️⃣ 字段赋值（显式，避免误更新）
        user.phone = payload.phone
        user.nickname = payload.nickname
        user.avatar_url = payload.avatar_url

        # 2️⃣ 更新时间（可选，SQLAlchemy onupdate 也会生效）
        user.updated_at = datetime.utcnow()

        # 3️⃣ 持久化
        db.add(user)
        db.commit()
        db.refresh(user)

        return user

    @staticmethod
    def list_all(db: Session, page: int = 1, size: int = 20, keyword: str=None):
        query = db.query(VasUser)
        
        query = apply_keyword_search(
            query=query,
            model=VasUser,
            keyword=keyword,
            fields=["id", "email", "nickname", "phone"]
        )
        return paginate(query, page, size)