Python django.contrib.postgres.fields 模块，ArrayField() 实例源码

def generate_form(parsed):
    """
    Generate a django form from a parsed kliko object

    args:
        params: A parsed kliko file.
    returns:
        form_utils.forms.BetterForm
    """
    all_fields = {}
    fieldsets = []
    for section in parsed['sections']:
        fields_in_section = []

        for field in section['fields']:
            kwargs = {}
            for kwarg in ('initial', 'max_length', 'label', 'help_text', 'file', 'required'):
                if kwarg in field:
                    kwargs[kwarg] = field[kwarg]
            if 'choices' in field:
                kwargs['choices'] = field['choices'].items()
            fields_in_section.append(field['name'])

            match = list_regex.match(field['type'])
            if match:
                type_ = match.group(1)
                djangofield = field_map[match.group(1)]
                all_fields[field['name']] = ArrayField(djangofield, **kwargs)
            else:
                djangofield = field_map[field['type']]
                all_fields[field['name']] = djangofield(**kwargs)

        fieldsets.append((section['name'],
                          {'fields': fields_in_section,
                           'description': section['description']}
                          ))

    Meta = type('Meta', (), {'fieldsets': fieldsets})
    all_fields['Meta'] = Meta

    return type('Form', (BetterForm,), all_fields)

def __init__(self, model=None, field=None, choices=None):
        self.model = model
        if isinstance(field, ArrayField):
            self.field = field.base_field
            self.array = True
        else:
            self.field = field
            self.array = False
        self._choices = choices

def formfield(self, **kwargs):
        defaults = {
            'form_class': forms.MultipleChoiceField,
            'choices': self.base_field.choices,
        }
        defaults.update(kwargs)
        return super(ArrayField, self).formfield(**defaults)

def __init__(self, base_field, form_size=None, **kwargs):
        super(ArrayField, self).__init__(base_field, **kwargs)
        self.form_size = form_size

def formfield(self, **kwargs):
        if self.form_size or self.choices:
            defaults = {
                'form_class': SplitArrayFormField,
                'base_field': self.base_field.formfield(),
                'choices_form_class': TypedMultipleChoiceField,
                'size': self.form_size,
                'remove_trailing_nulls': True
            }
            if self.choices:
                defaults['coerce'] = self.base_field.to_python
            defaults.update(kwargs)
            return super(fields.ArrayField, self).formfield(**defaults)
        return super(ArrayField, self).formfield(**kwargs)

def deconstruct(self):
        name, path, args, kwargs = super(ArrayField, self).deconstruct()
        kwargs.update({
            'form_size': self.form_size,
        })
        return name, path, args, kwargs

def test_should_query_postgres_fields():
    from django.contrib.postgres.fields import IntegerRangeField, ArrayField, JSONField, HStoreField

    class Event(models.Model):
        ages = IntegerRangeField(help_text='The age ranges')
        data = JSONField(help_text='Data')
        store = HStoreField()
        tags = ArrayField(models.CharField(max_length=50))

    class EventType(DjangoObjectType):

        class Meta:
            model = Event

    class Query(graphene.ObjectType):
        event = graphene.Field(EventType)

        def resolve_event(self, info):
            return Event(
                ages=(0, 10),
                data={'angry_babies': True},
                store={'h': 'store'},
                tags=['child', 'angry', 'babies']
            )

    schema = graphene.Schema(query=Query)
    query = '''
        query myQuery {
          event {
            ages
            tags
            data
            store
          }
        }
    '''
    expected = {
        'event': {
            'ages': [0, 10],
            'tags': ['child', 'angry', 'babies'],
            'data': '{"angry_babies": true}',
            'store': '{"h": "store"}',
        },
    }
    result = schema.execute(query)
    assert not result.errors
    assert result.data == expected

def full_dehydrate(self, obj, for_list=False):
        """Convert the given object into a dictionary.

        :param for_list: True when the object is being converted to belong
            in a list.
        """
        if for_list:
            allowed_fields = self._meta.list_fields
            exclude_fields = self._meta.list_exclude
        else:
            allowed_fields = self._meta.fields
            exclude_fields = self._meta.exclude

        data = {}
        for field in self._meta.object_class._meta.fields:
            # Convert the field name to unicode as some are stored in bytes.
            field_name = str(field.name)

            # Skip fields that are not allowed.
            if allowed_fields is not None and field_name not in allowed_fields:
                continue
            if exclude_fields is not None and field_name in exclude_fields:
                continue

            # Get the value from the field and set it in data. The value
            # will pass through the dehydrate method if present.
            field_obj = getattr(obj, field_name)
            dehydrate_method = getattr(
                self, "dehydrate_%s" % field_name, None)
            if dehydrate_method is not None:
                data[field_name] = dehydrate_method(field_obj)
            else:
                value = field._get_val_from_obj(obj)
                if is_protected_type(value):
                    data[field_name] = value
                elif isinstance(field, ArrayField):
                    data[field_name] = field.to_python(value)
                else:
                    data[field_name] = field.value_to_string(obj)

        # Return the data after the final dehydrate.
        return self.dehydrate(obj, data, for_list=for_list)