♻️ refactor(model): 重构ItemModel和UserModel以符合CRUD操作

- 移除ItemModel中的密码哈希相关代码，并更名ItemModel为Item
- 添加Item的类方法以支持CRUD操作
- 在UserService中添加密码哈希功能，并更名为UserModel
- 优化用户密码更新逻辑
- 优化查询逻辑，支持超级用户和非超级用户的权限区分【model】

♻️ refactor(service): 优化UserService服务

- 将UserService中的用户验证、密码更新等操作下放到UserModel中
- 优化用户密码更新逻辑，确保密码更新的安全性【service】

Author: lucien277

backend/app/model/item_model.py

@@ -4,50 +4,56 @@
 from sqlalchemy import func
 from sqlmodel import Session, select
 
-from app.core.security import get_password_hash
-from app.models import Item, ItemCreate, ItemUpdate
+from app.models import ItemCreate, ItemUpdate
 
 
-class ItemModel:
+class Item:
     def __init__(self, session: Session):
         self.session = session
 
-    def create(self, item_create: ItemCreate, owner_id: uuid.UUID) -> Item:
-        db_obj = Item.model_validate(
-            item_create,
-            update={"owner_id": owner_id},
-        )
-        self.session.add(db_obj)
-        self.session.commit()
-        self.session.refresh(db_obj)
-        return db_obj
-
-    def update(self, db_item: Item, item_in: ItemUpdate) -> Item:
-        item_data = item_in.model_dump(exclude_unset=True)
-        db_item.sqlmodel_update(item_data)
-        self.session.add(db_item)
-        self.session.commit()
-        self.session.refresh(db_item)
+    def create(cls, item_in: ItemCreate, owner_id: uuid.UUID) -> "Item":
+        db_item = Item.model_validate(item_in, update={"owner_id": owner_id})
+        cls.session.add(db_item)
+        cls.session.commit()
+        cls.session.refresh(db_item)
         return db_item
 
-    def get_by_id(self, item_id: str) -> "Item | None":
-        statement = select(Item).where(Item.id == uuid.UUID(item_id))
-        return self.session.exec(statement).first()
-
-    def get_items(self, skip: int = 0, limit: int = 100, owner_id: uuid.UUID | None = None) -> dict:
-        query = select(Item)
-        if owner_id:
-            query = query.where(Item.owner_id == owner_id)
-            
-        count_statement = select(func.count()).select_from(query.subquery())
-        count = self.session.exec(count_statement).one()
-        
-        query = query.offset(skip).limit(limit)
-        items = self.session.exec(query).all()
+    def get_items(
+        cls,
+        owner_id: uuid.UUID,
+        is_superuser: bool,
+        skip: int = 0,
+        limit: int = 100,
+    ) -> Any:
+        if is_superuser:
+            count_statement = select(func.count()).select_from(Item)
+            count = cls.session.exec(count_statement).one()
+            statement = select(Item).offset(skip).limit(limit)
+            items = cls.session.exec(statement).all()
+        else:
+            count_statement = (
+                select(func.count()).select_from(Item).where(Item.owner_id == owner_id)
+            )
+            count = cls.session.exec(count_statement).one()
+            statement = (
+                select(Item).where(Item.owner_id == owner_id).offset(skip).limit(limit)
+            )
+            items = cls.session.exec(statement).all()
         return {"data": items, "count": count}
 
-    def delete_item(self, item_id: str) -> None:
-        item = self.get_by_id(item_id)
+    def get_by_id(cls, item_id: uuid.UUID) -> "Item | None":
+        return cls.session.get(Item, item_id)
+
+    def update(cls, item: "Item", item_in: ItemUpdate) -> "Item":
+        update_dict = item_in.model_dump(exclude_unset=True)
+        item.sqlmodel_update(update_dict)
+        cls.session.add(item)
+        cls.session.commit()
+        cls.session.refresh(item)
+        return item
+
+    def delete(cls, item_id: uuid.UUID) -> None:
+        item = cls.session.get(Item, item_id)
         if item:
-            self.session.delete(item)
-            self.session.commit()
+            cls.session.delete(item)
+            cls.session.commit()

backend/app/service/user_service.py

@@ -1,68 +1,57 @@
+import uuid
 from typing import Any
 
-from sqlmodel import Session
+from sqlalchemy import func
+from sqlmodel import Session, select
 
-from app.core.security import verify_password
-from app.model.user_model import UserModel
-from app.models import User, UserCreate, UserUpdate, UserUpdateMe
+from app.core.security import get_password_hash
+from app.models import User, UserCreate, UserUpdate
 
 
-class UserService:
+class UserModel:
     def __init__(self, session: Session):
         self.session = session
 
-    def create_user(cls, user_create: UserCreate) -> User:
-        return UserModel(cls.session).create(user_create)
-
-    def update_user(cls, db_user: User, user_in: UserUpdate) -> Any:
-        return UserModel(cls.session).update(db_user, user_in)
-
-    def get_user_by_email(cls, email: str) -> User | None:
-        return UserModel(cls.session).get_by_email(email)
-
-    def get_user_by_id(cls, user_id: str) -> User | None:
-        return UserModel(cls.session).get_by_id(user_id)
-
-    def get_users(cls, skip: int = 0, limit: int = 100) -> dict[str, Any]:
-        return UserModel(cls.session).get_users(skip, limit)
-
-    def update_user_me(cls, current_user: User, user_in: UserUpdateMe) -> User:
-        if user_in.email:
-            existing_user = cls.get_user_by_email(email=user_in.email)
-            if existing_user and existing_user.id != current_user.id:
-                raise ValueError("User with this email already exists")
-
-        # Convert UserUpdateMe to UserUpdate since model expects UserUpdate
-        update_data = UserUpdate(email=user_in.email, full_name=user_in.full_name)
-        return UserModel(cls.session).update(current_user, update_data)
-
-    def update_password(
-        cls, current_user: User, current_password: str, new_password: str
-    ) -> None:
-        if not verify_password(current_password, current_user.hashed_password):
-            raise ValueError("Incorrect password")
-        if current_password == new_password:
-            raise ValueError("New password cannot be the same as the current one")
-
-        # Create UserUpdate with new password
-        update_data = UserUpdate(password=new_password)
-        UserModel(cls.session).update(current_user, update_data)
-
-    def delete_user(
-        cls, user_id: str, current_user_id: str, is_superuser: bool
-    ) -> None:
-        user = cls.get_user_by_id(user_id)
-        if not user:
-            raise ValueError("User not found")
-        if user.id == current_user_id and is_superuser:
-            raise ValueError("Super users are not allowed to delete themselves")
-
-        UserModel(cls.session).delete_user(user_id)
-
-    def authenticate(cls, email: str, password: str) -> User | None:
-        db_user = cls.get_user_by_email(email)
-        if not db_user:
-            return None
-        if not verify_password(password, db_user.hashed_password):
-            return None
+    def create(cls, user_create: UserCreate) -> "User":
+        db_obj = User.model_validate(
+            user_create,
+            update={"hashed_password": get_password_hash(user_create.password)},
+        )
+        cls.session.add(db_obj)
+        cls.session.commit()
+        cls.session.refresh(db_obj)
+        return db_obj
+
+    def update(cls, db_user: "User", user_in: UserUpdate) -> Any:
+        user_data = user_in.model_dump(exclude_unset=True)
+        extra_data = {}
+        if "password" in user_data:
+            password = user_data["password"]
+            hashed_password = get_password_hash(password)
+            extra_data["hashed_password"] = hashed_password
+        db_user.sqlmodel_update(user_data, update=extra_data)
+        cls.session.add(db_user)
+        cls.session.commit()
+        cls.session.refresh(db_user)
         return db_user
+
+    def get_by_email(cls, email: str) -> "User | None":
+        statement = select(User).where(User.email == email)
+        return cls.session.exec(statement).first()
+
+    def get_by_id(cls, user_id: str) -> "User | None":
+        statement = select(User).where(User.id == uuid.UUID(user_id))
+        return cls.session.exec(statement).first()
+
+    def get_users(cls, skip: int = 0, limit: int = 100) -> dict:
+        count_statement = select(func.count()).select_from(User)
+        count = cls.session.exec(count_statement).one()
+        statement = select(User).offset(skip).limit(limit)
+        users = cls.session.exec(statement).all()
+        return {"data": users, "count": count}
+
+    def delete_user(cls, user_id: str) -> None:
+        user = cls.get_by_id(user_id)
+        if user:
+            cls.session.delete(user)
+            cls.session.commit()