Python bottle.response 模块，headers() 实例源码

def _error(error, code):
    """Return an error as json"""
    _error = {
        "status": str(code),
        "title": errors[code],
        "detail": error.body
    }
    traceback.print_exception(type(error), error, error.traceback)
    response.headers["Content-Type"] = "application/vnd.api+json"
    return response_object(errors=[_error])

def get_resources():
    """Return the resources of the authenticated user.

    If authentication failed, this aborts the execution with
    401 Unauthorized.
    """
    #pprint(dict(request.headers))
    header = request.environ.get('HTTP_AUTHORIZATION','')
    if header:
        print("Authorization:", header)
    basic = request.auth
    if basic:
        username, password = basic
        if passwords.get(username) != password:
            abort(401, BASIC_ERROR)
    else:
        api_key = get_api_key()
        if api_key is not None:
            username = api_keys.get(api_key)
            if username is None:
                abort(401, API_KEY_ERROR)
        else:
            username = None
    return _resources[username]

def getStations(self, busca=''):
        print
        response.headers['Access-Control-Allow-Origin']='*'
        response.headers['Content-Type']='application/json'
        _estacoes = self._g._get() if busca == '' else self._g._busca(busca)
        _bloco = []
        for _estacao in _estacoes:
            _nome, _lat, _long, _endereco, _linha, _statusOnline, _StatusOperacional, _disp1, _disp2, _total, _internalStatus, _img, _id = _estacao
            _bloco.append({ "type" : "Feature",
                            "geometry" : {
                                "coordinates": [ _long, _lat ],
                                "type": "Point" },
                            "properties": {
                                "id" : _id,
                                "nome" : _nome,
                                "endereco" : _endereco,
                                "estacao" : _linha,
                                "status_online": _statusOnline,
                                "status_operacional" : _StatusOperacional,
                                "qtd_bikes_disp_1" : _disp1,
                                "qtd_bikes_disp_2" : _disp2,
                                "qtd_vagas_total" : _total,
                                "statusInterno" : _internalStatus }
                         })
        return json.dumps({ "features" : _bloco, "type": "FeatureCollection" }, indent=2, sort_keys=True)

def enable_cors(func):
    def _enable_cors(*args, **kwargs):
        hds = response.headers
        hds['Access-Control-Allow-Origin'] = '*'
        hds['Access-Control-Allow-Methods'] = ', '.join(['PUT', 'GET',
                                                         'POST', 'DELETE',
                                                         'OPTIONS'])
        allow = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'
        hds['Access-Control-Allow-Headers'] = allow
        if request.method != 'OPTIONS':
            return func(*args, **kwargs)
    return _enable_cors

def get_status(jid):
    # following headers are needed because of CORS
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, OPTIONS'
    response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'
    resp = db(jobs.id==jid).select(jobs.state).first()
    if resp is None: return 'X'
    else: return resp.state

def output():
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'PUT, GET, POST, DELETE, OPTIONS'
    response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

    app = request.query.app
    cid = request.query.cid
    user = request.query.user

    try:
        if re.search("/", cid):
            (u, c) = cid.split("/")
        else:
            u = user
            c = cid
        run_dir = os.path.join(user_dir, u, app, c)
        fn = os.path.join(run_dir, app + '.out')
        output = slurp_file(fn)
        # the following line will convert HTML chars like > to entities >
        # this is needed so that XML input files will show paramters labels
        output = cgi.escape(output)
        return output
        # params = { 'cid': cid, 'contents': output, 'app': app,
        #            'user': u, 'fn': fn, 'apps': myapps.keys() }
        # return template('more', params)
    except:
        return "ERROR: something went wrong!"
        # params = { 'app': app, 'apps': myapps.keys(),
        #            'err': "Couldn't read input file. Check casename." }
        # return template('error', params)

def output():
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'PUT, GET, POST, DELETE, OPTIONS'
    response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

    app = request.query.app
    cid = request.query.cid
    user = request.query.user

    try:
        if re.search("/", cid):
            (u, c) = cid.split("/")
        else:
            u = user
            c = cid
        run_dir = os.path.join(user_dir, u, app, c)
        fn = os.path.join(run_dir, app + '.out')
        output = slurp_file(fn)
        # the following line will convert HTML chars like > to entities >
        # this is needed so that XML input files will show paramters labels
        output = cgi.escape(output)
        return output
        # params = { 'cid': cid, 'contents': output, 'app': app,
        #            'user': u, 'fn': fn, 'apps': myapps.keys() }
        # return template('more', params)
    except:
        return "ERROR: something went wrong!"
        # params = { 'app': app, 'apps': myapps.keys(),
        #            'err': "Couldn't read input file. Check casename." }
        # return template('error', params)

def js_dynamic(path):
    response.headers['Expires'] = time.strftime("%a, %d %b %Y %H:%M:%S GMT",
                                                time.gmtime(time.time() + 60 * 60 * 24 * 2))
    response.headers['Cache-control'] = "public"
    response.headers['Content-Type'] = "text/javascript; charset=UTF-8"

    try:
        # static files are not rendered
        if "static" not in path and "mootools" not in path:
            t = env.get_template("js/%s" % path)
            return t.render()
        else:
            return static_file(path, root=join(PROJECT_DIR, "media", "js"))
    except:
        return HTTPError(404, "Not Found")

def server_static(path):
    response.headers['Expires'] = time.strftime("%a, %d %b %Y %H:%M:%S GMT",
                                                time.gmtime(time.time() + 60 * 60 * 24 * 7))
    response.headers['Cache-control'] = "public"
    return static_file(path, root=join(PROJECT_DIR, "media"))

def get_api_key():
    """Return the api key or None."""
    header = request.headers.get('Authorization')
    if not header: return
    try:
        method, data = header.split(None, 1)
        if method.lower() != 'api-key': return
        return touni(base64.b64decode(tob(data[4:])))
    except (ValueError, TypeError):
        abort(401, HEADER_ERROR)

def get_endpoint_url():
    """Return the url that this server is reachable at."""
    return "http://" + request.headers["Host"] + BASE

def test_jsonapi_header():
    """Make sure that the content type is set accordingly.

    http://jsonapi.org/format/#content-negotiation-clients
    """
    content_type = request.content_type
    content_type_expected = "application/vnd.api+json"
    if content_type != content_type_expected and content_type.startswith(content_type_expected):
        abort(415, "The Content-Type header must be \"{}\", not \"{}\".".format(
                   content_type_expected, content_type))
    accepts = request.headers.get("Accept", "*/*").split(",")
    expected_accept = ["*/*", "application/*", "application/vnd.api+json"]
    if not any([accept in expected_accept for accept in accepts]):
        abort(406, "The Accept header must one of \"{}\", not \"{}\".".format(
                       expected_accept, ",".join(accepts)))

def enable_cors():
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers[
        'Access-Control-Allow-Methods'] = 'PUT, GET, POST, DELETE, OPTIONS'
    response.headers[
        'Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

def _devices_info(self):
        response.headers['Content-Type'] = 'application/json'
        devices_list = { 'devices': self._get_devices_info() }
        return json.dumps(devices_list)

def _devices_value(self):
        response.headers['Content-Type'] = 'application/json'
        devices_list = { 'devices': self._get_devices_value() }
        return json.dumps(devices_list)

def _device_set(self):
        response.headers['Content-Type'] = 'application/json'
        retval = { }

        def verify_and_set(client):
            try:
                data = json.loads(request.body.read().decode('ascii'))
            except:
                raise ValueError('Unable to decode json data from PUT request')

            if data is None:
                raise ValueError('No data set in body of PUT request')

            if not 'name' in data:
                raise KeyError('No "name" field in body of PUT request')

            if not 'value' in data:
                raise KeyError('No "value" field in body of PUT request')

            client.set_device_value(data['name'], data['value'])

        if self._debug:
            verify_and_set(self)
        else:
            try:
                verify_and_set(self)
            except Exception as e:
                retval['error'] = str(e)

        return json.dumps(retval)

def custom_headers():
    """Set some custom headers across all HTTP responses."""
    response.headers["Server"] = "Machete Server"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Pragma"] = "no-cache"
    response.headers["Cache-Control"] = "no-cache"
    response.headers["Expires"] = "0"

def custom_headers():
    """Set some custom headers across all HTTP responses."""
    response.headers["Server"] = "Machete Server"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Pragma"] = "no-cache"
    response.headers["Cache-Control"] = "no-cache"
    response.headers["Expires"] = "0"

def enable_cors(fn):
    def _enable_cors(*args, **kwargs):
        # set CORS headers
        response.headers['Access-Control-Allow-Origin'] = '*'
        response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, OPTIONS'
        response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

        if bottle.request.method != 'OPTIONS':
            # actual request; reply with the actual response
            return fn(*args, **kwargs)

    return _enable_cors

def _add_cors_headers(response):
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, OPTIONS'
    response.headers['Access-Control-Allow-Headers'] = \
            'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

def to_json(fnc):
    @wraps(fnc)
    def inner(*args, **kwargs):
        bottle_resp.headers['Content-Type'] = 'application/json'
        from_func = fnc(*args, **kwargs)
        if from_func is not None:
            return json.dumps({'result': from_func})
    return inner

def enable_cors():
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS'
    response.headers['Access-Control-Allow-Headers'] = 'Origin, Accept, Content-Type, X-Requested-With, X-CSRF-Token'

# Enable SSL w Self Signed Cert
#   Source: https://github.com/mfm24/miscpython/blob/master/bottle_ssl.py
#   Generate a cert: openssl req -new -x509 -keyout mfa_slipstream.pem -out mfa_slipstream.pem -days 365 -nodes