django postgresql json 字段模式验证

Posted

技术标签:

【中文标题】django postgresql json 字段模式验证【英文标题】:django postgresql json field schema validation 【发布时间】:2016-10-05 04:39:52 【问题描述】:

我有一个带有JSONField (django.contrib.postgres.fields.JSONField) 的 django 模型 有什么方法可以根据 json 模式文件验证模型数据吗?

(预保存) 类似my_field = JSONField(schema_file=my_schema_file)

【问题讨论】:

@e4c5 将验证逻辑写入模型的 .save() 方法是一种反模式。模型验证应该存在于模型的验证器中,并在调用 .save() 之前通过 Model.full_clean() 调用。 .save() 应该在验证检查通过后调用。文档:docs.djangoproject.com/en/2.2/ref/models/instances/… 更多信息:***.com/questions/48220104/… 【参考方案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
            )

project/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)]
    )

【讨论】:

对于那些不能完全工作的人,我必须根据变化添加这个函数 Model.clean_fields() 然后 Model.save() 【参考方案2】:

这就是 Model.clean() 方法的用途 (see docs)。示例:

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

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

【讨论】:

【参考方案3】:

您可以使用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

【讨论】:

【参考方案4】:

我编写了一个自定义的JSONField,它扩展了models.JSONField,并使用jsonschema(Django 3.1、Python 3.7)验证了属性的值。

我没有使用validators参数有一个原因:我想让用户动态定义架构。所以我使用schema参数,应该是:

    None(默认):该字段的行为与其父类相似(不支持 JSON 模式验证)。 dict 对象。此选项适用于小型架构定义(例如:"type": "string"); str 对象,描述了包含架构代码的文件的路径。此选项适用于大模式定义(以保留模型类定义代码的美感)。对于搜索,我使用所有启用的查找器:django.contrib.staticfiles.finders.find()。 一个将模型实例作为参数并将架构作为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')

用法: 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)

【讨论】:

【参考方案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'])

【讨论】:

以上是关于django postgresql json 字段模式验证的主要内容,如果未能解决你的问题,请参考以下文章

如何在 Django ORM 中映射 PostgreSQL 数组字段

尝试在 Django 中使用外键列表创建 PostgreSQL 字段

我的 Django 模型中的一个字段已经存在于 PostgreSQL 数据库中

PostgreSQL 字段中的 Django ProgrammingError

检查 json 类型列 PostgreSQL 中是不是存在字段

PostgreSQL Json字段作为查询条件案例