python-3.x SQLAlchemy关系字段到Pydantic:验证错误

zpgglvta  于 2023-07-01  发布在  Python
关注(0)|答案(1)|浏览(170)

我用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

我该如何解决这个问题?

pwuypxnk

pwuypxnk1#

Pydantic不知道如何将每个关系ORM示例Map到其地址字段。为此,您需要添加一个pydantic验证器,并使用pre=True参数,以便将每个ORM示例Map到地址字段 before pydantic validation。
下面是它应该是什么样子

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()

相关问题