如何使用 JSONField 和 Pydantic 在 Django 中构建灵活的数据模型

在本文中，我将引导您了解如何使用 Django 的 JSONField（`JSON` 和 `JSONB` 包装器）来建模半结构化数据，以及如何使用 Pydantic 对该数据强制实施模式——这种方法对于 Python Web 开发人员来说应该很自然。

灵活的类型定义

让我们考虑一个处理付款的系统，例如“交易”表。它看起来像这样：

from django.db import models

class Transaction(models.Model):
    # Other relevant fields...
    payment_method = models.JSONField(default=dict, null=True, blank=True)

我们的重点是“payment_method”字段。在实际情况下，我们将有现有的处理付款的方法：

我们的系统必须能够适应存储每种支付方式所需的特定数据，同时保持一致且可验证的结构。

我们将使用 Pydantic 为不同的付款方式定义精确的模式：

from typing import Optional
from pydantic import BaseModel

class CreditCardSchema(BaseModel):
    last_four: str
    expiry_month: int
    expiry_year: int
    cvv: str


class PayPalSchema(BaseModel):
    email: EmailStr
    account_id: str


class CryptoSchema(BaseModel):
    wallet_address: str
    network: Optional[str] = None


class BillingAddressSchema(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str
    state: Optional[str] = None


class PaymentMethodSchema(BaseModel):
    credit_card: Optional[CreditCardSchema] = None
    paypal: Optional[PayPalSchema] = None
    crypto: Optional[CryptoSchema] = None
    billing_address: Optional[BillingAddressSchema] = None

这种方法有几个显著的好处：

为了在我们的“payment_method”字段上强制执行模式，我们利用 Pydantic 模型来确保传递给该字段的任何数据都符合我们定义的模式。

from typing import Optional, Mapping, Type, NoReturn
from pydantic import ValidationError as PydanticValidationError
from django.core.exceptions import ValidationError

def payment_method_validator(value: Optional[dict]) -> Optional[Type[BaseModel] | NoReturn]:
    if value is None:
        return

    if not isinstance(value, Mapping):
        raise TypeError("Payment method must be a dictionary")

    try:
        PaymentMethodSchema(**value)
    except (TypeError, PydanticValidationError) as e:
        raise ValidationError(f"Invalid payment method: {str(e)}")

在这里，我们执行一些检查，以确保输入验证器的数据属于正确类型，以便 Pydantic 可以对其进行验证。对于可空值，我们不执行任何操作，如果传入的值不是 Mapping 类型的子类（例如 `Dict` 或 `OrderedDict`），则会引发类型错误。

当我们使用传递给构造函数的值创建 Pydantic 模型的实例时。如果值的结构不符合 `PaymentMethodSchema` 定义的模式，Pydantic 将引发验证错误。例如，如果我们为 `PayPalSchema` 中的 `email` 字段传递了无效的电子邮件值，Pydantic 将引发如下验证错误：

ValidationError: 1 validation error for PaymentMethodSchema
paypal.email
  value is not a valid email address: An email address must have an @-sign. [type=value_error, input_value='Check me out on LinkedIn: https://linkedin.com/in/daniel-c-olah', input_type=str]

我们可以通过两种方式强制执行此验证：

我们甚至可以在将来轻松添加“pay_later”方法。字段的类型也可以改变，而且我们不会面临数据库字段迁移限制，比如从整数主键迁移到 UUID 主键时遇到的限制。您可以在此处查看完整代码，以完全理解该概念。

反规范化

非规范化涉及故意在多个文档或集合中复制数据，以优化性能和可扩展性。这种方法与传统关系数据库中使用的严格规范化形成鲜明对比，NoSQL 数据库通过引入灵活的面向文档的存储范例，在推广非规范化方面发挥了重要作用。

考虑一个电子商务场景，其中产品和订单有单独的表。当客户下订单时，必须捕获购物车中包含的产品详细信息的快照。我们不会引用当前产品记录（这些记录可能会因更新或删除而随时间而变化），而是将产品信息直接存储在订单中。这可确保订单保留其原始上下文和完整性，反映购买时产品的确切状态。非规范化在实现这种一致性方面起着至关重要的作用。

一种可能的方法可能是在订单表中复制一些产品字段。但是，这种方法可能会带来可扩展性挑战并损害订单架构的凝聚力。更有效的解决方案是将相关产品字段序列化为 JSON 结构，从而允许订单维护产品的独立记录，而无需依赖外部查询。以下代码说明了此技术：

from uuid import UUID
from typing import Optional, Any
from pydantic import constr, BaseModel, ValidationError as PydanticValidationError
from django.db import models
from django.core.exceptions import Validation as ValidationError

class FrozenProductSchema(BaseModel):
    id: UUID
    type: Any
    name: constr(min_length=2, strict=True)
    description = ""
    pricing: dict

class FrozenProductsSchema(BaseModel):
    products: list[FrozenProductSchema]

def frozen_products_validator(value: Optional[list[dict]]):
    if not value:
        return
    try:
        FrozenProductsSchema({"products": value})
    except PydanticValidationError as e:
        raise ValidationError(e.errors())

class Order(models.Model):
    # Other fields 
    products = models.JSONField(
            help_text="Validated snapshot of product details at time of order",
            validators=[frozen_products_validator]
            null=True,
            blank=True,
    )

由于我们在上一节中已经介绍了大部分概念，您应该开始了解 Pydantic 在所有这些方面的作用。在上面的示例中，我们使用 Pydantic 来验证与订单相关的产品列表。通过为产品结构定义架构，Pydantic 可确保添加到订单中的每个产品都符合预期要求。如果提供的数据不符合架构，Pydantic 会引发验证错误。

在 Django 中查询 JSONField

我们可以像在 Django 字段中执行查找一样查询“JSONField”键。以下是根据我们的用例给出的几个示例。

# Find transactions with credit card payments
credit_card_transactions = Transaction.objects.filter(
    payment_method__credit_card__isnull=False
)

# Find transactions from a specific country in billing address
us_transactions = Transaction.objects.filter(
    payment_method__billing_address__country='USA'
)

# Complex nested filtering
complex_filter = Transaction.objects.filter(
    payment_method__credit_card__last_four__startswith='4111',
    payment_method__billing_address__city='New York'
)

# Orders where any product has id = 1
orders = Order.objects.filter(products__contains=[{"id": 1}])

您可以查看文档以了解有关过滤 JSON 字段的更多信息。

结论

在 PostgreSQL 中使用 JSON 和 JSONB 为处理关系数据库中的半结构化数据提供了极大的灵活性。Pydantic 和 Django 的“JSONField”等工具有助于强制执行数据结构规则，从而更容易保持准确性并适应变化。但是，这种灵活性需要谨慎使用。如果没有适当的规划，随着数据随时间变化，它可能会导致性能下降或不必要的复杂性。

在 Django 中，仅当显式调用 full_clean() 时才会触发字段验证器 — 这通常在使用 Django Forms 或在 DRF 序列化器上调用 is_valid() 时发生。有关更多详细信息，您可以参考 Django 验证器文档。

解决此问题的更高级方法是实现一个自定义 Django 字段，该字段集成了 Pydantic 来内部处理 JSON 数据的序列化和验证。虽然这需要一篇专门的文章，但目前，您可以探索为该问题提供现成解决方案的库，例如：django-pydantic-jsonfield