SQLAlchemy relationships field to Pydantic : Validation Erro

Question

我有一些使用SQLAlchemy声明基声明的模型。它们的字段代表一些IP地址。当我试图通过orm_mode将这些模型的实例转换为pydantic模型时，它失败了，出现以下错误

E   pydantic.error_wrappers.ValidationError: 4 validation errors for IpSchema
E   ip_address -> 0
E     value is not a valid IPv4 address (type=value_error.ipv4address)
E   ip_address -> 0
E     value is not a valid IPv6 address (type=value_error.ipv6address)
E   ip_address -> 0
E     value is not a valid IPv4 or IPv6 address (type=value_error.ipvanyaddress)
E   ip_address -> 0
E     str type expected (type=type_error.str)

代码如下。我试过用pytest检查，但是失败了。

orm_mode代码可以被覆盖吗?

from typing import List, Union
from pydantic import BaseModel, Field, IPvAnyAddress
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")

class IpSchema(BaseModel):
ip_address: List[Union[IPv4Address, IPv6Address, IPvAnyAddress]] = 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)
assert str(s.ip_address[0]) == ipv4

我该如何解决这个问题?

Answer 1

Pydantic不知道如何将每个关系ORM实例映射到它的地址字段。为此，您需要使用pre=True参数添加一个pydantic验证器，以便在 pydantic验证之前将每个ORM实例映射到地址字段。

应该是这样的

class IpSchema(BaseModel):
ip_address: List[Union[IPv4Address, IPv6Address, IPvAnyAddress]] = Field()
class Config:
orm_mode = True
@validator('ip_address', pre=True)
def validate(cls, ip_adress_relationship, **kwargs):
return [ip.address for ip in ip_adress_relationship]

请注意，pre=True的验证器在之前运行后和设置值为Pydantic模型。在您的示例中，它没有改变任何东西，但是，例如，如果您想将ip列表转换为str，则需要首先检查值的类型:

class IpSchema(BaseModel):
ip_address: str
class Config:
orm_mode = True
@validator('ip_address', pre=True)
def validate(cls, ip_adress_relationship, **kwargs):
if isinstance(ip_adress_relationship, str):
return ip_adress_relationship
return ','.join([ip.address for ip in ip_adress_relationship])

这里是完整的可复制(和工作)的例子:

from typing import List, Union
from pydantic import BaseModel, Field, IPvAnyAddress
from pydantic import validator
from pydantic.schema import IPv4Address
from pydantic.schema import 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")

class IpSchema(BaseModel):
ip_address: List[Union[IPv4Address, IPv6Address, IPvAnyAddress]] = Field()
class Config:
orm_mode = True
@validator('ip_address', pre=True)
def validate(cls, ip_adress_relationship, **kwargs):
return [ip.address for ip in ip_adress_relationship]

def test_ipv4():
ipv4: str = "192.168.1.1"
ip = IpAddress(address=ipv4)
m = IpModel(ip_address=[ip])
s = IpSchema.from_orm(m)
assert str(s.ip_address[0]) == ipv4
if __name__ == '__main__':
test_ipv4()