timster / peewee-validates Goto Github PK

class Meta:
    constraints = [SQL('UNIQUE("name" COLLATE NOCASE, "email" COLLATE NOCASE)')]
    database = db`

class TestModel(peewee.Model):
    foo = peewee.ForeignKeyField(ModelOne)
    bar = peewee.ForeignKeyField(ModelTwo)
    class Meta:
        indexes = (
            (("foo", "bar"), True),
        )

    def perform_index_validation(self, data):
        """
        Validate any unique indexes specified on the model.
        This should happen after all the normal fields have been validated.
        This can add error messages to multiple fields.

        :return: None
        """
        # Build a list of dict containing query values for each unique index.
        index_data = []
        for columns, unique in self.instance._meta.indexes:
            if not unique:
                continue
            index_data.append({col: data.get(col, None) for col in columns})

        # Then query for each unique index to see if the value is unique.
        for index in index_data:
            _query = self.instance.filter(**index)_
            # If we have a primary key, need to exclude the current record from the check.
            if self.pk_field and self.pk_value:
                query = query.where(~(self.pk_field == self.pk_value))
            if query.count():
                err = ValidationError('index', fields=str.join(', ', index.keys()))
                for col in index.keys():
                    self.add_error(col, err)

Count(*) 
FROM testmodel AS t1 
INNER JOIN modelone AS t2 ON (t1.foo_id = t2.id) 
INNER JOIN modeltwo AS t3 ON (t1.bar_id = t3.id) 
WHERE ((t1.foo_id = 7) AND (t1.bar_id = (2, 4)));

    def perform_index_validation(self, data):
        """
        Validate any unique indexes specified on the model.
        This should happen after all the normal fields have been validated.
        This can add error messages to multiple fields.

        :return: None
        """
        # Build a list of dict containing query values for each unique index.
        index_data = []
        for columns, unique in self.instance._meta.indexes:
            if not unique:
                continue
            index_data.append({col: data.get(col, None) for col in columns})

        # Then query for each unique index to see if the value is unique.
        for index in index_data:
            new_index = {}
            for k, v in index.items():
                if issubclass(type(v), Iterable):
                    k = '%s__in' % k
                new_index[k] = v
            query = self.instance.filter(**new_index)
            # If we have a primary key, need to exclude the current record from the check.
            if self.pk_field and self.pk_value:
                if issubclass(type(self.pk_value), Iterable):
                    query = query.where(~(self.pk_field >> self.pk_value))
                else:
                    query = query.where(~(self.pk_field == self.pk_value))
            if query.count():
                err = ValidationError('index', fields=str.join(', ', index.keys()))
                for col in index.keys():
                    self.add_error(col, err)

def coerce(self, value):
     try:
         return int(value) if value else None
     except (TypeError, ValueError):
         raise ValidationError('coerce_int')

    def one_of_validator(field, data):
        if field.value is None:
            return
        options = values
        if callable(options):
            options = options()
        if field.value not in options:
            raise ValidationError('one_of', choices=str.join(', ', options))
    return one_of_validator