FastAPI - Pydantic | “输入应该是有效的字典或对象以从中提取字段”

Question

我使用

@model_validator

和

model_validate

来检查值是否与子类匹配：

示例请求负载：

{
  "name": "BC",
  "connectorType": "SNOWFLAKE",
  "productName": "bc",
  "secretArn": "foobararn",
  "warehouse": "foobarwarehouse",
  "tableSchema": "foobarschema",
  "database": "bardatabse"

}

雪花连接器类：

class SnowflakeConnector(SpotApiBaseModel):
    """Snowflake Connector."""
    model_config = ConfigDict(from_attributes=True)

    secretArn: str = Field(description="AWS secret arn")
    warehouse: str = Field(description="Snowflake warehouse. Required for connector type: SNOWFLAKE")
    database: str = Field(description="Database name. Required for connector type: SNOWFLAKE")
    tableSchema: str = Field(description="Database table schema. Required for connector type: SNOWFLAKE")


class ConnectorInterface(BaseConnector):

    @model_validator(mode='before')
    def check_allowed_products(cls, values):
        connector = values.get('connectorType').lower()
        if connector == "s3":
            S3Connector.model_validate(values)
        elif connector == "snowflake":
            SnowflakeConnector.model_validate(values)

当缺少任何必填字段时，我会得到预期的响应：

"Field required: ('body', 'database')"

但是当所有字段都存在时，我得到的不是正确的响应，而是：

"Input should be a valid dictionary or object to extract fields from: ('body',)"

例外：

{"time": "2024-06-18T10:46:45.641Z", "l": "ERROR", "rid": "2558770a-66ff-4da0-b8a2-977afc5043e6", "org": "", "act": "", "message": "An error occurred: [{'type': 'model_attributes_type', 'loc': ('body',), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': None, 'url': 'https://errors.pydantic.dev/2.6/v/model_attributes_type'}]", "component": {"name": "/app/common/exceptions/base_exception_handler.py", "version": "0.1.3"}, "exception": "Traceback (most recent call last):\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/starlette/middleware/exceptions.py\", line 68, in __call__\n    await self.app(scope, receive, sender)\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/fastapi/middleware/asyncexitstack.py\", line 20, in __call__\n    raise e\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/fastapi/middleware/asyncexitstack.py\", line 17, in __call__\n    await self.app(scope, receive, send)\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/starlette/routing.py\", line 718, in __call__\n    await route.handle(scope, receive, send)\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/starlette/routing.py\", line 276, in handle\n    await self.app(scope, receive, send)\n  File \"Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/starlette/routing.py\", line 66, in app\n    response = await func(request)\n               ^^^^^^^^^^^^^^^^^^^\n  File \"/Library/Caches/pypoetry/virtualenvs/service-Qx3_L5c2-py3.12/lib/python3.12/site-packages/fastapi/routing.py\", line 273, in app\n    raise RequestValidationError(_normalize_errors(errors), body=body)\nfastapi.exceptions.RequestValidationError: [{'type': 'model_attributes_type', 'loc': ('body',), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': None, 'url': 'https://errors.pydantic.dev/2.6/v/model_attributes_type'}]"}

模型已验证并请求继续其余逻辑

Answer 1

它适用于

mode='wrap'

 
    @model_validator(mode='wrap')
    def check_connector_params(self, data):
        if self.get('connectorType').lower() == "snowflake":
            SnowflakeConnector.model_validate(self)
   
        elif self.get('connectorType').lower() == "s3":
            S3Connector.model_validate(self)