SQLAlchemy关联IP字段转Pydantic报ValidationError如何解决
问题根因
报错来自两个核心问题:
- Pydantic开启
orm_mode时,只会按字段名直接从ORM实例取同名属性做类型转换。你的ip_address字段声明为IP类型列表,但SQLAlchemy的ip_address关系返回的是IpAddressORM实例列表,不是存IP字符串的字段值,Pydantic拿到ORM对象而非字符串,自然触发IP格式校验、字符串类型校验失败。 - 原代码漏导入
IPv4Address、IPv6Address两个类型,即使逻辑正确运行时也会抛出导入错误。
解决方案
不需要重写orm_mode的底层逻辑,官方已经提供了两种适配方式:
方案1:定义嵌套Pydantic模型(推荐)
这是最符合Pydantic设计规范的写法,给关联表单独定义对应Schema,自动完成关联实例的字段提取和类型转换:
from typing import List, Union # 补全缺失的类型导入 from pydantic import BaseModel, Field, IPvAnyAddress, IPv4Address, IPv6Address from sqlalchemy import INTEGER, Column, ForeignKey, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() class IpModel(Base): __tablename__ = "ip_model" id = Column(INTEGER, primary_key=True, autoincrement=True, index=True) ip_address = relationship("IpAddress", back_populates="ip_model") class IpAddress(Base): __tablename__ = "ip" id = Column(INTEGER, primary_key=True, autoincrement=True, index=True) address = Column(String(64), nullable=False) ip_model_id = Column(INTEGER, ForeignKey("ip_model.id"), nullable=False) ip_model = relationship("IpModel", back_populates="ip_address") # 定义关联IP表对应的Schema class IpAddressSchema(BaseModel): address: IPvAnyAddress class Config: orm_mode = True class IpSchema(BaseModel): # 字段类型改为嵌套Schema列表 ip_address: List[IpAddressSchema] = Field() class Config: orm_mode = True def test_ipv4(): ipv4: str = "192.168.1.1" ip = IpAddress(address=ipv4) m = IpModel(ip_address=[ip]) s = IpSchema.from_orm(m) # 取值时取嵌套对象的address属性即可 assert str(s.ip_address[0].address) == ipv4
方案2:自定义GetterDict适配原有Schema结构
如果你不想修改原有Schema的字段结构,希望ip_address直接返回IP对象列表,可以通过Pydantic内置的GetterDict自定义ORM取值逻辑:
from typing import Any, List from pydantic import BaseModel, Field, IPvAnyAddress from pydantic.utils import GetterDict from sqlalchemy import INTEGER, Column, ForeignKey, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() # ORM模型定义和上文一致,此处省略重复代码 class IpModel(Base): __tablename__ = "ip_model" id = Column(INTEGER, primary_key=True, autoincrement=True, index=True) ip_address = relationship("IpAddress", back_populates="ip_model") class IpAddress(Base): __tablename__ = "ip" id = Column(INTEGER, primary_key=True, autoincrement=True, index=True) address = Column(String(64), nullable=False) ip_model_id = Column(INTEGER, ForeignKey("ip_model.id"), nullable=False) ip_model = relationship("IpModel", back_populates="ip_address") # 自定义ORM取值逻辑 class IpModelGetter(GetterDict): def get(self, key: str, default: Any = None) -> Any: if key == "ip_address": # 从关联实例中提取IP字符串返回 return [ip.address for ip in self._obj.ip_address] return super().get(key, default) class IpSchema(BaseModel): # IPvAnyAddress本身就同时支持IPv4/IPv6校验,无需嵌套Union ip_address: List[IPvAnyAddress] = Field() class Config: orm_mode = True # 绑定自定义取值类 getter_dict = IpModelGetter def test_ipv4(): ipv4: str = "192.168.1.1" ip = IpAddress(address=ipv4) m = IpModel(ip_address=[ip]) s = IpSchema.from_orm(m) # 可以直接取列表内的IP对象,和你原有代码逻辑完全匹配 assert str(s.ip_address[0]) == ipv4
补充说明
- 第一种方案结构清晰,适合关联表有多个字段需要返回的场景,后续扩展字段更方便。
- 第二种方案不需要修改上层调用的取值逻辑,适配成本更低。
内容的提问来源于stack exchange,提问作者tome119
相关产品推荐
相关产品推荐

