35 lines
1.7 KiB
Python
35 lines
1.7 KiB
Python
"""前端和 AI 共用的只读查询协议；不接受 SQL、表名或任意表达式。"""
|
|
from typing import Literal
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
class QueryFilter(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
|
|
field: str = Field(min_length=1, max_length=40)
|
|
op: Literal["eq", "contains", "gt", "ge", "lt", "le", "is_null", "not_null"] = "eq"
|
|
value: str | int | float | None = None
|
|
|
|
|
|
class DataQuery(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", allow_inf_nan=False)
|
|
module: Literal["students", "classes", "teachers", "advisors", "scores", "employment", "teacher_classes", "teacher_students"]
|
|
filters: list[QueryFilter] = Field(default_factory=list, max_length=12)
|
|
group_by: list[str] = Field(default_factory=list, max_length=3)
|
|
aggregate: Literal["count", "avg", "sum", "min", "max"] | None = None
|
|
aggregate_field: str | None = None
|
|
having_min: float | None = None
|
|
sort_by: str | None = None
|
|
descending: bool = False
|
|
page: int = Field(default=1, ge=1, le=100000)
|
|
page_size: int = Field(default=10, ge=1, le=100)
|
|
|
|
@model_validator(mode="after")
|
|
def valid_aggregate(self):
|
|
if not self.aggregate and (self.group_by or self.aggregate_field or self.having_min is not None):
|
|
raise ValueError("分组、聚合字段和聚合筛选需要指定 aggregate")
|
|
if self.aggregate and self.aggregate != "count" and not self.aggregate_field:
|
|
raise ValueError("请指定需要统计的数值字段")
|
|
if self.aggregate == "count" and self.aggregate_field:
|
|
raise ValueError("count 统计记录数，无需 aggregate_field")
|
|
return self
|