(
request: Request,
data: RecordListRequest = Depends(),
path_params: Dict = Depends(path_params_required),
auth_info: Dict = Depends(auth_info_required),
)
| 56 | responses={422: {"description": "Unprocessable Entity"}}, |
| 57 | ) |
| 58 | async def api_list( |
| 59 | request: Request, |
| 60 | data: RecordListRequest = Depends(), |
| 61 | path_params: Dict = Depends(path_params_required), |
| 62 | auth_info: Dict = Depends(auth_info_required), |
| 63 | ): |
| 64 | check_path_params( |
| 65 | model_operator=ops, |
| 66 | object_id_required=False, |
| 67 | path_params=path_params, |
| 68 | ) |
| 69 | |
| 70 | data_prefix_filter = getattr(data, "prefix_filter", "") |
| 71 | data_equal_filter = getattr(data, "equal_filter", "") |
| 72 | prefix_filter_dict, equal_filter_dict = await validate_list_filter( |
| 73 | model_operator=ops, |
| 74 | path_params=path_params, |
| 75 | prefix_filter=data_prefix_filter, |
| 76 | equal_filter=data_equal_filter, |
| 77 | ) |
| 78 | |
| 79 | entities, has_more = await ops.list( |
| 80 | limit=data.limit, |
| 81 | order=data.order, |
| 82 | after_id=data.after, |
| 83 | before_id=data.before, |
| 84 | prefix_filters=prefix_filter_dict, |
| 85 | equal_filters=equal_filter_dict, |
| 86 | **path_params, |
| 87 | ) |
| 88 | return RecordListResponse( |
| 89 | data=[entity.to_response_dict() for entity in entities], |
| 90 | fetched_count=len(entities), |
| 91 | has_more=has_more, |
| 92 | ) |
| 93 | |
| 94 | @router.post( |
| 95 | path="/collections/{collection_id}/records", |
nothing calls this directly
no test coverage detected