Feature: GET /datasets to support ids array (#17606)

This commit is contained in:
Wang Qi
2026-07-31 13:23:02 +08:00
committed by GitHub
parent e66cfe1372
commit 0969c04eca
6 changed files with 90 additions and 5 deletions

View File

@@ -19,7 +19,8 @@ import pathlib
import re
from collections import Counter
import string
from typing import Annotated, Any, Literal
from types import UnionType
from typing import Annotated, Any, Literal, Union, get_args, get_origin
from uuid import UUID
from quart import Request
@@ -33,6 +34,15 @@ from api.utils.pagination_utils import validate_rest_api_page_size
from common.constants import RetCode
def _is_list_annotation(annotation: Any) -> bool:
origin = get_origin(annotation)
if origin is list:
return True
if origin in (Union, UnionType):
return any(_is_list_annotation(arg) for arg in get_args(annotation))
return False
async def validate_and_parse_json_request(
request: Request, validator: type[BaseModel], *, extras: dict[str, Any] | None = None, exclude_unset: bool = False
) -> tuple[dict[str, Any] | None, str | None]:
@@ -161,6 +171,10 @@ def validate_and_parse_request_args(request: Request, validator: type[BaseModel]
- Preserves type conversion from Pydantic validation
"""
args = request.args.to_dict(flat=True)
for field_name, field_info in validator.model_fields.items():
query_name = field_info.alias or field_name
if query_name in request.args and _is_list_annotation(field_info.annotation):
args[query_name] = [value for item in request.args.getlist(query_name) for value in item.split(",") if value]
# Handle ext parameter: parse JSON string to dict if it's a string
if "ext" in args and isinstance(args["ext"], str):
@@ -1002,9 +1016,26 @@ class BaseListReq(BaseModel):
class ListDatasetReq(BaseListReq):
"""Request model for listing datasets."""
ids: Annotated[list[str] | None, Field(default=None)]
include_parsing_status: Annotated[bool, Field(default=False)]
ext: Annotated[dict, Field(default={})]
@field_validator("ids", mode="after")
@classmethod
def validate_ids(cls, v_list: list[str] | None) -> list[str] | None:
if v_list is None:
return None
ids_list = []
for v in v_list:
ids_list.append(validate_uuid1_hex(v))
duplicates = [item for item, count in Counter(ids_list).items() if count > 1]
if duplicates:
raise PydanticCustomError("duplicate_uuids", "Duplicate ids: '{duplicate_ids}'", {"duplicate_ids": ", ".join(duplicates)})
return ids_list
# ---- File Management Request Models ----