Python flask.request 模块，is_xhr() 实例源码

def generate_diff(test_id, regression_test_id, output_id):
    from run import config
    if request.is_xhr:
        # Fetch test
        result = TestResultFile.query.filter(and_(
            TestResultFile.test_id == test_id,
            TestResultFile.regression_test_id == regression_test_id,
            TestResultFile.regression_test_output_id == output_id)).first()
        if result is not None:
            path = os.path.join(
                config.get('SAMPLE_REPOSITORY', ''), 'TestResults')
            return result.generate_html_diff(path)
        abort(404)
    abort(403, 'generate_diff')

def update_user():
    if request.is_xhr:
        user_id = request.args.get('id')
        update_profile(user_id)
        return "OK"
    else:
        abort(404)

def jsonify_resource(resource):
    indent = None
    if current_app.config['JSONIFY_PRETTYPRINT_REGULAR'] \
            and not request.is_xhr:
        indent = 2
    return current_app.response_class(json.dumps(resource, indent=indent),
                                      mimetype='application/json')

def test_instance_attachment_collection_service(test_id):
    resp = flask.Response(json.dumps({'status': 'failed'}))
    if request.method == 'POST':
        if request.is_xhr:
            file = request.files['attachments']
            if file and allowed_file(file.filename):
                filename = secure_filename(file.filename)
                dirpath = os.path.join(UPLOAD_FOLDER, str(test_id), 'attachments')
                os.makedirs(dirpath, exist_ok=True)
                filepath = os.path.join(dirpath, filename)
                file.save(filepath)

                # db: update the attachment for the test where test.id = test_id
                test = Test.query.get(test_id)
                test.test_attachment.append(TestAttachment(name=filename, attachment_url=filepath))
                db.session.commit()

                resp = flask.Response(json.dumps({'status': 'success', 'url': filepath}))
    elif request.method == 'DELETE':
        if request.is_json:
            filename = secure_filename(request.json['removedFile'])
            dirpath = os.path.join(UPLOAD_FOLDER, str(test_id), 'attachments')
            filepath = os.path.join(dirpath, filename)
            try:
                os.remove(filepath)
                # db: delete the attachment for the test where test.id = test_id
                TestAttachment.query.filter(
                    (TestAttachment.test_id == test_id) &
                    (TestAttachment.name == filename)
                ).delete()
                db.session.commit()

                resp = flask.Response(json.dumps({'status': 'success', 'url': filepath}))
            except FileNotFoundError:
                print('FileNotFound: ', filepath)
                resp = flask.Response(json.dumps({'status': 'failed', 'url': filepath}))

    return set_debug_response_header(resp)

def view_embed(organization_id: str):
    if not request.is_xhr:
        return index(organization_id)

    api = pillar_api()

    organization: Organization = Organization.find(organization_id, api=api)

    om = current_app.org_manager
    organization_oid = str2id(organization_id)

    members = om.org_members(organization.members)
    for member in members:
        member['avatar'] = gravatar(member.get('email'))
        member['_id'] = str(member['_id'])

    admin_user = User.find(organization.admin_uid, api=api)

    # Make sure it's never None
    organization.unknown_members = organization.unknown_members or []

    can_super_edit = current_user.has_cap('admin')
    can_edit = can_super_edit or om.user_is_admin(organization_oid)

    csrf = flask_wtf.csrf.generate_csrf()

    return render_template('organizations/view_embed.html',
                           organization=organization,
                           admin_user=admin_user,
                           members=members,
                           can_edit=can_edit,
                           can_super_edit=can_super_edit,
                           seats_used=len(members) + len(organization.unknown_members),
                           csrf=csrf)

def pillar_error_handler(self, error_ob):

        # 'error_ob' can be any exception. If it's not a Werkzeug exception,
        # handle it as a 500.
        if not hasattr(error_ob, 'code'):
            error_ob.code = 500
        if not hasattr(error_ob, 'description'):
            error_ob.description = str(error_ob)

        if request.full_path.startswith('/%s/' % self.config['URL_PREFIX']):
            from pillar.api.utils import jsonify
            # This is an API request, so respond in JSON.
            return jsonify({
                '_status': 'ERR',
                '_code': error_ob.code,
                '_message': error_ob.description,
            }, status=error_ob.code)

        # See whether we should return an embedded page or a regular one.
        if request.is_xhr:
            fname = 'errors/%i_embed.html' % error_ob.code
        else:
            fname = 'errors/%i.html' % error_ob.code

        # Also handle the case where we didn't create a template for this error.
        try:
            return render_template(fname, description=error_ob.description), error_ob.code
        except TemplateNotFound:
            self.log.warning('Error template %s for code %i not found',
                             fname, error_ob.code)
            return render_template('errors/500.html'), error_ob.code

def to_json(content):
    """Converts content to json while respecting config options."""
    indent = None
    separators = (',', ':')
    try:
        if current_app.config['JSONIFY_PRETTYPRINT_REGULAR'] and not request.is_xhr:
            indent = 2
            separators = (', ', ': ')
    except:
        pass
    return json.dumps(content, indent=indent, separators=separators, cls=JsonEncoder)

def _check_post_settings(board: BoardModel, post_details):
    site_config = site_service.get_site_config()
    if not site_config.posting_enabled:
        raise BadRequestError(MESSAGE_POSTING_DISABLED)

    if post_details.has_file and not site_config.file_posting:
        raise BadRequestError(MESSAGE_FILE_POSTING_DISABLED)

    if board.config.posting_verification_required and not verification_service.is_verified(request):
        method = verification_service.get_method()
        if method.verification_in_request(request):
            try:
                method.verify_request(request)
                verification_service.set_verified(request)
            except ArgumentError as e:
                raise BadRequestError(e)
        else:
            message = 'Please verify here first before posting.'

            if request.is_xhr:
                xhr_response = {
                    'error': True,
                    'message': page_formatting('[{}](_/verify/)'.format(message))
                }

                return jsonify(xhr_response), 400
            else:
                with_refresh = '[{}](_/verify/)\n\n**Refresh this page after verifying.**'.format(message)

                return render_template('error.html', message=with_refresh, with_retry=True), 400

def _create_post_response(post_result):
    if request.is_xhr:
        return jsonify({
            'boardName': post_result.board_name,
            'threadRefno': post_result.thread_refno,
            'postRefno': post_result.post_refno
        })
    else:
        return redirect(url_for_post(post_result.board_name, post_result.thread_refno, post_result.post_refno))

def render_error(user_message, code=400, with_retry=False):
    if request.is_xhr:
        xhr_response = {
            'error': True
        }

        if user_message:
            xhr_response['message'] = page_formatting(user_message)

        return jsonify(xhr_response), code
    else:
        return render_template('error.html', message=user_message, with_retry=with_retry), code