python类json()的实例源码

def info_view(request, name, value):
    """Retrieve JSON or CSV encoded items give the name/value filter."""
    limit = int(request.args.get('limit', 3))
    offset = int(request.args.get('offset', 0))
    format = request.args.get('format', 'json')
    logger.info('?? Requesting {0} (offset={1}) items in {2}'.format(
                limit, offset, format))
    columns = request.args.get('columns')
    column_names = columns and columns.split(',') or app.config.columns
    sirets = retrieve_sirets(name, value, offset=offset, limit=limit)
    if not sirets:
        raise ServerError('?? No entry found for {0}/{1}'.format(name, value))
    logger.info('?? Retrieving {0} results for the {1}/{2} couple'.format(
                len(sirets), name, value))
    logger.info('?? Returning {0} results in {1}'.format(len(sirets), format))
    rows = [decode_siret(retrieve_siret(siret), column_names)
            for siret in sirets]
    return format_response(rows, format, column_names)

def response_json(data=None, status_code=None, message=None, page=None, total_pages=None):
        response_data = {
            'data': data,
            'status_code': status_code,
            'message': message
        }

        if page:
            response_data['page'] = page
        if total_pages:
            response_data['total_pages'] = total_pages

        return json(response_data, status=status_code)

def _validate_json(self, request):
        try:
            valid = request.json
            return True
        except Exception:
            return self.response_json(status_code=400,
                                      message=self.config.response_messages.ErrorInvalidJSON)

def _validate_primary_key_immutable(self, request):
        if self.model.shortcuts.primary_key in request.json:
            return self.response_json(status_code=400,
                                      message=self.config.response_messages.ErrorPrimaryKeyUpdateInsert)
        else:
            return True

def _validate_fields(self, request):
        fields = self.model.shortcuts.editable_fields
        request_data = request.json

        for key in request_data:
            if key not in fields:
                return self.response_json(status_code=400,
                                          message=self.config.response_messages.ErrorInvalidField.format(key, fields.keys()))

        return True

def _generate_base_route(model_array):
    tables = {}

    for model in model_array:
        fields = []
        table_name = model.shortcuts.table_name
        required_fields = model.shortcuts.required_fields
        for field, field_object in model.shortcuts.fields.items():
            field_name = field_object.name
            field_type = field_object.get_db_field()
            is_required = field_name in required_fields or field_type == 'primary_key'

            fields.append({
                'field_name': field_name,
                'field_type': field_type,
                'is_required': is_required
            })

        tables[table_name] = {
            'route_url': model.route_url if hasattr(model, 'route_url') else '/{}'.format(table_name),
            'fields': fields
        }

    async def base_route(request):
        response_data = {
            'data': {'routes': tables},
            'status_code': 200,
            'message': 'OK'
        }

        return json(response_data, status=200)

    return base_route

def list_browser_pages(request):
    renderer = request.app.prerender
    pages = await renderer.pages()
    return response.json(pages, ensure_ascii=False, indent=2, escape_forward_slashes=False)

def show_brower_version(request):
    renderer = request.app.prerender
    version = await renderer.version()
    return response.json(version, ensure_ascii=False, indent=2, escape_forward_slashes=False)

def disable_browser_rendering(request):
    global CONCURRENCY

    CONCURRENCY = 0
    return response.json({'message': 'success'})

def enable_browser_rendering(request):
    global CONCURRENCY

    CONCURRENCY = int(os.environ.get('CONCURRENCY', cpu_count() * 2))
    return response.json({'message': 'success'})

def test_async(request):
    return json({"test": True})

def test_sync(request):
    return json({"test": True})

def test(request, exception):
    return json({"exception": "{}".format(exception), "status": exception.status_code}, status=exception.status_code)


# ----------------------------------------------- #
# Read from request
# ----------------------------------------------- #

def post_json(request):
    return json({"received": True, "message": request.json})

def post_json(request):
    return json({"received": True, "form_data": request.form, "test": request.form.get('test')})

def test(request):
    log.info("Received GET /")
    return json(await expensive_call())

def foo(request):
    return json({'msg': 'hi from blueprint'})

def foo2(request):
    return json({'msg': 'hi from blueprint2'})


#
# ????:
#   - ?? flask
#

def handle(request):
    async with engine.acquire() as conn:
        result = []
        async for row in conn.execute(polls.select()):
            result.append({"question": row.question, "pub_date": row.pub_date})
        return json({"polls": result})

def fetch(session, url):
    """
    Use session object to perform 'get' request on url
    """
    async with session.get(url) as response:
        return await response.json()


#
# ???? HTTP ????:
#   - ???? HTTP ??
#