主页 / 问题 / 1539114
Accepted
virvaldium
Asked:2023-09-05 05:43:37 +0000 UTC2023-09-05 05:43:37 +0000 UTC

如何检查 Pydantic 2 模型中是否存在主键

  • 772

我有一个用 Pydantic 2 编写的电路，用于验证提交的字段：

from pydantic import BaseModel

class CommandSchema(BaseModel):
    product: int
    title: str

如何检查发送的产品字段是否是产品模型中的键？这个想法是使用类似以下代码的内容：

class CommandSchema(BaseModel):
   product: int
   title: str

    @validator('product')
    def validate_product_field(cls, value):
        session = Session(bind=engine)
        if session.query(Product).filter(Product.id == value).first():
            return value
       
        raise ValueError(f"Field '{value}' already not exists in the database.")

但我使用异步连接数据库。因此，我不能等待这样的代码：

async with engine.begin() as conn:
            async_session = sessionmaker(conn, class_=AsyncSession, expire_on_commit=False)
            async with async_session() as session:
                if await session.execute(Product.select().where(Product.id == value)):
                    return value
                    
            raise ValueError(f"Field '{value}' already not exists in the database.")

因为它将在非异步方法内启动

python

1 个回答

  1. Best Answer

    尝试使用pydantic-async-validation。

    安装

    pip install pydantic-async-validation

    使用示例

    import pydantic
from pydantic_async_validation import async_field_validator, AsyncValidationModelMixin


class SomethingModel(AsyncValidationModelMixin, pydantic.BaseModel):
    name: str

    @async_field_validator('name')
    async def validate_name(self, value: str) -> None:
        if value == "invalid":
            raise ValueError("Invalid name")


valid_instance = SomethingModel(name="valid")
await valid_instance.model_async_validate()

invalid_instance = SomethingModel(name="invalid")
await invalid_instance.model_async_validate()
    • 2

相关问题