Django PostgreSQL JSON字段模式验证

Question

我有一个Django模型，其中包含一个JSONField（django.contrib.postgres.fields.JSONField）。

是否有任何方法可以根据JSON模式文件验证模型数据？

（保存前）
类似于my_field = JSONField（schema_file = my_schema_file）的方式。

- Mr T.

---

@e4c5 在模型的.save()方法中编写验证逻辑是一种反模式。模型验证应该存在于模型的验证器中，并通过在调用.save()之前调用Model.full_clean()进行调用。只有在验证检查通过后才应该调用.save()。文档：https://docs.djangoproject.com/en/2.2/ref/models/instances/#validating-objects 更多信息：https://dev59.com/T6jja4cB1Zd3GeqP6imX - user1847

Answer 1

我使用jsonschema编写了一个自定义验证器来实现这一功能。

project/validators.py

import django
from django.core.validators import BaseValidator
import jsonschema
    

class JSONSchemaValidator(BaseValidator):
    def compare(self, value, schema):
        try:
            jsonschema.validate(value, schema)
        except jsonschema.exceptions.ValidationError:
            raise django.core.exceptions.ValidationError(
                '%(value)s failed JSON schema check', params={'value': value}
            )

项目/app/models.py

from django.db import models

from project.validators import JSONSchemaValidator

MY_JSON_FIELD_SCHEMA = {
    'schema': 'http://json-schema.org/draft-07/schema#',
    'type': 'object',
    'properties': {
        'my_key': {
            'type': 'string'
        }
    },
    'required': ['my_key']
}

class MyModel(models.Model):
    my_json_field = models.JSONField(
        default=dict,
        validators=[JSONSchemaValidator(limit_value=MY_JSON_FIELD_SCHEMA)]
    )

- Corwin Cole

---

对于那些这个方法没有完全起作用的人，我不得不添加这个函数 Model.clean_fields() ，然后根据更改添加 Model.save()。 - Jose

如果您在验证深度嵌套的JSON结构时遇到性能问题，请查看fastjsonschema。对我们来说，加速非常惊人。 - phoenix

Answer 2

您可以使用 Cerberus 对数据进行模式验证。

from cerberus import Validator

schema = {'name': {'type': 'string'}}
v = Validator(schema)
data = {'name': 'john doe'}
v.validate(data)  # returns "True" (if passed)
v.errors  # this would return the error dict (or on empty dict in case of no errors)

这个工具非常易用（也得益于它的良好文档）。您可以在http://docs.python-cerberus.org/en/stable/validation-rules.html中查看验证规则。

- udo

Answer 3

这就是Model.clean()方法的作用(参见文档)。例如：

class MyData(models.Model):
    some_json = JSONField()
    ...

    def clean(self):
        if not is_my_schema(self.some_json):
            raise ValidationError('Invalid schema.')

- C14L

---

但是清理方法不会自动调用。对吗？ - Nimesh Kumar

Answer 4

我编写了一个自定义的JSONField，扩展了models.JSONField，并使用jsonschema（Django 3.1，Python 3.7）验证属性值。我没有使用validators参数，原因是我想让用户动态定义模式。因此，我使用了一个schema参数，应该是以下内容之一：

1. None（默认值）：该字段将像其父类一样运行（不支持JSON模式验证）。
2. 一个dict对象。此选项适用于小型模式定义（例如：{"type": "string"}）；
3. 描述包含模式代码的文件路径的str对象。此选项适用于大型模式定义（以保留模型类定义代码的美观）。对于搜索，我使用所有启用的查找器：django.contrib.staticfiles.finders.find()。
4. 一个以模型实例作为参数并返回dict对象模式的函数。因此，您可以基于给定模型实例的状态构建模式。每次调用validate()时都会调用该函数。

myapp/models/fields.py

import json

from jsonschema import validators as json_validators
from jsonschema import exceptions as json_exceptions

from django.contrib.staticfiles import finders
from django.core import checks, exceptions
from django.db import models
from django.utils.functional import cached_property


class SchemaMode:
    STATIC = 'static'
    DYNAMIC = 'dynamic'


class JSONField(models.JSONField):
    """
    A models.JSONField subclass that supports the JSON schema validation.
    """
    def __init__(self, *args, schema=None, **kwargs):
        if schema is not None:
            if not(isinstance(schema, (bool, dict, str)) or callable(schema)):
                raise ValueError('The "schema" parameter must be bool, dict, str, or callable object.')
            self.validate = self._validate
        else:
            self.__dict__['schema_mode'] = False
        self.schema = schema
        super().__init__(*args, **kwargs)

    def check(self, **kwargs):
        errors = super().check(**kwargs)
        if self.schema_mode == SchemaMode.STATIC:
            errors.extend(self._check_static_schema(**kwargs))
        return errors

    def _check_static_schema(self, **kwargs):
        try:
            schema = self.get_schema()
        except (TypeError, OSError):
            return [
                checks.Error(
                    f"The file '{self.schema}' cannot be found.",
                    hint="Make sure that 'STATICFILES_DIRS' and 'STATICFILES_FINDERS' settings "
                         "are configured correctly.",
                    obj=self,
                    id='myapp.E001',
                )
            ]
        except json.JSONDecodeError:
            return [
                checks.Error(
                    f"The file '{self.schema}' contains an invalid JSON data.",
                    obj=self,
                    id='myapp.E002'
                )
            ]

        validator_cls = json_validators.validator_for(schema)

        try:
            validator_cls.check_schema(schema)
        except json_exceptions.SchemaError:
            return [
                checks.Error(
                    f"{schema} must be a valid JSON Schema.",
                    obj=self,
                    id='myapp.E003'
                )
            ]
        else:
            return []

    def deconstruct(self):
        name, path, args, kwargs = super().deconstruct()
        if self.schema is not None:
            kwargs['schema'] = self.schema
        return name, path, args, kwargs

    @cached_property
    def schema_mode(self):
        if callable(self.schema):
            return SchemaMode.DYNAMIC
        return SchemaMode.STATIC

    @cached_property
    def _get_schema(self):
        if callable(self.schema):
            return self.schema
        elif isinstance(self.schema, str):
            with open(finders.find(self.schema)) as fp:
                schema = json.load(fp)
        else:
            schema = self.schema
        return lambda obj: schema

    def get_schema(self, obj=None):
        """
        Return schema data for this field.
        """
        return self._get_schema(obj)

    def _validate(self, value, model_instance):
        super(models.JSONField, self).validate(value, model_instance)
        schema = self.get_schema(model_instance)
        try:
            json_validators.validate(value, schema)
        except json_exceptions.ValidationError as e:
            raise exceptions.ValidationError(e.message, code='invalid')

Usage: myapp/models/__init__.py

def schema(instance):
    schema = {}
    # Here is your code that uses the other
    # instance's fields to create a schema.
    return schema


class JSONSchemaModel(models.Model):
    dynamic = JSONField(schema=schema, default=dict)
    from_dict = JSONField(schema={'type': 'object'}, default=dict)

    # A static file: myapp/static/myapp/schema.json
    from_file = JSONField(schema='myapp/schema.json', default=dict)

- Sultan

Answer 5

对于简单情况，可以使用jsonschema来解决问题。

class JSONValidatedField(models.JSONField):
    def __init__(self, *args, **kwargs):
        self.props = kwargs.pop('props')
        self.required_props = kwargs.pop('required_props', [])
        super().__init__(*args, **kwargs)

    def validate(self, value, model_instance):
        try:
            jsonschema.validate(
                value, {
                    'schema': 'http://json-schema.org/draft-07/schema#',
                    'type': 'object',
                    'properties': self.props,
                    'required': self.required_props
                }
            )
        except jsonschema.exceptions.ValidationError:
            raise ValidationError(
                    f'Value "{value}" failed schema validation.')


class SomeModel(models.Model):
    my_json_field = JSONValidatedField(
            props={
                'foo': {'type': 'string'}, 
                'bar': {'type': 'integer'}
            }, 
            required_props=['foo'])

- Mark Mishyn