增加针对动态字段的数据解析 #652

wang935415150 · 2025-06-05T02:53:59Z

我在工作中遇到了一个问题,就是针对动态的BaseModel 使用项目中的page_data方法并不能做到解析,这导致了,我的字段缺失,所以修改了如下方法, 希望wu可以增加进去. 我的应用场景是这样的
class UserSchema(SchemaBase,SchemaBaseOut):
name: str = Field(..., title="姓名")
age: int = Field(..., title="年龄")
sex: int = Field(..., title="性别")
model_config = ConfigDict(from_attributes=True)

wang935415150 · 2025-06-05T02:55:45Z

当然,wu你会有更好的方法, 去完成这段代码,我是根据我现在的情况,提供的适合我自己的代码方式, 希望你能有更优秀的解决方案

wang935415150 · 2025-06-05T02:57:36Z

我还需要处理另一种状态
class DeptSchema(SchemaBase,SchemaBaseOut):
name: str = Field(..., title="部门名称")
leader: str = Field(..., title="部门领导")
phone: str = Field(..., title="联系电话")
model_config = ConfigDict(from_attributes=True)

class UserSchema(SchemaBase,SchemaBaseOut):
    name: str = Field(..., title="姓名")
    age: int = Field(..., title="年龄")
    sex: int = Field(..., title="性别")
    dept: list[DeptSchema] | None = Field(default=None, title="部门")
    model_config = ConfigDict(from_attributes=True)

wang935415150 · 2025-06-05T02:58:02Z

希望以上遇到的处理情况对你有帮助

wang935415150 · 2025-06-05T03:05:31Z

我尝试修改了一下让他适配除了 n 和c以外的 schema :
# 确保items中的每个对象都保留其所有属性（包括所有动态字段）
if 'items' in page_data and page_data['items']:
for i, item in enumerate(page_data['items']):
# 检查原始对象是否有__dict__属性（SQLAlchemy模型对象通常有）
if hasattr(paginated_data.items[i], 'dict'):
# 获取原始对象的所有属性
obj_dict = paginated_data.items[i].dict
# 添加所有属性（排除SQLAlchemy内部属性）
for key, value in obj_dict.items():
# 排除SQLAlchemy内部属性（以_开头）和已存在的属性
if not key.startswith('_') and key not in item:
item[key] = value

wu-clan · 2025-06-05T03:14:16Z

我没理解这里这些动态字段的实际作用，直接目的是用于什么？

wang935415150 · 2025-06-05T03:20:37Z

我正在使用你写的框架来建立一个低代码工具平台, 这个平台最大的一个特点是每一个表模型都拥有冗余的字段如 C1-N20他的作用是让用户自动创造字段对应的意义, 我不可能每个模型的建立一个 c1-n20字段所以我创建了一个积累class DataClassBase(MappedAsDataclass, MappedBase):
"""数据基础类 - 包含动态字段c1-c20和n1-n20"""
abstract = True
id: Mapped[id_key] = mapped_column(init=False)

【！重要】动态字段生成 - 在类定义完成后立即执行

for i in range(1, 21):
field_name = f'c{i}'
setattr(DataClassBase, field_name,
mapped_column(String(20), init=True, nullable=True,
insert_default=None, default=None,
comment=f'字符串{i}'))

for i in range(1, 21):
field_name = f'n{i}'
setattr(DataClassBase, field_name,
mapped_column(Float, init=True, nullable=True,
insert_default=None, default=None,
sort_order=998, comment=f'浮点数{i}'))

由所有模型去继承. 同样的在schema层我也创建了一个基础 schema 进行共同继承.
以上是业务层面.
而我遇到的一个问题, 是在通过 paging_data 方法的时候 schema会将动态 schema过滤掉,这就造成了, 我必须手动的去写一个 schema而不是使用 pydantic的create_model 自动生成然后通过继承的方式来获得数据清洗

wang935415150 · 2025-06-05T03:23:24Z

简单的说 , 如果 pydantic的create_model 动态生成一个解析 schema , 经过 paging_data 的时候, paging_data不会处理动态create_model的schema的数据, 导致数据消失, 检验的方式是在 paging_data 方法执行前使用 db.execute 执行一下select 和执行 paging_data 吐出的数据来做对照既可以发现问题

wang935415150 · 2025-06-05T03:23:59Z

问题不是在于我的业务需求,问题是在于 paging_data 针对 pydantic的create_model 生成的 schema并不支持

wu-clan · 2025-06-05T03:55:13Z

分页查询接口返回类型是怎样的？这个也会影响最终序列化返回

wu-clan · 2025-06-05T03:59:26Z

能否提供一个MRE？

wang935415150 · 2025-06-05T05:14:28Z

没理解你的意思, 返回体 json 啊
ResponseSchemaModel[PageData[ProcessOrderProcessRouteSchemaOut]]

wu-clan · 2025-06-05T08:59:30Z

Minimal Reproducible Example (MRE)：最小可重现示例

我需要一个示例进行复现

wu-clan · 2025-06-05T09:06:27Z

可以先看下 paging_data 中的 paginated_data 是否和 paging_data 方法执行前使用 db.execute 执行 select 返回的一致，如果这里不一致，那可能是 fastapi-pagination 内部默认设置导致的

没有实例无法排查到底哪里导致的问题

wang935415150 · 2025-06-06T01:23:38Z

class DataClassBase(MappedAsDataclass, MappedBase):
    """数据基础类 - 包含动态字段c1-c20和n1-n20"""
    __abstract__ = True
    id: Mapped[id_key] = mapped_column(init=False)

# 【！重要】动态字段生成 - 在类定义完成后立即执行
for i in range(1, 21):
    field_name = f'c{i}'
    setattr(DataClassBase, field_name, 
           mapped_column(String(20), init=True, nullable=True, 
                        insert_default=None, default=None, 
                        comment=f'字符串{i}'))

for i in range(1, 21):
    field_name = f'n{i}'
    setattr(DataClassBase, field_name, 
           mapped_column(Float, init=True, nullable=True, 
                        insert_default=None, default=None, 
                        sort_order=998, comment=f'浮点数{i}'))

class Menu(DataClassBase):
    test1: Mapped[str] = mapped_column(String(50), unique=True, comment='test1')
    test2: Mapped[str] = mapped_column(String(16), comment='test2')
    parent: .......

class SchemaBaseNoId(BaseModel):
    model_config = ConfigDict(use_enum_values=True)

# 动态创建字段定义
fields = {}
for i in range(1, 21):
    field_name = f'c{i}'
    # 使用新版Pydantic的字段定义方式
    fields[field_name] = (Optional[str], Field(default=None))
for i in range(1, 21):
    field_name = f'n{i}'
    # 使用新版Pydantic的字段定义方式
    fields[field_name] = (Optional[float], Field(default=None))

# 使用 create_model 动态创建 SchemaBaseOut 类
SchemaBaseOut = create_model(
    'SchemaBaseOut',  # 模型名称
    **fields  # 字段名和对应的类型及默认值
)
class BaseSchema(BaseModel):
    model_config = ConfigDict(use_enum_values=True)

class GetMenuDetail(BaseSchema,SchemaBaseOut):
    test1: str = Field(..., description='测试')

class MenuSchema(BaseSchema,SchemaBaseOut):
    menus: list[GetMenuDetail] = Field(..., description='菜单列表')
    test1: str = Field(..., description='测试')

@router.post('/menu_list'
, summary='分页获取菜单的详情', dependencies=[
    DependsJwtAuth
    ])
async def get_menu_list(
    db: CurrentSession,
    request: Request,
    permission_id: int = Body(..., description='权限组id'),
    pagination: PageData = DependsPagination,
                          ) -> ResponseSchemaModel[PageData[GetMenuDetail]]:
    
    stmt = await menu_service.get_menu_list(permission_id=permission_id, request=request)
    db.execute(stmt)
    sql_data = await stmt.scalars.all()
    print(sql_data)  # 一切正常 能够查询到 自动生成的字段
    page_data = await paging_data(db, stmt)
    print(page_data) # 出现问题 无法解析正常的自动生成字段
    
    return response_base.success(data=page_data)

一句 MRE我以为大佬改行玩DL了呢,怎么误差都搞出来了,我还想咱们也没搞Traindata啊 0.0

wu-clan · 2025-06-06T06:41:27Z

模型列最终会创建到数据库，所以本质上，动不动态关系不大，正常 select 查询出来肯定是包含动态列的

你这里的 schema 动态列，本质上只是使用 create_model 创建了类，动不动态也没关系

以下是参考你的 MRE 在现有的代码种做的测试
api/opera_log

@router.get(
    '',
    summary='分页获取操作日志',
    dependencies=[
        DependsJwtAuth,
        # DependsPagination,
    ],
)
async def get_pagination_opera_logs(
    db: CurrentSession,
    username: Annotated[str | None, Query(description='用户名')] = None,
    status: Annotated[int | None, Query(description='状态')] = None,
    ip: Annotated[str | None, Query(description='IP 地址')] = None,
    pagination: PageData = DependsPagination,  # 添加
) -> ResponseSchemaModel[PageData[GetOperaLogDetail]]:
    log_select = await opera_log_service.get_select(username=username, status=status, ip=ip)
    page_data = await paging_data(db, log_select)
    return response_base.success(data=page_data)

schema/opera_log

fields = {}
fields['trace_id'] = (str, Field(description='追踪 ID'))
fields['username'] = (str | None, Field(None, description='用户名'))
fields['method'] = (str, Field(description='请求方法'))
fields['title'] = (str, Field(description='操作标题'))
fields['path'] = (str, Field(description='请求路径'))
fields['ip'] = (str, Field(description='IP 地址'))
fields['country'] = (str | None, Field(None, description='国家'))
fields['region'] = (str | None, Field(None, description='地区'))
fields['city'] = (str | None, Field(None, description='城市'))
fields['user_agent'] = (str, Field(description='用户代理'))
fields['os'] = (str | None, Field(None, description='操作系统'))
fields['browser'] = (str | None, Field(None, description='浏览器'))
fields['device'] = (str | None, Field(None, description='设备'))
fields['args'] = (dict[str, Any] | None, Field(None, description='请求参数'))
fields['status'] = (StatusType, Field(StatusType.enable, description='状态'))
fields['code'] = (str, Field(description='状态码'))
fields['msg'] = (str | None, Field(None, description='消息'))
fields['cost_time'] = (float, Field(description='耗时'))
fields['opera_time'] = (datetime, Field(description='操作时间'))

SchemaBaseOut = create_model('SchemaBaseOut', **fields)


class BaseSchema(BaseModel):
    model_config = ConfigDict(use_enum_values=True)


class GetOperaLogDetail(BaseSchema, SchemaBaseOut):
    """操作日志详情"""

    # model_config = ConfigDict(from_attributes=True)

    id: int = Field(description='日志 ID')
    created_time: datetime = Field(description='创建时间')

分页查询接口返回正常

我认为，你的问题出在 stmt = await menu_service.get_menu_list(permission_id=permission_id, request=request)，这里的查询应返回 Select，按照你的写法，当我修改 opera_log 为相同逻辑时

此时调用接口将直接报错：

{
  "code": 500,
  "msg": "'Select' object has no attribute 'scalars'",
  "data": null,
  "trace_id": "2d9c358b25004892b1be58e068c88ef0"
}

参考：https://uriyyo-fastapi-pagination.netlify.app/integrations/sqlalchemy/relationships/
分页查询应直接返回 sqla 的 select 查询

wang935415150 · 2025-06-06T08:58:26Z

    stmt = await menu_service.get_menu_list(permission_id=permission_id, request=request)
    a=await db.execute(stmt)
    sql_data = a.scalars.all()
    print(sql_data)  # 一切正常 能够查询到 自动生成的字段
    page_data = await paging_data(db, stmt)
    print(page_data) # 出现问题 无法解析正常的自动生成字段

怪我大佬是我写的不够细节了

wang935415150 · 2025-06-06T09:17:40Z

简言

首先我先向你表达一下歉意, 发你的 MRE还是错误的哈哈哈.其次我想我没有说清楚MRE中那里是重点
然后从代码中TODO 部分做为实验的分割点, 其中假设数据中n2 n3 插入了数据 ,我嗯就可以根据两次打印来发现问题

代码

import sys

from pathlib import Path

project_root = str(Path(__file__).resolve().parent.parent)
sys.path.append(project_root)


from typing import Optional

from fastapi import APIRouter, Body, Request
from fastapi_pagination.ext.sqlalchemy import paginate as paging_data
from pydantic import BaseModel, ConfigDict, Field, create_model
from sqlalchemy import Float, String,select
from sqlalchemy.orm import (
    Mapped,
    MappedAsDataclass,
    mapped_column,
)
from backend.database.db import CurrentSession
from backend.common.model import MappedBase, id_key
from backend.common.pagination import DependsPagination, PageData, paging_data
from backend.common.response.response_schema import ResponseModel, response_base,ResponseSchemaModel
from backend.common.schema import SchemaBase, SchemaBaseNoId
from backend.common.security.jwt import DependsJwtAuth

router = APIRouter()


class DataClassBase(MappedAsDataclass, MappedBase):
    """数据基础类 - 包含动态字段c1-c20和n1-n20"""
    __abstract__ = True
    id: Mapped[id_key] = mapped_column(init=False)


# 【！重要】动态字段生成 - 在类定义完成后立即执行
for i in range(1, 21):
    field_name = f'c{i}'
    setattr(DataClassBase, field_name, 
           mapped_column(String(20), init=True, nullable=True, 
                        insert_default=None, default=None, 
                        comment=f'字符串{i}'))

for i in range(1, 21):
    field_name = f'n{i}'
    setattr(DataClassBase, field_name, 
           mapped_column(Float, init=True, nullable=True, 
                        insert_default=None, default=None, 
                        sort_order=998, comment=f'浮点数{i}'))


class Menu(DataClassBase):
    test1: Mapped[str] = mapped_column(String(50), unique=True, comment='test1')
    test2: Mapped[str] = mapped_column(String(16), comment='test2')


class SchemaBaseNoId(BaseModel):
    model_config = ConfigDict(use_enum_values=True)


# 动态创建字段定义
fields = {}
for i in range(1, 21):
    field_name = f'c{i}'
    # 使用新版Pydantic的字段定义方式
    fields[field_name] = (Optional[str], Field(default=None))
for i in range(1, 21):
    field_name = f'n{i}'
    # 使用新版Pydantic的字段定义方式
    fields[field_name] = (Optional[float], Field(default=None))

# 使用 create_model 动态创建 SchemaBaseOut 类
SchemaBaseOut = create_model(
    'SchemaBaseOut',  # 模型名称
    **fields  # 字段名和对应的类型及默认值
)


class BaseSchema(BaseModel):
    model_config = ConfigDict(use_enum_values=True)


class GetMenuDetail(BaseSchema, SchemaBaseOut):
    test1: str = Field(..., description='测试')


class MenuSchema(BaseSchema, SchemaBaseOut):
    menus: list[GetMenuDetail] = Field(..., description='菜单列表')
    test1: str = Field(..., description='测试')

async def menu_service():
    stmt = select(Menu)
    return stmt


# TODO 假设现在数据库中针对 n2 n3 分别插入了数据 100 200

@router.post('/menu_list'
, summary='分页获取菜单的详情', dependencies=[
    DependsJwtAuth
    ])
async def get_menu_list(
    db: CurrentSession,
    request: Request,
    permission_id: int = Body(..., description='权限组id'),
    pagination: PageData = DependsPagination,
                          ) -> ResponseSchemaModel[PageData[GetMenuDetail]]:
    
    stmt = await menu_service()
    sql_excute = await db.execute(stmt)
    sql_data = sql_excute.scalars.all()
    if sql_data:
        for data in sql_data:
            print(data.test1)
            print(data.test2)
            print(data.n2)
            print(data.n3)

    page_data = await paging_data(db, stmt)
    for data in page_data.items:
        print(data.test1)
        print(data.test2)
        print(data.n2)
        print(data.n3)
    
    return response_base.success(data=page_data)

wu-clan · 2025-06-06T19:07:05Z

基于你提供的最新 MRE 的本地测试：

# 为了简化部分操作，接口做了部分修改
@router.post('/menu_list'
    , summary='分页获取菜单的详情')
async def get_menu_list(
        db: CurrentSession,
        request: Request,
        pagination: PageData = DependsPagination,
) -> ResponseSchemaModel[PageData[GetMenuDetail]]:
    stmt = await menu_service()
    sql_excute = await db.execute(stmt)
    sql_data = sql_excute.scalars().all()
    if sql_data:
        for data in sql_data:
            print(data)

    page_data = await paging_data(db, stmt)
    for data in page_data['items']:
        print(data)

    return response_base.success(data=page_data)

插入数据 sql:

insert into fba.menu (id, test1, test2, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14, c15, c16, c17, c18, c19, c20, n1, n2, n3, n4, n5, n6, n7, n8, n9, n10, n11, n12, n13, n14, n15, n16, n17, n18, n19, n20)
values  (1, '1', '1', null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, 1, 1, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null),
        (2, '2', '2', null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, 2, 2, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null);

接口返回结果：

{
  "code": 200,
  "msg": "请求成功",
  "data": {
    "items": [
      {
        "c1": null,
        "c2": null,
        "c3": null,
        "c4": null,
        "c5": null,
        "c6": null,
        "c7": null,
        "c8": null,
        "c9": null,
        "c10": null,
        "c11": null,
        "c12": null,
        "c13": null,
        "c14": null,
        "c15": null,
        "c16": null,
        "c17": null,
        "c18": null,
        "c19": null,
        "c20": null,
        "n1": null,
        "n2": null,
        "n3": null,
        "n4": null,
        "n5": null,
        "n6": null,
        "n7": null,
        "n8": null,
        "n9": null,
        "n10": null,
        "n11": null,
        "n12": null,
        "n13": null,
        "n14": null,
        "n15": null,
        "n16": null,
        "n17": null,
        "n18": null,
        "n19": null,
        "n20": null,
        "test1": "1"
      },
      {
        "c1": null,
        "c2": null,
        "c3": null,
        "c4": null,
        "c5": null,
        "c6": null,
        "c7": null,
        "c8": null,
        "c9": null,
        "c10": null,
        "c11": null,
        "c12": null,
        "c13": null,
        "c14": null,
        "c15": null,
        "c16": null,
        "c17": null,
        "c18": null,
        "c19": null,
        "c20": null,
        "n1": null,
        "n2": null,
        "n3": null,
        "n4": null,
        "n5": null,
        "n6": null,
        "n7": null,
        "n8": null,
        "n9": null,
        "n10": null,
        "n11": null,
        "n12": null,
        "n13": null,
        "n14": null,
        "n15": null,
        "n16": null,
        "n17": null,
        "n18": null,
        "n19": null,
        "n20": null,
        "test1": "2"
      }
    ],
    "total": 2,
    "page": 1,
    "size": 20,
    "total_pages": 1,
    "links": {
      "first": "/api/v1/logs/tn/menu_list?page=1&size=20",
      "last": "/api/v1/logs/tn/menu_list?page=1&size=20",
      "self": "/api/v1/logs/tn/menu_list?page=1&size=20",
      "next": null,
      "prev": null
    }
  }
}

打印结果：

Menu(id=1, test1='1', test2='1')
Menu(id=2, test1='2', test2='2')
{'id': 1, 'test1': '1', 'test2': '1'}
{'id': 2, 'test1': '2', 'test2': '2'}
2025-06-07 02:52:10.308 | INFO     |  5c82b5ae161f47348ef06b63a1f6ec91  | 127.0.0.1       | POST     | 200    | /api/v1/logs/tn/menu_list | 42.0ms

wu-clan · 2025-06-06T19:20:56Z

看起来paging_data 针对 pydantic的create_model 生成的 schema 解析是正常的

不过新的问题是，sqla 查询只返回了 id、test1、test2，但这似乎不是你真正遇到的问题？

wang935415150 · 2025-06-09T08:25:20Z

我的天天哪, 居然在你这里好使,好吧咱们关了这个 pr吧, 大概率是我本地的问题,

wu-clan · 2025-06-09T09:47:03Z

以上测试均基于当前最新代码，有可能你本地环境有所差异？

增加针对动态字段的数据解析

7d7b2d5

增加多层深层嵌套BaseModel的验证

28b1c85

wu-clan closed this Jun 9, 2025

Uh oh!

增加针对动态字段的数据解析 #652

增加针对动态字段的数据解析 #652

Uh oh!

Conversation

wang935415150 commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wu-clan commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

【！重要】动态字段生成 - 在类定义完成后立即执行

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wu-clan commented Jun 5, 2025

Uh oh!

wu-clan commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 5, 2025

Uh oh!

wu-clan commented Jun 5, 2025

Uh oh!

wu-clan commented Jun 5, 2025

Uh oh!

wang935415150 commented Jun 6, 2025

Uh oh!

wu-clan commented Jun 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wang935415150 commented Jun 6, 2025

Uh oh!

wang935415150 commented Jun 6, 2025

简言

代码

Uh oh!

wu-clan commented Jun 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

wu-clan commented Jun 6, 2025

Uh oh!

wang935415150 commented Jun 9, 2025

Uh oh!

wu-clan commented Jun 9, 2025

Uh oh!

Uh oh!

wu-clan commented Jun 6, 2025 •

edited

Loading

wu-clan commented Jun 6, 2025 •

edited

Loading