[feat] User list pagination function

api/app/services/memory_dashboard_service.py

@@ -1,11 +1,12 @@
 from sqlalchemy.orm import Session
-from typing import List, Optional
+from sqlalchemy import desc, nullslast, or_, and_, cast, String
+from typing import List, Optional, Dict, Any
 import uuid
 from fastapi import HTTPException
 
 from app.models.user_model import User
 from app.models.app_model import App
-from app.models.end_user_model import EndUser
+from app.models.end_user_model import EndUser, EndUser as EndUserModel
 from app.models.memory_increment_model import MemoryIncrement
 
 from app.repositories import (
@@ -49,44 +50,40 @@ def get_current_workspace_type(
 
 
 def get_workspace_end_users(
-    db: Session, 
-    workspace_id: uuid.UUID, 
+    db: Session,
+    workspace_id: uuid.UUID,
     current_user: User
 ) -> List[EndUser]:
     """获取工作空间的所有宿主（优化版本：减少数据库查询次数）
-    
     返回结果按 created_at 从新到旧排序（NULL 值排在最后）
     """
     business_logger.info(f"获取工作空间宿主列表: workspace_id={workspace_id}, 操作者: {current_user.username}")
-    
-    try:        
+
+    try:
         # 查询应用（ORM）
         apps_orm = app_repository.get_apps_by_workspace_id(db, workspace_id)
-        
+
         if not apps_orm:
             business_logger.info("工作空间下没有应用")
             return []
-        
+
         # 提取所有 app_id
         # app_ids = [app.id for app in apps_orm]
-        
         # 批量查询所有 end_users（一次查询而非循环查询）
         # 按 created_at 降序排序，NULL 值排在最后；id 作为次级排序键保证确定性
-        from app.models.end_user_model import EndUser as EndUserModel
-        from sqlalchemy import desc, nullslast
         end_users_orm = db.query(EndUserModel).filter(
             EndUserModel.workspace_id == workspace_id
         ).order_by(
             nullslast(desc(EndUserModel.created_at)),
             desc(EndUserModel.id)
         ).all()
-        
+
         # 转换为 Pydantic 模型（只在需要时转换）
         end_users = [EndUserSchema.model_validate(eu) for eu in end_users_orm]
-        
+
         business_logger.info(f"成功获取 {len(end_users)} 个宿主记录")
         return end_users
-        
+
     except HTTPException:
         raise
     except Exception as e:
@@ -94,6 +91,92 @@ def get_workspace_end_users(
         raise
 
 
+def get_workspace_end_users_paginated(
+    db: Session,
+    workspace_id: uuid.UUID,
+    current_user: User,
+    page: int,
+    pagesize: int,
+    keyword: Optional[str] = None
+) -> Dict[str, Any]:
+    """获取工作空间的宿主列表（分页版本，支持模糊搜索）
+
+    返回结果按 created_at 从新到旧排序（NULL 值排在最后）
+    支持通过 keyword 参数同时模糊搜索 other_name 和 id 字段
+
+    Args:
+        db: 数据库会话
+        workspace_id: 工作空间ID
+        current_user: 当前用户
+        page: 页码（从1开始）
+        pagesize: 每页数量
+        keyword: 搜索关键词（可选，同时模糊匹配 other_name 和 id）
+
+    Returns:
+        dict: 包含 items（宿主列表）和 total（总记录数）的字典
+    """
+    business_logger.info(f"获取工作空间宿主列表（分页）: workspace_id={workspace_id}, keyword={keyword}, page={page}, pagesize={pagesize}, 操作者: {current_user.username}")
+
+    try:
+        # 构建基础查询
+        base_query = db.query(EndUserModel).filter(
+            EndUserModel.workspace_id == workspace_id
+        )
+
+        # 构建搜索条件（过滤空字符串和None）
+        keyword = keyword.strip() if keyword else None
+
+        if keyword:
+            keyword_pattern = f"%{keyword}%"
+            # 优先匹配 other_name，如果 other_name 为空则匹配 id
+            # 使用 OR 条件：匹配 other_name 不为空的数据，或者 other_name 为空但 id 匹配的数据
+            base_query = base_query.filter(
+                or_(
+                    # 情况1：other_name 不为空且匹配 keyword
+                    and_(
+                        EndUserModel.other_name != "",
+                        EndUserModel.other_name.isnot(None),
+                        EndUserModel.other_name.ilike(keyword_pattern)
+                    ),
+                    # 情况2：other_name 为空或 None，但 id 匹配 keyword
+                    and_(
+                        or_(
+                            EndUserModel.other_name == "",
+                            EndUserModel.other_name.is_(None)
+                        ),
+                        cast(EndUserModel.id, String).ilike(keyword_pattern)
+                    )
+                )
+            )
+            business_logger.info(f"应用模糊搜索: keyword={keyword}（优先匹配 other_name，无 other_name 时匹配 id）")
+
+        # 获取总记录数
+        total = base_query.count()
+
+        if total == 0:
+            business_logger.info("工作空间下没有宿主")
+            return {"items": [], "total": 0}
+
+        # 分页查询
+        # 按 created_at 降序排序，NULL 值排在最后；id 作为次级排序键保证确定性
+        end_users_orm = base_query.order_by(
+            nullslast(desc(EndUserModel.created_at)),
+            desc(EndUserModel.id)
+        ).offset((page - 1) * pagesize).limit(pagesize).all()
+
+        # 转换为 Pydantic 模型
+        end_users = [EndUserSchema.model_validate(eu) for eu in end_users_orm]
+
+        business_logger.info(f"成功获取 {len(end_users)} 个宿主记录，总计 {total} 条")
+        return {"items": end_users, "total": total}
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        business_logger.error(f"获取工作空间宿主列表（分页）失败: workspace_id={workspace_id} - {str(e)}")
+        raise
+
+
 def get_workspace_memory_increment(
     db: Session, 
     workspace_id: uuid.UUID,