|
|
from dataclasses import field
|
|
|
from marshmallow_dataclass import dataclass
|
|
|
from apiflask.fields import Integer, URL, Field
|
|
|
from marshmallow import fields
|
|
|
from apiflask import Schema
|
|
|
from .str import camel_case
|
|
|
|
|
|
|
|
|
# ==================== Schema 定义 ====================
|
|
|
class BaseSchema(Schema):
|
|
|
"""
|
|
|
基础 Schema 扩展
|
|
|
1. 有序返回
|
|
|
2. 未知字段不报错
|
|
|
"""
|
|
|
|
|
|
class Meta:
|
|
|
ordered = True
|
|
|
unknown = "INCLUDE"
|
|
|
|
|
|
def on_bind_field(self, field_name: str, field_obj: Field) -> None:
|
|
|
"""
|
|
|
绑定字段时处理
|
|
|
1.统一驼峰命名返回(小驼峰)
|
|
|
"""
|
|
|
if field_obj.data_key is None:
|
|
|
field_obj.data_key = camel_case(field_name)
|
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
@dataclass(base_schema=Schema)
|
|
|
class Pagination:
|
|
|
page: int = field(default=1, metadata={"metadata": {"description": "当前页码"}})
|
|
|
size: int = field(default=10, metadata={"metadata": {"description": "每页数量"}})
|
|
|
pages: int = field(default=0, metadata={"metadata": {"description": "总页数"}})
|
|
|
total: int = field(default=0, metadata={"metadata": {"description": "总记录数"}})
|
|
|
current: str = field(
|
|
|
default="",
|
|
|
metadata={"metadata": {"description": "当前页URL"}, "dump_only": True},
|
|
|
)
|
|
|
next: str = field(
|
|
|
default="",
|
|
|
metadata={"metadata": {"description": "下一页URL"}, "dump_only": True},
|
|
|
)
|
|
|
prev: str = field(
|
|
|
default="",
|
|
|
metadata={"metadata": {"description": "上一页URL"}, "dump_only": True},
|
|
|
)
|
|
|
first: str = field(
|
|
|
default="", metadata={"metadata": {"description": "首页URL"}, "dump_only": True}
|
|
|
)
|
|
|
last: str = field(
|
|
|
default="", metadata={"metadata": {"description": "末页URL"}, "dump_only": True}
|
|
|
)
|
|
|
|
|
|
|
|
|
# 获取 Pagination 类中的所有字段
|
|
|
pagination_fields = [field.name for field in Pagination.__dataclass_fields__.values()]
|
|
|
|
|
|
|
|
|
class PaginationSchema(Schema):
|
|
|
"""自定义分页信息 Schema（与 APIFlask 保持一致）"""
|
|
|
|
|
|
page = Integer(metadata={"description": "当前页码"})
|
|
|
size = Integer(metadata={"description": "每页数量"}) # per_page → size
|
|
|
pages = Integer(metadata={"description": "总页数"})
|
|
|
total = Integer(metadata={"description": "总记录数"})
|
|
|
current = URL(metadata={"description": "当前页URL"}, dump_only=True)
|
|
|
next = URL(metadata={"description": "下一页URL"}, dump_only=True)
|
|
|
prev = URL(metadata={"description": "上一页URL"}, dump_only=True)
|
|
|
first = URL(metadata={"description": "首页URL"}, dump_only=True)
|
|
|
last = URL(metadata={"description": "末页URL"}, dump_only=True)
|
|
|
|
|
|
|
|
|
# 获取 PaginationSchema 类中的所有字段
|
|
|
pagination_schema_fields = list(PaginationSchema._declared_fields.keys())
|
|
|
|
|
|
|
|
|
def page_schema(item_schema_cls: type, *, schema_name: str | None = None) -> type:
|
|
|
"""
|
|
|
根据传入的 Item OutSchema 生成通用分页负载 Schema：
|
|
|
{ items: [Item], page: Pagination }
|
|
|
|
|
|
- item_schema_cls: 基础 Schema
|
|
|
- schema_name: 可选，自定义 schema 名称
|
|
|
"""
|
|
|
if not schema_name:
|
|
|
schema_name = f"PageItems[{getattr(item_schema_cls, '__name__', 'Item')}]"
|
|
|
|
|
|
class _PageItemsSchema(Schema):
|
|
|
items = fields.Nested(item_schema_cls, many=True, required=True)
|
|
|
page = fields.Nested(PaginationSchema, required=True)
|
|
|
|
|
|
_PageItemsSchema.__name__ = schema_name
|
|
|
return _PageItemsSchema
|
|
|
|
|
|
|
|
|
def condition_schema(base_schema_cls: type, control_config):
|
|
|
"""
|
|
|
多字段控制动态Schema创建
|
|
|
|
|
|
Args:
|
|
|
base_schema_class: 基础 Schema 类
|
|
|
control_config: 控制配置字典
|
|
|
{
|
|
|
"withDataList": ["data_list"], # 当 withDataList=true 时包含 data_list
|
|
|
"withTimestamps": ["created_at", "updated_at"], # 当 withTimestamps=true 时包含时间字段
|
|
|
"withStatus": ["status"], # 当 withStatus=true 时包含状态字段
|
|
|
}
|
|
|
"""
|
|
|
|
|
|
class _ConditionSchema(base_schema_cls):
|
|
|
def __init__(self, *args, **kwargs):
|
|
|
self._control_config = control_config
|
|
|
super().__init__(*args, **kwargs)
|
|
|
|
|
|
def dump(self, obj, many=None, **kwargs):
|
|
|
from flask import request
|
|
|
|
|
|
# 收集排除字段列表
|
|
|
exclude_fields = []
|
|
|
for control_field, target_fields in self._control_config.items():
|
|
|
try:
|
|
|
control_value = (
|
|
|
request.args.get(control_field, "false").lower() == "true"
|
|
|
)
|
|
|
except Exception:
|
|
|
# 没有请求上下文时，默认不排除字段
|
|
|
control_value = True
|
|
|
|
|
|
if not control_value:
|
|
|
exclude_fields.extend(target_fields)
|
|
|
|
|
|
# 如果有字段需要排除，创建临时 Schema
|
|
|
if exclude_fields:
|
|
|
temp_schema = base_schema_cls(exclude=exclude_fields)
|
|
|
return temp_schema.dump(obj, many=many, **kwargs)
|
|
|
else:
|
|
|
return super().dump(obj, many=many, **kwargs)
|
|
|
|
|
|
return _ConditionSchema
|
|
|
|
|
|
|
|
|
def custom_schema_name_resolver(schema):
|
|
|
"""
|
|
|
自定义 schema 名称解析器，解决循环引用导致的命名冲突
|
|
|
|
|
|
根据 APIFlask 官方文档：
|
|
|
- 函数接收一个 schema 对象作为参数
|
|
|
- 返回一个字符串作为 schema 的名称
|
|
|
- 用于解决多个 schema 解析为相同名称的问题
|
|
|
|
|
|
处理策略：
|
|
|
1. 优先使用 Meta.name（如果定义）
|
|
|
2. 自动移除 Schema 后缀
|
|
|
3. 为带有 exclude 参数的嵌套 schema 生成唯一名称
|
|
|
"""
|
|
|
schema_class = schema.__class__
|
|
|
|
|
|
# 1. 优先检查是否在 Meta 中定义了 name
|
|
|
if hasattr(schema_class, "Meta") and hasattr(schema_class.Meta, "name"):
|
|
|
base_name = schema_class.Meta.name
|
|
|
else:
|
|
|
# 2. 使用类名，移除 Schema 后缀
|
|
|
base_name = schema_class.__name__
|
|
|
if base_name.endswith("Schema"):
|
|
|
base_name = base_name[:-6]
|
|
|
if schema.partial: # 为部分模式添加 "Update" 后缀
|
|
|
base_name += "Update"
|
|
|
|
|
|
# 3. 处理嵌套时的 exclude 参数
|
|
|
# 当使用 Nested("SomeSchema", exclude=["field1", "field2"]) 时
|
|
|
# apispec 会创建新的 schema 实例，需要为其生成唯一名称
|
|
|
if hasattr(schema, "exclude") and schema.exclude:
|
|
|
# 将 exclude 的字段排序，确保相同的 exclude 组合生成相同的名称
|
|
|
excluded_fields = sorted(schema.exclude)
|
|
|
# 生成简洁的后缀：首字母大写拼接
|
|
|
# 例如：exclude=["children", "parent"] -> "ChildrenParent"
|
|
|
suffix = "".join([field.capitalize() for field in excluded_fields])
|
|
|
return f"{base_name}Exclude{suffix}"
|
|
|
|
|
|
# 4. 处理 only 参数（如果使用）
|
|
|
if hasattr(schema, "only") and schema.only:
|
|
|
only_fields = sorted(schema.only)
|
|
|
suffix = "".join([field.capitalize() for field in only_fields])
|
|
|
return f"{base_name}Only{suffix}"
|
|
|
|
|
|
return base_name
|