Python falcon 模块，HTTP_BAD_REQUEST 实例源码

def on_post(self, req, resp, *args, **kwargs):
        data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
        data, errors = self.clean(data)
        if errors:
            result = {'errors': errors}
            status_code = falcon.HTTP_BAD_REQUEST
            self.render_response(result, req, resp, status_code)
            return

        try:
            with self.session_scope(self.db_engine) as db_session:
                result = self.create(req, resp, data, db_session=db_session)
        except IntegrityError:
            raise HTTPConflict('Conflict', 'Unique constraint violated')
        except ProgrammingError as err:
            # Cases such as unallowed NULL value should have been checked before we got here (e.g. validate against
            # schema using falconjsonio) - therefore assume this is a UNIQUE constraint violation
            if len(err.orig.args) > 1 and err.orig.args[1] == self.VIOLATION_UNIQUE:
                raise HTTPConflict('Conflict', 'Unique constraint violated')
            raise
        status_code = falcon.HTTP_CREATED

        self.render_response(result, req, resp, status_code)

def on_put(self, req, resp, *args, **kwargs):
        status_code = falcon.HTTP_OK
        try:
            with self.session_scope(self.db_engine) as db_session:
                obj = self.get_object(req, resp, kwargs, for_update=True, db_session=db_session)

                data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
                data, errors = self.clean(data)
                if errors:
                    result = {'errors': errors}
                    status_code = falcon.HTTP_BAD_REQUEST
                else:
                    result = self.update(req, resp, data, obj, db_session)
        except (IntegrityError, ProgrammingError) as err:
            # Cases such as unallowed NULL value should have been checked before we got here (e.g. validate against
            # schema using falconjsonio) - therefore assume this is a UNIQUE constraint violation
            if isinstance(err, IntegrityError) or err.orig.args[1] == self.VIOLATION_FOREIGN_KEY:
                raise HTTPConflict('Conflict', 'Unique constraint violated')
            else:
                raise

        self.render_response(result, req, resp, status_code)

def on_post(self, req, resp, *args, **kwargs):
        """
        Add (create) a new record to the collection.

        :param req: Falcon request
        :type req: falcon.request.Request

        :param resp: Falcon response
        :type resp: falcon.response.Response
        """
        data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
        data, errors = self.clean(data)
        if errors:
            result = {'errors': errors}
            status_code = falcon.HTTP_BAD_REQUEST
        else:
            result = self.create(req, resp, data)
            status_code = falcon.HTTP_CREATED

        self.render_response(result, req, resp, status_code)

def on_put(self, req, resp, *args, **kwargs):
        """
        Updates a single record.
        This should set all missing fields to default values, but we're not going to be so strict.

        :param req: Falcon request
        :type req: falcon.request.Request

        :param resp: Falcon response
        :type resp: falcon.response.Response
        """
        obj = self.get_object(req, resp, kwargs, for_update=True)

        data = self.deserialize(req.context['doc'] if 'doc' in req.context else None)
        data, errors = self.clean(data)
        if errors:
            result = {'errors': errors}
            status_code = falcon.HTTP_BAD_REQUEST
        else:
            result = self.update(req, resp, data, obj)
            status_code = falcon.HTTP_OK
        self.render_response(result, req, resp, status_code)

def on_get(self, req, resp, fid):
        '''Return the requested file as a stream.

        Args:
            fid (str): The file id. This is a URL safe base64 representation
                of the file path. The representation can be got by first querying
                the file tree of e.g the model run 'results' folder

        Example::

                http --download localhost:8000/files/{fid}
        '''
        try:
            resp.stream, resp.stream_len, resp.content_type = self._data_store.open(fid)
        except PermissionError as error:
            resp.status = falcon.HTTP_BAD_REQUEST
            resp.body = str(error)

def register(username, password, response=None):
    if not username or not username.strip():
        logger.debug("Empty username")
        response.status = falcon.HTTP_BAD_REQUEST
        return "Empty username"
    username = username.strip().lower()
    if len(username) > 64:
        logger.debug("Username too long")
        response.status = falcon.HTTP_UNPROCESSABLE_ENTITY
        return "Username too long"
    if _get_client(username):
        logger.debug("Username %s already in use", username)
        response.status = falcon.HTTP_CONFLICT
        return "Name already in use"

    if len(password.strip()) < 6:
        logger.debug("Password too short")
        response.status = falcon.HTTP_BAD_REQUEST
        return "Invalid password. Must be of length >= 6"
    pw_hash = bcrypt_sha256.encrypt(password)
    client = _SecretClient(username, pw_hash, ["user"])
    try:
        _add_client(client)
    except ValueError:
        logger.debug("Username %s already in database", username)
        response.status = falcon.HTTP_CONFLICT
        return "Name already in use"
    logger.debug("Registered new user: %s", username)
    return _create_user_token(username, client.permissions)

def change_password(old_password, new_password, user: hug.directives.user, response=None):
    """
    Change the password of the user if the old password is correct and the new one satisfies the password requirements.
    :param old_password: the old password
    :param new_password: the new password
    """
    client = _get_client(user['name'])
    success = bcrypt_sha256.verify(old_password, client.pw_hash)
    if not success:
        logger.debug("%s tried to change his password with a wrong old password", user['name'])
        response.status = falcon.HTTP_FORBIDDEN
        return "Wrong password"

    new_password = new_password.strip()
    if len(new_password) < 6:
        logger.debug("Password too short")
        response.status = falcon.HTTP_BAD_REQUEST
        return "Invalid password. Must be of length >= 6"

    pw_hash = bcrypt_sha256.encrypt(new_password)

    db = _get_db_conn()
    try:
        with db:
            db.execute("UPDATE clients SET pw_hash=? WHERE username=?", (pw_hash, user['name']))
        return "OK"
    finally:
        db.close()

def on_get(self, req, resp, fid):
        ''' Return a JSON representation of the directory tree. The JSON response has the
        following attributes:

            - ``type``: file or folder
            - ``name``: The base name of the file/folder
            - ``path``: Absolute path to the object.
            - ``id``: URL-safe base64 encoding of the ``path``
            - ``children``: Only present if ``type`` is ``folder``. A list of all
                children of this folder, each having the same representation.

        Args:
            fid (str): the base64 encoded url safe ID for the path to the root folder

        Example::

            http localhost:8000/files/tree/{fid}

        '''
        try:
            self._data_store.validate_fid(fid)
            resp.status = falcon.HTTP_OK
            path = serializers.path_from_id(fid)
            resp.body = json.dumps(serializers.directory_tree_serializer(path))
        except PermissionError as error:
            resp.status = falcon.HTTP_BAD_REQUEST
            resp.body = str(error)

def __init__(self, message, input_=None):
        FalconSwaggerError.__init__(self, message, HTTP_BAD_REQUEST)
        self.input_ = input_

def __init__(self, message, headers=None, input_=None):
        FalconSwaggerError.__init__(self, message, HTTP_BAD_REQUEST, headers=headers, input_=input_)

def on_post(self, req, resp):
        self.process_request(req, resp)

        password = req.context.get('_body', {}).get('password')
        if password is None:
            resp.status = falcon.HTTP_BAD_REQUEST
            return None

        result = self.parse_password(password)
        resp.body = json.dumps(result)

def on_post(self, req, resp):
        '''Create a new ``Model`` by uploading a zip file containing the input data
        required to run the model using a supported flood modelling software.

        The zip file will be saved and unpacked to the data folder configured in the
        ``data_store`` instance. This folder path will be stored in the ``Model`` instance.

        The name of the model is generated automatically and can be overwritten with a PUT
        request.

        A JSON representation of the created model is returned in the response body and
        the GET URL in the response location header.

        Example::

            http post localhost:8000/models @/path/to/my/archive.zip
        '''
        if req.content_type == 'application/json':
            # Create a model without data.
            data = json.load(req.bounded_stream)
            model = self._document(
                name=data['name']
            )
            if 'description' in data:
                model.description = data['description']
            if 'email' in data:
                model.email = data['email']
            model.save()

            resp.status = falcon.HTTP_CREATED
            resp.location = '/models/{}'.format(model.name)
            resp.body = model.to_json()

        elif req.content_type == 'application/zip':
            directory, name = self._data_store.save_zip(req.stream, req.content_type)
            # Create the model
            model = self._document(
                name=name,
                folder=directory
            ).save()

            resp.status = falcon.HTTP_CREATED
            resp.location = '/models/{}'.format(model.name)
            resp.body = model.to_json()
        else:
            resp.status = falcon.HTTP_BAD_REQUEST
            resp.body = 'Content type must be zip or json'

def on_patch(self, req, resp, name):
        '''Update a ``Model`` instance with new data.

        Data is submitted as a JSON object. Any of the following attributes are accepted:

        - ``description``: A new description of the model
        - ``name``: A new name for the model
        - ``email``: Email address of an existing user to transfer ownership
            of the model to.

        '''
        model = self.get_object(name)
        if req.content_type == 'application/json':
            data = json.load(req.bounded_stream)
            if 'description' in data:
                model.description = data['description']
            if 'name' in data:
                model.name = data['name']
            if 'email' in data:
                model.email = data['email']

            try:
                model.save()
                resp.status = falcon.HTTP_NO_CONTENT
            except fmdb.NotUniqueError:
                resp.status = falcon.HTTP_BAD_REQUEST
                resp.body = 'Name {} is already taken'.format(data['name'])

        else:
            if model.folder:
                folder = model.resolve_folder()
            else:
                folder = model.name

            if req.content_type == 'application/zip':
                # Extract any uploaded zip file into the model data folder
                root, name = self._data_store.save_zip(req.stream, req.content_type, folder)
            else:
                # Save a single uploaded file to the model folder.
                # If the filename is given, use that, otherwise
                # try to guess the file type and generate a name
                disp = req.get_header('content-disposition')
                result = disp.split('filename=')
                if len(result) == 2:
                    filename = result[1].replace('"', '').replace("'", '')
                else:
                    ext = mimetypes.guess_extension(req.content_type)
                    filename = '{}{}'.format(uuid.uuid4(), ext)
                root, fid = self._data_store.save(req.stream, folder, filename)
                resp.body = json.dumps({
                    'fid': fid
                })


            model.folder = root
            model.save()
            resp.status = falcon.HTTP_ACCEPTED