(
session: SessionDep,
pageNum: int = Path(..., title=f"{PLACEHOLDER_PREFIX}page_num", description=f"{PLACEHOLDER_PREFIX}page_num"),
pageSize: int = Path(..., title=f"{PLACEHOLDER_PREFIX}page_size", description=f"{PLACEHOLDER_PREFIX}page_size"),
keyword: Optional[str] = Query(None, description=f"{PLACEHOLDER_PREFIX}keyword"),
status: Optional[int] = Query(None, description=f"{PLACEHOLDER_PREFIX}status"),
origins: Optional[list[int]] = Query(None, description=f"{PLACEHOLDER_PREFIX}origin"),
oidlist: Optional[list[int]] = Query(None, description=f"{PLACEHOLDER_PREFIX}oid"),
)
| 51 | @router.get("/pager/{pageNum}/{pageSize}", response_model=PaginatedResponse[UserGrid], summary=f"{PLACEHOLDER_PREFIX}system_user_grid", description=f"{PLACEHOLDER_PREFIX}system_user_grid") |
| 52 | @require_permissions(permission=SqlbotPermission(role=['admin'])) |
| 53 | async def pager( |
| 54 | session: SessionDep, |
| 55 | pageNum: int = Path(..., title=f"{PLACEHOLDER_PREFIX}page_num", description=f"{PLACEHOLDER_PREFIX}page_num"), |
| 56 | pageSize: int = Path(..., title=f"{PLACEHOLDER_PREFIX}page_size", description=f"{PLACEHOLDER_PREFIX}page_size"), |
| 57 | keyword: Optional[str] = Query(None, description=f"{PLACEHOLDER_PREFIX}keyword"), |
| 58 | status: Optional[int] = Query(None, description=f"{PLACEHOLDER_PREFIX}status"), |
| 59 | origins: Optional[list[int]] = Query(None, description=f"{PLACEHOLDER_PREFIX}origin"), |
| 60 | oidlist: Optional[list[int]] = Query(None, description=f"{PLACEHOLDER_PREFIX}oid"), |
| 61 | ): |
| 62 | pagination = PaginationParams(page=pageNum, size=pageSize) |
| 63 | paginator = Paginator(session) |
| 64 | filters = {} |
| 65 | |
| 66 | origin_stmt = ( |
| 67 | select(UserModel.id, UserModel.account) |
| 68 | .join(UserWsModel, UserModel.id == UserWsModel.uid, isouter=True) |
| 69 | .where(UserModel.id != 1) |
| 70 | .distinct() |
| 71 | .order_by(UserModel.account) |
| 72 | ) |
| 73 | |
| 74 | if oidlist: |
| 75 | origin_stmt = origin_stmt.where(UserWsModel.oid.in_(oidlist)) |
| 76 | if origins: |
| 77 | origin_stmt = origin_stmt.where(UserModel.origin.in_(origins)) |
| 78 | if status is not None: |
| 79 | origin_stmt = origin_stmt.where(UserModel.status == status) |
| 80 | if keyword: |
| 81 | keyword_pattern = f"%{keyword}%" |
| 82 | origin_stmt = origin_stmt.where( |
| 83 | or_( |
| 84 | UserModel.account.ilike(keyword_pattern), |
| 85 | UserModel.name.ilike(keyword_pattern), |
| 86 | UserModel.email.ilike(keyword_pattern) |
| 87 | ) |
| 88 | ) |
| 89 | |
| 90 | user_page = await paginator.get_paginated_response( |
| 91 | stmt=origin_stmt, |
| 92 | pagination=pagination, |
| 93 | **filters) |
| 94 | uid_list = [item.get('id') for item in user_page.items] |
| 95 | if not uid_list: |
| 96 | return user_page |
| 97 | stmt = ( |
| 98 | select(UserModel, UserWsModel.oid.label('ws_oid')) |
| 99 | .join(UserWsModel, UserModel.id == UserWsModel.uid, isouter=True) |
| 100 | .where(UserModel.id.in_(uid_list)) |
| 101 | .order_by(UserModel.account, UserModel.create_time) |
| 102 | ) |
| 103 | user_workspaces = session.exec(stmt).all() |
| 104 | merged = defaultdict(list) |
| 105 | extra_attrs = {} |
| 106 | |
| 107 | for (user, ws_oid) in user_workspaces: |
| 108 | item = {} |
| 109 | item.update(user.model_dump()) |
| 110 | user_id = item['id'] |
nothing calls this directly
no test coverage detected