python类full_path()的实例源码

def key_prefix():
    cache_key = 'view_%s' % request.full_path
    return cache_key

def after_request(response):
    timestamp = strftime('[%Y-%b-%d %H:%M]')
    f = open("server.log","a").write( "\n"+"--"*10+"\n"+'%s %s %s %s %s %s'%(timestamp, request.remote_addr, request.method, request.scheme, request.full_path, response.status) )
    return response

def exceptions(e):
    tb = traceback.format_exc()
    timestamp = strftime('[%Y-%b-%d %H:%M]')
    f = open("server.log","a").write( "\n"+"--"*10+"\n"+'%s %s %s %s %s 5xx INTERNAL SERVER ERROR\n%s'%(timestamp, request.remote_addr, request.method, request.scheme, request.full_path, tb) )
    return abort(500)

def get_api(config, tools, models):
    # config
    API_NAME = config.get('api_journey', 'name')
    DEFAULT_PAGE_SIZE = config.get('api_journey', 'page_size')
    CACHE_TIMEOUT_SECS = config.getint('cache', 'timeout')
    cache = tools['cache']
    mongo = tools['mongo']
    auth = tools['auth']
    event_model = models['events']

    # controller
    ctrler = Blueprint(API_NAME, __name__)

    def make_cache_key():
      """Make a key that includes GET parameters."""
      return request.full_path

    # route
    @ctrler.route('/actors/<actor_id>/events', methods=['GET'])
    @cache.cached(key_prefix=make_cache_key, timeout=CACHE_TIMEOUT_SECS)
    def get_actor_events(actor_id):
        pagesize = int(request.args.get('_page_size', default=DEFAULT_PAGE_SIZE))
        page = int(request.args.get('_page', default=1))

        # parse condition
        cond = parse_query_to_mongo_cond(request.args, {
                'actor.id': actor_id
            })
        output = event_model.query_by_page(cond, pagesize, page)
        return jsonify(output)


    @ctrler.route('/events', methods=['GET'])
    @auth.login_required
    def list_all_events():
        return jsonify({'events': [e for e in event_model.find()]})


    @ctrler.route('/events', methods=['POST'])
    def send_new_event():
        # async operation
        pub = tools['pubsub']['producer']
        event = request.get_json()
        pub.publish(event)
        return jsonify({'status': 'message is sent to backend for processing', 'event': event}), 201


    return {'prefix': '/'+API_NAME, 'ctrler': ctrler}

def ignition():
    """
    Ignition
    ---
    tags:
      - matchbox
    responses:
      200:
        description: Ignition configuration
        schema:
            type: dict
      403:
        description: Matchbox unavailable
        schema:
            type: text/plain
      503:
        description: Matchbox is out of sync
        schema:
            type: text/plain
    """
    cache_key = "sync-notify"
    last_sync_ts = CACHE.get(cache_key)
    app.logger.debug("cacheKey: %s is set with value %s" % (cache_key, last_sync_ts))

    # we ignore the sync status if we arrived from /ignition-pxe because it's a discovery PXE boot
    if last_sync_ts is None and request.path != "/ignition-pxe":
        app.logger.error("matchbox state is out of sync: cacheKey: %s is None" % cache_key)
        return Response("matchbox is out of sync", status=503, mimetype="text/plain")

    if request.path == "/ignition-pxe":
        app.logger.info("%s %s" % (request.method, request.url))

    matchbox_uri = application.config.get("MATCHBOX_URI")
    if matchbox_uri:
        try:
            # remove the -pxe from the path because matchbox only serve /ignition
            path = request.full_path.replace("/ignition-pxe?", "/ignition?")
            matchbox_resp = requests.get("%s%s" % (matchbox_uri, path))
            resp = matchbox_resp.content
            matchbox_resp.close()
            return Response(resp, status=matchbox_resp.status_code, mimetype="text/plain")
        except requests.RequestException as e:
            app.logger.error("fail to query matchbox ignition %s" % e)
            return Response("matchbox doesn't respond", status=502, mimetype="text/plain")

    return Response("matchbox=%s" % matchbox_uri, status=403, mimetype="text/plain")