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

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

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

【问题讨论】:

标签: json django postgresql jsonschema


【解决方案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参数,应该是:

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

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

        【讨论】:

          猜你喜欢
          • 2014-04-09
          • 2010-12-10
          • 2011-07-23
          • 2019-12-24
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 2011-06-08
          相关资源
          最近更新 更多