Python flask.json 模块，jsonify() 实例源码

def get_data():
    cname = request.args.get("collection")
    settings = CSETTINGS[cname]
    search_string = request.args.get("search_string")
    xaxis = request.args.get("xaxis")
    yaxis = request.args.get("yaxis")

    xaxis = get_mapped_name(settings, xaxis)
    yaxis = get_mapped_name(settings, yaxis)

    projection = [xaxis, yaxis]

    if search_string.strip() != "":
        criteria = process_search_string(search_string, settings)
        data = []
        for r in DB[cname].find(criteria, projection=projection):
            x = _get_val(xaxis, r, None)
            y = _get_val(yaxis, r, None)
            if x and y:
                data.append([x, y])
    else:
        data = []
    return jsonify(jsanitize(data))

def _serve_webui(self, file_name='index.html'):  # pylint: disable=redefined-builtin
        try:
            assert file_name
            web3 = self.flask_app.config.get('WEB3_ENDPOINT')
            if web3 and 'config.' in file_name and file_name.endswith('.json'):
                host = request.headers.get('Host')
                if any(h in web3 for h in ('localhost', '127.0.0.1')) and host:
                    _, _port = split_endpoint(web3)
                    _host, _ = split_endpoint(host)
                    web3 = 'http://{}:{}'.format(_host, _port)
                response = jsonify({'raiden': self._api_prefix, 'web3': web3})
            else:
                response = send_from_directory(self.flask_app.config['WEBUI_PATH'], file_name)
        except (NotFound, AssertionError):
            response = send_from_directory(self.flask_app.config['WEBUI_PATH'], 'index.html')
        return response

def delete_executor(executor_id):
    """
    Delete an executor

    Example request::

        DEL /api/1/executors/6890192d8b6c40e5af16f13aa036c7dc
    """
    manager.remove_executor(executor_id)
    return json.jsonify({
        "msg": "Executor %s succesfully deleted" % executor_id
    })

########
# UTILS
#######

def join(id):
    event = find_event(id)
    if event == None:
        return not_found()

    user = find_user(request.values.get('user_param'), request.values.get('user_param_value'))
    if user == None:
        return user_not_specified()

    join_event(user, event)

    return jsonify({
        'joined': True,
        'rsvps_disabled': False,
        'event_archived': False,
        'over_capacity': False,
        'past_deadline': False,
    })

def handle_authn_response():
    # parse authn response
    authn_response = current_app.rp.client.parse_response(AuthorizationResponse,
                                              info=request.query_string.decode("utf-8"),
                                              sformat="urlencoded")

    auth_code = None
    if "code" in authn_response:
        auth_code = authn_response["code"]
        # make token request
        args = {
            "code": auth_code,
            "client_id": current_app.rp.client.client_id,
            "client_secret": current_app.rp.client.client_secret
        }

        token_response = current_app.rp.client.do_access_token_request(scope="openid", request_args=args)
        access_token = token_response["access_token"]
        id_token = token_response["id_token"].to_dict()
        # TODO do userinfo req
    else:
        id_token = authn_response["id_token"].to_dict()
        access_token = authn_response.get("access_token")

    return jsonify(dict(auth_code=auth_code, token=access_token, id_token=id_token))

def api_private_counties_by_month():
    NOW = datetime.now()
    result = []
    r = current_app.db_session.query(
            Report.test_start_time,
            Report.probe_cc) \
        .order_by(Report.test_start_time)

    # XXX this can be done better in a SQL that is not sqlite
    monthly_buckets = {}
    for tst, country in r:
        if tst > NOW:
            # We ignore measurements from time travelers
            continue
        bkt = tst.strftime("%Y-%m-01")
        monthly_buckets[bkt] = monthly_buckets.get(bkt, [])
        if country not in monthly_buckets[bkt]:
            monthly_buckets[bkt].append(country)

    for bkt in monthly_buckets.keys():
        result.append({
            'date': bkt,
            'value': len(monthly_buckets[bkt])
        })
    return jsonify(result)

def api_private_runs_by_month():
    NOW = datetime.now()
    result = []
    r = current_app.db_session.query(
            Report.test_start_time) \
        .order_by(Report.test_start_time)

    # XXX this can be done better in a SQL that is not sqlite
    monthly_buckets = {}
    for res in r:
        tst = res.test_start_time
        if tst > NOW:
            # We ignore measurements from time travelers
            continue
        bkt = tst.strftime("%Y-%m-01")
        monthly_buckets[bkt] = monthly_buckets.get(bkt, 0)
        monthly_buckets[bkt] += 1

    for bkt in monthly_buckets.keys():
        result.append({
            'date': bkt,
            'value': monthly_buckets[bkt]
        })

    return jsonify(result)

def api_private_blockpage_detected():
    q = current_app.db_session.query(
        distinct(Report.probe_cc).label('probe_cc'),
    ).join(Measurement, Measurement.report_no == Report.report_no) \
     .filter(Measurement.confirmed == True) \
     .filter(or_(
        Report.test_name == 'http_requests',
        Report.test_name == 'web_connectivity'
      ))

    results = []
    for row in q:
        results.append({
            'probe_cc': row.probe_cc
        })

    return jsonify({
        'results': results
    })

def variantset_search():
    if not request.json:
        return makeGAException(
            message='Bad Content Type, please send application/json',
            ecode=-1,
            rcode=415)
    datasetId = request.json.get('datasetId', "")
    pageSize = request.json.get('pageSize', None)
    pageToken = request.json.get('pageToken', None)

    try:
        garesp = tileSearch.searchVariantSets(
            datasetId=datasetId, pageSize=pageSize, pageToken=pageToken)

    except:
        return makeGAException(
            message='search command exited abnormally', ecode=500, rcode=500)

    return jsonify(garesp.gavsetresponse_info)

def put(self, username, transaction_id):
        user = self.authenticate()
        if user:
            new_transaction = request.get_json(force=True)
            transaction = user.transactions.filter_by(
                transaction_id=transaction_id).first_or_404()
            transaction.category_id = new_transaction['category_id']
            transaction.person_id = new_transaction['person_id']
            transaction.transaction_date = new_transaction['transaction_date']
            transaction.value = new_transaction['value']
            transaction.notes = new_transaction['notes']
            transaction.type = new_transaction['type']
            transaction.done = new_transaction['done']
            db.session.commit()
            return json.jsonify(transaction.as_dict())
        raise InvalidUsage()

def post(self, username, transaction_id):
        user = self.authenticate()
        if user:
            transaction = user.transactions.filter_by(
                transaction_id=transaction_id).first_or_404()
            supposed_item = request.get_json(force=True)
            item = TransactionItem()
            item.item_id = 1 if len(transaction.transaction_items.all(
            )) else transaction.transaction_items.all()[-1].item_id + 1
            item.user_id = user.user_id
            item.transaction_id = transaction.transaction_id
            item.person_id = supposed_item['person_id']
            item.item_date = supposed_item['item_date']
            item.value = supposed_item['value']
            item.notes = supposed_item['notes']
            item.type = supposed_item['type']
            item.done = supposed_item['done']
            db.session.add(item)
            db.session.commit()
            if item.item_id:
                return json.jsonify(item.as_dict())
        raise InvalidUsage()

def put(self, username, transaction_id, item_id):
        user = self.authenticate()
        if user:
            transaction = user.transactions.filter_by(
                transaction_id=transaction_id).first_or_404()
            new_item = request.get_json(force=True)
            item = transaction.transaction_items.filter_by(
                item_id=item_id).first_or_404()
            item.person_id = new_item['person_id']
            item.item_date = new_item['item_date']
            item.value = new_item['value']
            item.notes = new_item['notes']
            item.type = new_item['type']
            item.done = new_item['done']
            db.session.commit()
            return json.jsonify(item.as_dict())
        raise InvalidUsage()

def stack_analyses_debug(external_request_id):
    """Debug endpoint exposing operational data for particular stack analysis.

    This endpoint is not part of the public API.

    Note the existence of the data is not guaranteed,
    therefore the endpoint can return 404 even for valid request IDs.
    """

    results = retrieve_worker_results(rdb, external_request_id)
    if not results:
        return jsonify(error='No operational data for the request ID'), 404

    response = {'tasks': []}
    for result in results:
        op_data = result.to_dict()
        audit = op_data.get('task_result', {}).get('_audit', {})
        task_data = {'task_name': op_data.get('worker')}
        task_data['started_at'] = audit.get('started_at')
        task_data['ended_at'] = audit.get('ended_at')
        task_data['error'] = op_data.get('error')
        response['tasks'].append(task_data)
    return jsonify(response), 200

def hipfrog():
    requestdata = json.loads(request.get_data())
    callingMessage = requestdata['item']['message']['message'].lower().split()
    oauthId = requestdata['oauth_client_id']
    installation = messageFunctions.getInstallationFromOauthId(oauthId)

    if installation.glassfrogToken is None:
        message = strings.set_token_first
        message_dict = messageFunctions.createMessageDict(strings.error_color, message)
    elif len(callingMessage) == 1:
        message = strings.help_hipfrog
        message_dict = messageFunctions.createMessageDict(strings.succes_color, message)
    elif len(callingMessage) > 1:
        # /hipfrog something
        message = strings.missing_functionality.format(callingMessage[1])
        message_dict = messageFunctions.createMessageDict(strings.error_color, message)
    # TODO Generate message_dict and color here
    return json.jsonify(message_dict)

def error_handler(error):
    """
    Standard Error Handler
    """
    if isinstance(error, HTTPException):
        return jsonify({
            'statusCode': error.code,
            'name': error.name,
            'description': error.description
        }), error.code
    else:
        return jsonify({
            'statusCode': 500,
            'name': 'Internal Server Error',
            'description': 'An unknown error has occurred'
        }), 500

# common errors - add others as needed

def get_group_sections():
    q = request.query_string.decode()
    if not q:
        return json.jsonify({'status': 'error'})

    all_cn = []
    for cn in rd.smembers('group:%s' % q):
        try:
            all_cn.insert(0, int(cn))
        except ValueError:
            continue

    schedule = tuple(
        lesson.details
        for lesson in Lesson.query.filter(Lesson.class_no.in_(all_cn))
        .order_by(Lesson.start).all()
    )

    return json.jsonify({
        'status': 'ok',
        'events': schedule,
    })

def get_ids(collection_name):
    settings = CSETTINGS[collection_name]
    doc = DB[collection_name].distinct(settings["unique_key"])
    return jsonify(jsanitize(doc))

def get_doc_json(collection_name, uid):
    settings = CSETTINGS[collection_name]
    criteria = {
        settings["unique_key"]: process(uid, settings["unique_key_type"])}
    doc = DB[collection_name].find_one(criteria)
    return jsonify(jsanitize(doc))

def task_in_json(taskid):
    if ':' not in taskid:
        return json.jsonify({'code': 400, 'error': 'bad project:task_id format'})
    project, taskid = taskid.split(':', 1)

    taskdb = app.config['taskdb']
    task = taskdb.get_task(project, taskid)

    if not task:
        return json.jsonify({'code': 404, 'error': 'not found'})
    task['status_string'] = app.config['taskdb'].status_to_string(task['status'])
    return json.jsonify(task)

def handle_invalid_usage(error):
    """
    Handler for APIErrors thrown by API endpoints
    """
    log.info('%d %s', error.status_code, error.message)
    response = json.jsonify(error.to_dict())
    response.status_code = error.status_code
    return response

def add_plan():
    """
    Add a plan.

    Example request::

        PUT /api/1/executors/3b373155577b4d1bbc62216ffea013a4
        Body:
            {
                "name": "Terminate instances in Playground",
                "attack": {
                    "args": {
                        "region": "eu-west-1",
                        "filters": {
                            "tag:Name": "playground-asg"
                        }
                    },
                    "ref": "terminate_ec2_instance:TerminateEC2Instance"
                },
                "planner": {
                    "ref": "simple_planner:SimplePlanner",
                    "args": {
                        "min_time" : "10:00",
                        "max_time" : "19:00",
                        "times": 4
                    }
                }
            }

    """
    assert validate_payload(request, plan_schema)
    req_json = request.get_json()

    name = req_json["name"]
    planner_config = req_json["planner"]
    attack_config = req_json["attack"]

    manager.execute_plan(name, planner_config, attack_config)

    return json.jsonify({"msg": "ok"})

def delete_plan(plan_id):
    """
    Delete a plan

    Example request::

        DEL /api/1/plans/6890192d8b6c40e5af16f13aa036c7dc


    """
    manager.delete_plan(plan_id)
    return json.jsonify({
        "msg": "Plan %s successfully deleted" % plan_id
    })

def render_event(event):
    return jsonify(filter_participants(event))

def render_events(events):
    return jsonify([filter_participants(event) for event in events])

def not_found():
    response = jsonify({"message":"not_found"})
    response.status_code = 404
    return response

def user_not_specified():
    response = jsonify({"message": "user_not_specified"})
    response.status_code = 422
    return response

def custom_401(error):
    # API CALL
    if 'Content-Type' in request.headers and request.headers['Content-Type'].lower() == 'application/json':
        return jsonify(errors=['Not authorized']), 401
    return render_template('home/401.html', message=error.description['message']), 401

def get_version():
    return jsonify({
        "version": __version__
    })

def api_private_asn_by_month():
    NOW = datetime.now()
    result = []
    r = current_app.db_session.query(
            Report.test_start_time,
            Report.probe_asn) \
        .order_by(Report.test_start_time)

    # XXX this can be done better in a SQL that is not sqlite
    monthly_buckets = {}
    for tst, asn in r:
        asn = 'AS%d' % asn
        if tst > NOW:
            # We ignore measurements from time travelers
            continue
        bkt = tst.strftime("%Y-%m-01")
        monthly_buckets[bkt] = monthly_buckets.get(bkt, [])
        if asn not in monthly_buckets[bkt]:
            monthly_buckets[bkt].append(asn)

    for bkt in monthly_buckets.keys():
        result.append({
            'date': bkt,
            'value': len(monthly_buckets[bkt])
        })
    return jsonify(result)

def api_private_reports_per_day():
    q = current_app.db_session.query(
        func.count(func.date_trunc('day', Report.test_start_time)),
        func.date_trunc('day', Report.test_start_time)
    ).group_by(func.date_trunc('day', Report.test_start_time)).order_by(
        func.date_trunc('day', Report.test_start_time)
    )
    result = []
    for count, date in q:
        result.append({
            'count': count,
            'date': date.strftime("%Y-%m-%d")
        })
    return jsonify(result)

def api_private_countries():
    with_counts = request.args.get("with_counts")
    country_list = []
    if not with_counts:
        country_list = [{ 'alpha_2': c.alpha_2, 'name': c.name } for c in countries]
    else:
        # XXX we probably actually want to get this from redis or compute it
        # periodically since it take 60 seconds to run.
        q = current_app.db_session.query(
                func.count(Measurement.id),
                Report.probe_cc.label('probe_cc')
        ).join(Report, Report.report_no == Measurement.report_no) \
         .group_by(Report.probe_cc)
        for count, probe_cc in q:
            if probe_cc.upper() == 'ZZ':
                continue

            try:
                c = countries.lookup(probe_cc)
                country_list.append({
                    'count': count,
                    'alpha_2': c.alpha_2,
                    'name': c.name,
                })
            except Exception as exc:
                current_app.logger.warning("Failed to lookup: %s" % probe_cc)

    return jsonify({
        "countries": country_list
    })

# XXX Everything below here are ghetto hax to support legacy OONI Explorer

def api_private_blockpages():
    probe_cc = request.args.get('probe_cc')
    if probe_cc is None:
        raise Exception('err')

    q = current_app.db_session.query(
        Report.report_id.label('report_id'),
        Report.probe_cc.label('probe_cc'),
        Report.probe_asn.label('probe_asn'),
        Report.test_start_time.label('test_start_time'),
        Input.input.label('input')
    ).join(Measurement, Measurement.report_no == Report.report_no) \
     .join(Input, Measurement.input_no == Input.input_no) \
     .filter(Measurement.confirmed == True) \
     .filter(or_(
        Report.test_name == 'http_requests',
        Report.test_name == 'web_connectivity'
      )) \
     .filter(Report.probe_cc == probe_cc)

    results = []
    for row in q:
        results.append({
            'report_id': row.report_id,
            'probe_cc': row.probe_cc,
            'probe_asn': "AS{}".format(row.probe_asn),
            'test_start_time': row.test_start_time,
            'input': row.input
        })

    return jsonify({
        'results': results
    })

def api_private_website_measurements():
    input_ = request.args.get('input')
    if input_ is None:
        raise Exception('err')

    q = current_app.db_session.query(
        Report.report_id.label('report_id'),
        Report.probe_cc.label('probe_cc'),
        Report.probe_asn.label('probe_asn'),
        Report.test_start_time.label('test_start_time'),
        Input.input.label('input'),
    ).join(Measurement, Measurement.report_no == Report.report_no) \
     .join(Input, Measurement.input_no == Input.input_no) \
     .filter(Measurement.confirmed == True) \
     .filter(or_(
        Report.test_name == 'http_requests',
        Report.test_name == 'web_connectivity'
      )) \
     .filter(Input.input.contains(input_))

    results = []
    for row in q:
        results.append({
            'report_id': row.report_id,
            'probe_cc': row.probe_cc,
            'probe_asn': "AS{}".format(row.probe_asn),
            'test_start_time': row.test_start_time,
            'input': row.input
        })

    return jsonify({
        'results': results
    })

def makeGAException(message, ecode, rcode):
    error = {
        'message': message,
        'error_code': ecode
    }
    GAException = json.jsonify(error)
    GAException.status_code = rcode
    return GAException

def bad_method(e):
    return jsonify(error=405, text=str(e)), 405

def get(last_name, first_name):
        """ Return an user key information based on his name """
        user = UserRepository.get(last_name=last_name, first_name=first_name)
        return jsonify({'user': user.json})

def post(last_name, first_name, age):
        """ Create an user based on the sent information """
        user = UserRepository.create(
            last_name=last_name,
            first_name=first_name,
            age=age
        )
        return jsonify({'user': user.json})

def put(last_name, first_name, age):
        """ Update an user based on the sent information """
        repository = UserRepository()
        user = repository.update(
            last_name=last_name,
            first_name=first_name,
            age=age
        )
        return jsonify({'user': user.json})

def get(self, lid):
        league = League.query.filter_by(id=lid).first()
        return self.league_schema.jsonify(league)

def get(self, mid):
        membership = Membership.query.filter_by(id=mid).first()
        return self.membership_schema.jsonify(membership)

def get(self, uid):
        user = User.query.filter_by(id=uid).first()
        return self.user_schema.jsonify(user)

def get(self):
        users = User.query.all()
        users_raw = {'users': []}
        for user in users:
            users_raw['users'].append(self.user_schema.dump(user).data)
        return json.jsonify(users_raw)

def handle_invalid_usage(error):
    return json.jsonify({'error': str(error)}), error.status_code

def handle_not_allowed(error):
    return json.jsonify({'error': str(error)}), 405

def handle_not_found(error):
    return json.jsonify({'error': str(error)}), 404

def post(self):
        supposed_user = request.get_json(force=True)
        user = User.query.filter_by(username=supposed_user[
                                    'username']).first_or_404()
        if user and user.password == Crypt.hash_sha256(supposed_user['password']):
            token = jwt.encode({
                'exp': time.time() + 60 * 60 * 24 * 7,
                'user_id': user.user_id,
                'username': user.username,
                'name': user.name
            }, config.SECRET_KEY, algorithm='HS256')
            return json.jsonify({'token': token})
        raise InvalidUsage("Username and password does not match.")

def get(self, username):
        user = self.authenticate()
        if user:
            return json.jsonify(user.as_dict())
        raise InvalidUsage()

def put(self, username):
        user = self.authenticate()
        if user:
            new_user = request.get_json(force=True)
            user.name = new_user['name']
            user.username = new_user['username']
            user.password = Crypt.hash_sha256(new_user['password'])
            db.session.commit()
            return json.jsonify(user.as_dict())
        raise InvalidUsage()

def delete(self, username):
        user = self.authenticate()
        if user:
            db.session.delete(user)
            db.session.commit()
            return json.jsonify({'success': '{} was deleted'.format(user)})
        raise InvalidUsage()

def get(self, username, category_id):
        user = self.authenticate()
        if user:
            if category_id:
                return json.jsonify(
                    user.categories.filter_by(category_id=category_id)
                    .first_or_404()
                    .as_dict())
            return json.jsonify({'categories': [category.as_dict() for category
                                                in user.categories]})
        raise InvalidUsage()