Python mongoengine 模块，ValidationError() 实例源码

def save_model_action(model_action_data, access, process):
    from audit_tools.audit.models import Access
    from audit_tools.audit.models.models_factory import create_model_action
    try:
        logger.debug("Pre save ModelAction")
        try:
            a = Access.objects.get(id=access.id)
            m = create_model_action(model_action_data, a, process)
        except AttributeError:
            m = create_model_action(model_action_data, None, process)
        except ValidationError:
            access.save()
            a = Access.objects.get(id=access.id)
            m = create_model_action(model_action_data, a, process)
        m.save()
        logger.debug("Post save ModelAction: %s", m.id)
    except:
        logger.exception("Error saving ModelAction document")

    return True

def create_zone(self, zone, **kwargs):
        """
        This is the public method that is called to create a new DNS zone.
        """
        self._create_zone__prepare_args(kwargs)
        pr_zone = self._create_zone__for_cloud(**kwargs)
        # Set fields to cloud model and perform early validation.
        zone.zone_id = pr_zone.id
        zone.domain = pr_zone.domain
        zone.type = pr_zone.type
        zone.ttl = pr_zone.ttl
        zone.extra = pr_zone.extra
        # Attempt to save.
        try:
            zone.save()
        except me.ValidationError as exc:
            log.error("Error updating %s: %s", zone, exc.to_dict())
            raise BadRequestError({'msg': exc.message,
                                   'errors': exc.to_dict()})
        except me.NotUniqueError as exc:
            log.error("Zone %s not unique error: %s", zone, exc)
            raise ZoneExistsError()
        self.cloud.owner.mapper.update(zone)

def init_mappings(self):
        """Initialize RBAC Mappings.

        RBAC Mappings always refer to a (Organization, Team) combination.

        In order to reference the Organization instance of this Team, we
        are using `self._instance`, which is a proxy object to the upper
        level Document.

        """
        if not RBACMapping:
            return
        if self.name == 'Owners':
            return
        if RBACMapping.objects(org=self._instance.id, team=self.id).only('id'):
            raise me.ValidationError(
                'RBAC Mappings already initialized for Team %s' % self
            )
        for perm in ('read', 'read_logs'):
            RBACMapping(
                org=self._instance.id, team=self.id, permission=perm
            ).save()

def create_org_for_user(user, org_name='', promo_code=None, token=None,
                        selected_plan=None):
    org = Organization(name=org_name, selected_plan=selected_plan)
    org.add_member_to_team('Owners', user)
    org.name = org_name
    try:
        org.save()
    except ValidationError as e:
        raise BadRequestError({"msg": e.message, "errors": e.to_dict()})
    except OperationError:
        raise OrganizationOperationError()

    # assign promo if applicable
    if promo_code or token:
        assign_promo(org, promo_code, token)
    return org

def update_whitelist_ips(auth_context, ips):
    """
    This function takes a list of dicts in the form:
    [{cidr:'cidr1', 'description:'desc1'},
    {cidr:'cidr2', 'description:'desc2'}]
    and saves them in the User.ips field.
    """

    user = auth_context.user

    user.ips = []
    for ip_dict in ips:
        wip = WhitelistIP()
        wip.cidr = ip_dict['cidr']
        wip.description = ip_dict['description']
        user.ips.append(wip)

    try:
        user.save()
    except ValidationError as e:
        raise BadRequestError({"msg": e.message, "errors": e.to_dict()})

def clean(self):
        """Ensures that self is a valid RSA keypair."""

        from Crypto import Random
        Random.atfork()

        if 'RSA' not in self.private:
            raise me.ValidationError("Private key is not a valid RSA key.")

        # Generate public key from private key file.
        try:
            key = RSA.importKey(self.private)
            self.public = key.publickey().exportKey('OpenSSH')
        except Exception:
            log.exception("Error while constructing public key "
                          "from private.")
            raise me.ValidationError("Private key is not a valid RSA key.")

def clean(self):
        """Custom validation for GCE Networks.

        GCE enforces:

            - Regex constrains on network names.
            - CIDR assignment only if `legacy` mode has been selected.

        """
        if self.mode == 'legacy':
            super(GoogleNetwork, self).clean()
        elif self.cidr is not None:
            raise me.ValidationError('CIDR cannot be set for modes other than '
                                     '"legacy" - Current mode: %s' % self.mode)

        if not re.match('^(?:[a-z](?:[-a-z0-9]{0,61}[a-z0-9])?)$', self.name):
            raise me.ValidationError('A **lowercase** name must be specified')

def user_toggle(user_id=None, action=None):
    try:
        user = umodels.User.get_user(user_id)
        if action == "remove":
            user.delete()
            flash.success("User \"{}\" was successfully deleted!"
                          "".format(str(user)))
        elif action == "is_admin":
            user.is_admin = not user.is_admin
            user.save()
            flash.success("User \"{}\" field \"{}\" was successfully "
                          "updated to \"{}\"!".format(str(user), action,
                                                      user.is_admin))
    except (me.DoesNotExist, me.ValidationError) as e:
        flash.warning("User with id \"{}\" does not exist."
                      "".format(user_id))

    return redirect(url_for("admin.home"))

def index_stock(stockid):
    try:
        page = int(request.args.get('page')) if request.args.get('page') else 1
        stock = models.Stock.objects.filter(id=stockid, blacklisted=False).first()
        if not stock:
            return redirect(url_for('index'))
        leaders = models.get_leaders()
        return render_template('index.html',
            view="market",
            base_url=settings.SERVER_NAME,
            leaders=leaders,
            stock=stock,
            stocks=api_views.get_paged_stocks(page),
            page=page)
    except DoesNotExist as e:
        return redirect(url_for('index'))
    except ValidationError as e:
        return redirect(url_for('index'))

def test_save_with_access_not_saved(self, logger, access_klass, create_model_action):
        model_action = MagicMock()
        model_action.save.return_value = True
        create_model_action.return_value = model_action
        access = MagicMock()
        access.save.return_value = True
        access_klass.objects.get.side_effect = (ValidationError, access)

        result = tasks.save_model_action(model_action, access, None)

        self.assertEqual(access.save.call_count, 1)
        self.assertTrue(result)

def get_or_404(cls, *args, **kwargs):
        message = kwargs.pop('message', None)
        try:
            return cls.objects.get(*args, **kwargs)
        except (MultipleObjectsReturned, DoesNotExist, ValidationError):
            if message is not None:
                abort(404, message=message)
            abort(404)

def validates(model):
        try:
            model.validate(clean=True)
        except ValidationError:
            return False
        return True

def validates(model):
        try:
            model.validate(clean=True)
        except ValidationError:
            return False
        return True

def clean(self):
        """Overriding the default clean method to implement param checking"""
        super(ARecord, self).clean()
        try:
            ip_addr = self.rdata[0].decode('utf-8')
            ip.ip_address(ip_addr)
        except ValueError:
            raise me.ValidationError('IPv4 address provided is not valid')
        if not len(self.rdata) == 1:
            raise me.ValidationError('We cannot have more than one rdata'
                                     'values for this type of record.')

def clean(self):
        """Overriding the default clean method to implement param checking"""
        super(AAAARecord, self).clean()
        try:
            ip_addr = self.rdata[0].decode('utf-8')
            ip.ip_address(ip_addr)
        except ValueError:
            raise me.ValidationError('IPv6 address provided is not valid')
        if not len(self.rdata) == 1:
            raise me.ValidationError('We cannot have more than one rdata'
                                     'values for this type of record.')

def clean(self):
        """Overriding the default clean method to implement param checking"""
        super(CNAMERecord, self).clean()
        if not self.rdata[0].endswith('.'):
            self.rdata[0] += '.'
        if not len(self.rdata) == 1:
            raise me.ValidationError('We cannot have more than one rdata'
                                     'values for this type of record.')

def clean(self):
        if not self.tags:
            self.tags = {}
        elif not isinstance(self.tags, dict):
            raise me.ValidationError('Tags must be a dictionary')

def exception_handler_mist(exc, request):
    """
    Here we catch exceptions and transform them to proper http responses
    This is a special pyramid view that gets triggered whenever an exception
    is raised from any other view. It catches all exceptions exc where
    isinstance(exc, context) is True.

    """
    # mongoengine ValidationError
    if isinstance(exc, me.ValidationError):
        trace = traceback.format_exc()
        log.warning("Uncaught me.ValidationError!\n%s", trace)
        return Response("Validation Error", 400)

    # mongoengine NotUniqueError
    if isinstance(exc, me.NotUniqueError):
        trace = traceback.format_exc()
        log.warning("Uncaught me.NotUniqueError!\n%s", trace)
        return Response("NotUniqueError", 409)

    # non-mist exceptions. that shouldn't happen! never!
    if not isinstance(exc, MistError):
        if not isinstance(exc, (me.ValidationError, me.NotUniqueError)):
            trace = traceback.format_exc()
            log.critical("Uncaught non-mist exception? WTF!\n%s", trace)
            return Response("Internal Server Error", 500)

    # mist exceptions are ok.
    log.info("MistError: %r", exc)

    # if it is a RedirectError, then send an HTTP Redirect
    if isinstance(exc, RedirectError):
        return HTTPFound(exc.url or '')

    # else translate it to HTTP response based on http_code attribute
    return Response(str(exc), exc.http_code)

def create_record(self, record, **kwargs):
        """
        This is the public method that is called to create a new DNS record
        under a specific zone.
        """
        record.name = kwargs['name']
        record.type = kwargs['type']
        if isinstance(kwargs['data'], list):
            record.rdata = kwargs['data']
        else:
            record.rdata = [kwargs['data']]
        record.ttl = kwargs['ttl']

        try:
            record.clean()
        except me.ValidationError as exc:
            log.error("Error validating %s: %s", record, exc.to_dict())
            raise BadRequestError({'msg': exc.message,
                                   'errors': exc.to_dict()})

        self._create_record__prepare_args(record.zone, kwargs)
        pr_record = self._create_record__for_zone(record.zone, **kwargs)
        record.record_id = pr_record.id
        # This is not something that should be given by the user, e.g. we
        # are only using this to store the ttl, so we should onl save this
        # value if it's returned by the provider.
        record.extra = pr_record.extra
        try:
            record.save()
        except me.ValidationError as exc:
            log.error("Error validating %s: %s", record, exc.to_dict())
            raise BadRequestError({'msg': exc.message,
                                   'errors': exc.to_dict()})
        except me.NotUniqueError as exc:
            log.error("Record %s not unique error: %s", record, exc)
            raise RecordExistsError()
        self.cloud.owner.mapper.update(record)

def add(self, fail_on_error=True, fail_on_invalid_params=True, **kwargs):
        """Add new Cloud to the database

        This is the only cloud controller subclass that overrides the `add`
        method of `BaseMainController`.

        This is only expected to be called by `Cloud.add` classmethod to create
        a cloud. Fields `owner` and `title` are already populated in
        `self.cloud`. The `self.cloud` model is not yet saved.

        If appropriate kwargs are passed, this can currently also act as a
        shortcut to also add the first machine on this dummy cloud.

        """
        # Attempt to save.
        try:
            self.cloud.save()
        except me.ValidationError as exc:
            raise BadRequestError({'msg': exc.message,
                                   'errors': exc.to_dict()})
        except me.NotUniqueError:
            raise CloudExistsError("Cloud with name %s already exists"
                                   % self.cloud.title)

        # Add machine.
        if kwargs:
            try:
                self.add_machine_wrapper(
                    self.cloud.title, fail_on_error=fail_on_error,
                    fail_on_invalid_params=fail_on_invalid_params, **kwargs
                )
            except Exception as exc:
                if fail_on_error:
                    self.cloud.delete()
                raise

def clean(self):
        """Checks the CIDR to determine if it maps to a valid IPv4 network."""

        try:
            self.cidr = str(netaddr.IPNetwork(self.cidr))
        except (TypeError, netaddr.AddrFormatError) as err:
            raise me.ValidationError(err)

def clean(self):
        # make sure user.email is unique - we can't use the unique keyword on
        # the field definition because both User and Organization subclass
        # Owner but only user has an email field
        if User.objects(email=self.email, id__ne=self.id):
            raise me.ValidationError("User with email '%s' already exists."
                                     % self.email)

        super(User, self).clean()

def update(self, fail_on_error=True, **kwargs):
        for key, value in kwargs.iteritems():
            if key not in type(self)._fields:
                if not fail_on_error:
                    continue
                raise me.ValidationError('Field "%s" does not exist on %s',
                                         key, type(self))
            setattr(self, key, value)

def clean(self):
        if self.timedelta.total_seconds() < 60:
            raise me.ValidationError("%s's timedelta cannot be less than "
                                     "a minute", self.__class__.__name__)

def clean(self):
        # FIXME This is needed in order to ensure rule name convention remains
        # backwards compatible with the old monitoring stack. However, it will
        # have to change in the future due to uniqueness constrains.
        if not self.title:
            self.title = 'rule%d' % self.owner.rule_counter
        # FIXME Ensure a single query condition is specified for backwards
        # compatibility with the existing monitoring/alert stack.
        if not len(self.queries) is 1:
            raise me.ValidationError()

def update(self, fail_on_error=True, **kwargs):
        for key, value in kwargs.iteritems():
            if key not in self._fields:
                if not fail_on_error:
                    continue
                raise me.ValidationError('Field "%s" does not exist on %s',
                                         key, type(self))
            setattr(self, key, value)

def add(self, fail_on_error=True, **kwargs):
        """Add a new Rule.

        This method is meant to be invoked only by the `Rule.add` classmethod.

        """
        for field in ('queries', ):  # TODO 'window', 'interval', ):
            if field not in kwargs:
                raise RequiredParameterMissingError(field)
        try:
            self.update(fail_on_error=fail_on_error, **kwargs)
        except (me.ValidationError, BadRequestError) as err:
            log.error('Error adding %s: %s', self.rule.name, err)
            raise

def clean(self):
        """Extended validation for EC2 Networks to ensure CIDR assignment."""
        if not self.cidr:
            raise me.ValidationError('Missing IPv4 range in CIDR notation')
        super(AmazonNetwork, self).clean()

def clean(self):
        """Extended validation for EC2 Networks to ensure CIDR assignment."""
        if not self.cidr:
            raise me.ValidationError('Missing IPv4 range in CIDR notation')
        super(AzureArmNetwork, self).clean()

def clean(self):
        """Checks the CIDR to determine if it maps to a valid IPv4 network."""
        try:
            netaddr.cidr_to_glob(self.cidr)
        except (TypeError, netaddr.AddrFormatError) as err:
            raise me.ValidationError(err)

def clean(self):
        """Extended validation for GCE Subnets."""
        if not re.match('^(?:[a-z](?:[-a-z0-9]{0,61}[a-z0-9])?)$', self.name):
            raise me.ValidationError('A **lowercase** name must be specified')
        super(GoogleSubnet, self).clean()

def add(self, **kwargs):
        """Add an entry to the database

        This is only to be called by `Schedule.add` classmethod to create
        a schedule. Fields `owner` and `name` are already populated in
        `self.schedule`. The `self.schedule` is not yet saved.

        """

        # check if required variables exist.
        if not (kwargs.get('script_id', '') or kwargs.get('action', '')):
            raise BadRequestError("You must provide script_id "
                                  "or machine's action")

        if not kwargs.get('conditions'):
            raise BadRequestError("You must provide a list of conditions, "
                                  "at least machine ids or tags")

        if kwargs.get('schedule_type') not in ['crontab',
                                               'interval', 'one_off']:
            raise BadRequestError('schedule type must be one of these '
                                  '(crontab, interval, one_off)]')

        if kwargs.get('schedule_type') == 'one_off' and not kwargs.get(
                'schedule_entry', ''):
            raise BadRequestError('one_off schedule '
                                  'requires date given in schedule_entry')

        try:
            self.update(**kwargs)
        except (me.ValidationError, me.NotUniqueError) as exc:
            # Propagate original error.
            log.error("Error adding %s: %s", self.schedule.name,
                      exc.to_dict())
            raise
        log.info("Added schedule with name '%s'", self.schedule.name)
        self.schedule.owner.mapper.update(self.schedule)

def load_user(user_id):
    if user_id not in ['None', None]:
        try:
            user = umodels.User.objects.filter(id=user_id).first()
            return user
        except (me.DoesNotExist, me.ValidationError):
            app.logger.error("User id \"{}\" does not exist".format(user_id))
    return False

def test_get_user(self):
        """Ensure get_user classmethod is returning always what we expect"""
        # test: invalid input
        self.assertRaises(ValidationError, umodels.User.get_user, None)
        # test: UserMixin (from flask.login) object
        self.assertEqual(self.user, umodels.User.get_user(self.user))
        # test: LocalProxy object
        login_user(self.user)
        self.assertEqual(self.user, umodels.User.get_user(current_user))
        # test: User Object from db but by "username" and by "id"
        self.assertEqual(self.user, umodels.User.get_user("username"))
        self.assertEqual(self.user, umodels.User.get_user(self.user.id))

def get_user(cls, user):
        """Get user object:
        1º check if user is already User, LocalProxy, or AnnonymousUser object
            if so, just return it
        2º if not, search for this user and return if found.
        3º otherwise, return DoesNotExists exception"""
        if any([isinstance(user, obj) for obj in [UserMixin,
                                                  AnonymousUserMixin]]):
            return user
        if isinstance(user, LocalProxy):
            return user._get_current_object()
        try:
            return User.objects.get(username=user)
        except (me.DoesNotExist, me.ValidationError):
            return User.objects.get(id=user)

def user_detail(user_id=None):
    """"""
    try:
        user = umodels.User.objects.get(id=user_id)
    except (me.DoesNotExist, me.ValidationError):
        abort(404)

    return render_template('admin/user/detail.html', **{'user': user})

def get_user(self, identifier):
        from mongoengine import ValidationError
        try:
            return self.user_model.objects(id=identifier).first()
        except ValidationError:
            pass
        for attr in get_identity_attributes():
            query_key = '%s__iexact' % attr
            query = {query_key: identifier}
            rv = self.user_model.objects(**query).first()
            if rv is not None:
                return rv

def find_user(self, **kwargs):
        try:
            from mongoengine.queryset import Q, QCombination
        except ImportError:
            from mongoengine.queryset.visitor import Q, QCombination
        from mongoengine.errors import ValidationError

        queries = map(lambda i: Q(**{i[0]: i[1]}), kwargs.items())
        query = QCombination(QCombination.AND, queries)
        try:
            return self.user_model.objects(query).first()
        except ValidationError:  # pragma: no cover
            return None

def create_organization(request):
    """
    Create organization.
    The user creating it will be assigned to the
    owners team. For now owner has only org
    ---
    name:
      description: The new org  name (id)
      type: string
      required: true
    """
    auth_context = auth_context_from_request(request)

    user = auth_context.user
    # SEC
    if not user.can_create_org:
        raise OrganizationAuthorizationFailure('Unauthorized to '
                                               'create organization')
    params = params_from_request(request)

    name = params.get('name')
    super_org = params.get('super_org')

    if not name:
        raise RequiredParameterMissingError()
    if Organization.objects(name=name):
        raise OrganizationNameExistsError()

    org = Organization()
    org.add_member_to_team('Owners', user)
    org.name = name

    # mechanism for sub-org creation
    # the owner of super-org has the ability to create a sub-org
    if super_org:
        org.parent = auth_context.org

    try:
        org.save()
    except me.ValidationError as e:
        raise BadRequestError({"msg": e.message, "errors": e.to_dict()})
    except me.OperationError:
        raise OrganizationOperationError()

    trigger_session_update(auth_context.user, ['user'])
    return org.as_dict()

def add_team(request):
    """
    Create new team.
    Append it at org's teams list.
    Only available to organization owners.
    ---
    name:
      description: The new team name
      type: string
      required: true
    description:
      description: The new team description
      type: string
    """

    log.info("Adding team")

    auth_context = auth_context_from_request(request)
    org_id = request.matchdict['org_id']

    params = params_from_request(request)
    name = params.get('name')
    description = params.get('description', '')
    visibility = params.get('visible', True)

    if not name:
        raise RequiredParameterMissingError()

    # SEC check if owner
    if not (auth_context.org and auth_context.is_owner()
            and auth_context.org.id == org_id):
        raise OrganizationAuthorizationFailure()

    team = Team()
    team.name = name
    team.description = description
    team.visible = visibility
    auth_context.org.teams.append(team)

    try:
        auth_context.org.save()
    except me.ValidationError as e:
        raise BadRequestError({"msg": e.message, "errors": e.to_dict()})
    except me.OperationError:
        raise TeamOperationError()

    log.info("Adding team with name '%s'.", name)
    trigger_session_update(auth_context.owner, ['org'])

    return team.as_dict()


# SEC

def delete_team(request):
    """
    Delete a team entry in the db.
    Only available to organization owners.
    ---
    org_id:
      description: The team's org id
      type: string
      required: true
    team_id:
      description: The team's id
      type: string
      required: true
    """
    auth_context = auth_context_from_request(request)
    org_id = request.matchdict['org_id']
    team_id = request.matchdict['team_id']

    # SEC check if owner
    if not (auth_context.org and auth_context.is_owner() and
            auth_context.org.id == org_id):
        raise OrganizationAuthorizationFailure()

    if auth_context.org.get_team('Owners').id == team_id:
        raise ForbiddenError()

    try:
        team = auth_context.org.get_team_by_id(team_id)
    except me.DoesNotExist:
        raise NotFoundError()

    if team.members:
        raise BadRequestError(
            'Team not empty. Remove all members and try again')

    try:
        team.drop_mappings()
        auth_context.org.update(pull__teams__id=team_id)
    except me.ValidationError as e:
        raise BadRequestError({"msg": e.message, "errors": e.to_dict()})
    except me.OperationError:
        raise TeamOperationError()

    trigger_session_update(auth_context.owner, ['org'])

    return OK


# SEC

def clean(self):
        # make sure that each team's name is unique
        used = set()
        for team in self.teams:
            if team.name in used:
                raise me.ValidationError("Team name exists.")
            used.add(team.name)

        # make sure that all team members are also org members
        for team in self.teams:
            for i, member in enumerate(list(team.members)):
                if member not in self.members:
                    team.members.pop(i)

        # make sure that owners team is present
        try:
            owners = self.teams.get(name='Owners')
        except me.DoesNotExist:
            raise me.ValidationError("Owners team can't be removed.")

        # make sure that owners team is not empty
        if not owners.members:
            raise me.ValidationError("Owners team can't be empty.")

        if HAS_POLICY:
            # make sure owners policy allows all permissions
            if owners.policy.operator != 'ALLOW':
                raise me.ValidationError("Owners policy must be set to ALLOW.")

            # make sure owners policy doesn't contain specific rules
            if owners.policy.rules:
                raise me.ValidationError("Can't set policy rules for Owners.")

        # make sure org name is unique - we can't use the unique keyword on the
        # field definition because both User and Organization subclass Owner
        # but only Organization has a name
        if self.name and Organization.objects(name=self.name, id__ne=self.id):
            raise me.ValidationError("Organization with name '%s' "
                                     "already exists." % self.name)

        self.members_count = len(self.members)
        self.teams_count = len(self.teams)
        super(Organization, self).clean()