python类application()的实例源码

def search_entity_aware_index(query, offset, count, draw, qf):
        """
        This function is responsible for hitting the solr endpoint
        and returning the results back.
        """
        results = SOLR_ENTITYAWAREINDEX.search(q=query, **{
            'start': int(offset),
            'rows': int(count),
            'qf': qf
        })
        print("Saw {0} result(s) for query {1}.".format(len(results), query))
        formatted_hits = []
        for hit in results.docs:
            formatted_hits.append(
                [hit['_news_title'], hit['_news_publisher'], CATEGORY[hit['_news_category'][0]], hit['_news_url']])
        response = {'draw': draw,
                    'recordsFiltered': results.hits,
                    'data': formatted_hits}
        web.header('Content-Type', 'application/json')
        return json.dumps(response)

def search(query, offset, count, draw, solr_endpoint):
    """
    This function is responsible for hitting the solr endpoint
    and returning the results back.
    """
    results = solr_endpoint.search(q=query, **{
        'start': int(offset),
        'rows': int(count)
    })
    print("Saw {0} result(s) for query {1}.".format(len(results), query))
    formatted_hits = []
    for hit in results.docs:
        formatted_hits.append(
            [hit['_news_title'], hit['_news_publisher'], CATEGORY[hit['_news_category'][0]], hit['_news_url']])
    response = {'draw': draw,
                'recordsFiltered': results.hits,
                'data': formatted_hits}
    web.header('Content-Type', 'application/json')
    return json.dumps(response)

def search_simple_index(query, offset, count, draw):
        """
        This function is responsible for hitting the solr endpoint
        and returning the results back.
        """
        results = SOLR_SIMPLEINDEX.search(q=query, **{
            'start': int(offset),
            'rows': int(count),
            'cache': 'false'
        })
        print("Saw {0} result(s) for query {1}.".format(len(results), query))
        formatted_hits = []
        for hit in results.docs:
            formatted_hits.append(
                [hit['_news_title'], hit['_news_publisher'], CATEGORY[hit['_news_category'][0]], hit['_news_url']])
        response = {'draw': draw,
                    'recordsFiltered': results.hits,
                    'data': formatted_hits}
        web.header('Content-Type', 'application/json')
        return json.dumps(response)

def __init__(self):
        web.application.__init__(self)
        self.__name__ = self
        self.app = None
        self.lock = threading.Lock()

def __init__(self, port=8080, host='0.0.0.0'):  # pragma: no cover
        d_client = docker.from_env()
        d_client.images.pull('cyberreboot/vent-ncapture', tag='master')
        nf_inst = NControl()
        urls = nf_inst.urls()
        app = web.application(urls, globals())
        web.httpserver.runsimple(app.wsgifunc(), (host, port))

def start_web_app():
    """ starts the web app in a TestApp for testing """
    nf_inst = ncontrol.NControl()
    urls = nf_inst.urls()
    app = web.application(urls, globals())
    test_app = TestApp(app.wsgifunc())
    return test_app

def test_create_r():
    """ tests the restful endpoint: create """
    # get web app
    test_app = start_web_app()

    # test create
    r = test_app.post('/create', params={'id': 'foo',
                                         'interval': '60',
                                         'filter': '',
                                         'nic': 'eth1'},
                      headers={'Content-Type': 'application/json'})
    assert r.status == 200
    r = test_app.post('/create', params={'id': 'foo',
                                         'interval': '60',
                                         'iters': '1',
                                         'filter': '',
                                         'nic': 'eth1'},
                      headers={'Content-Type': 'application/json'})
    assert r.status == 200
    r = test_app.post('/create', params={})
    assert r.status == 200
    r = test_app.post('/create', params={'nic': 'eth1'})
    assert r.status == 200
    r = test_app.post('/create', params={'nic': 'eth1', 'id': 'foo'})
    assert r.status == 200
    r = test_app.post(
        '/create', params={'nic': 'eth1', 'id': 'foo', 'interval': '61'})
    assert r.status == 200
    r = test_app.post('/create', params='{}')
    assert r.status == 200
    r = test_app.post('/create', params={'id': 'foo',
                                         'interval': '60',
                                         'filter': '',
                                         'metadata': '{"foo": "bar"}',
                                         'iters': '1',
                                         'nic': 'eth1'},
                      headers={'Content-Type': 'application/json'})
    assert r.status == 200

def startServer(self):
        sys.argv.append('0.0.0.0:%s' % config.PORT)
        app = web.application(self.urls, globals())
        app.run()

def set_json_response():
        """
        ????content-type?json
        :return:
        """
        web.header('content-type', 'application/json;charset=utf-8', unique=True)

def __init__( self, title, port, main_page ):
        self.title = title
        portal_pages['main/structure'] = main_page 
        main_page.addToPortal( self )
        self.app = web.application( self.urls, globals() )

def GET( self, x ):
        print 'in structure: '+x
        web.header('Content-Type', 'application/json')
        layout = portal_pages[x]
        return '{"layout":'+ layout.to_JSON() +'}'

def GET(self):
        web.header('Content-Type', "application/json")
        return json.dumps({'status': 'ok'})

def POST(self):
        data = web.input()
        if not ("ident" in data and "csr" in data):
            raise web.BadRequest()

        crt = sign_csr(data.csr, data.ident)
        web.header('Content-Type', "application/json")
        return json.dumps({'certificate': crt})

def GET(self):
        web.header('Content-Type','application/json') 

        topic = web.input(value=' ')

        try:
            output = run_model.model(requirement = [topic.value])
        except:
            output = 'Invalid query'
            pass

        return output

def POST(self):
    global modelCache
    params = json.loads(web.data())
    requestInput = web.input()
    id = requestInput["id"]
    # We will always return the active cells because they are cheap.
    returnSnapshots = [TM_SNAPS.ACT_CELLS]
    from pprint import pprint; pprint(params)
    tm = TM(**params)

    tmFacade = TmFacade(tm, ioClient, modelId=id)

    modelId = tmFacade.getId()
    modelCache[modelId]["tm"] = tmFacade
    modelCache[modelId]["classifier"] = SDRClassifierFactory.create(implementation="py")
    modelCache[modelId]["recordsSeen"] = 0

    print "Created TM {}".format(modelId)

    payload = {
      "meta": {
        "id": modelId,
        "saving": returnSnapshots
      }
    }

    tmState = tmFacade.getState(*returnSnapshots)

    for key in tmState:
      payload[key] = tmState[key]

    web.header("Content-Type", "application/json")
    return json.dumps(payload)

def run(self):
        port = 8080
        app = web.application(self.urls, {
            'bot':self.bot, 
            'bot_commands':self.bot_commands, 
            'bot_status_all':self.bot_status_all})
        web.httpserver.runsimple(app.wsgifunc(), ("0.0.0.0", self.port))

def start_api_server():
    sys.argv.append('0.0.0.0:%s' % config.API_PORT)
    app = web.application(urls, globals())
    app.run()

def main():
    # arguments = parse_arguments()

    app = web.application(urls, globals())
    print 'Starting daemons for updating the cache and confidence rating..'
    costi_api.start_update_daemon()
    costi_api.start_rating_daemon()
    print 'Start server..'
    app.run()

def GET(self):
       all_col = ('name','description','owner','family_id','time','change_log')
       citype_input = web.input()
       condition = " "
       for col in range(len(all_col)):
           col_name = all_col[col]
           value = citype_input.get(col_name,None)
           if value <> None:
               if col_name == 'time' :
                   condition = condition + "ct.starttime <= '" + value + "' and ct.endtime > '" + value + "' and "
               else :
                   condition = condition + "ct." + col_name + " = '" + value + "' and "
           if value == None and col_name == 'time':
               condition = condition + "ct.endtime = '" + ENDTIME + "' and "

       v_sql = "select ct.name , convert(ct.description,'utf8') description,ct.owner,ct.family_id,convert(ct.displayname,'utf8') " \
               "displayname,ct.change_log from t_ci_type ct where " + condition + "1=1"
       #v_sql = "select ct.name , ct.description,ct.owner,ct.family_id,ct.displayname,ct.change_log from t_ci_type ct where " + condition + "1=1"
       ci_type = db.query(v_sql)
       ci_as_dict = []
       for ci in ci_type:
           ci_as_dict.append(ci)
       ci_type_json = json.dumps(ci_as_dict, indent = 4,ensure_ascii=False, separators = (',',':')).decode("GB2312")
       #Type is unicode
#        import sys,httplib, urllib
#        params = urllib.urlencode({'fid':'FCIT00000004','change_log':'test'})
#        headers = {'Content-type': 'application/x-www-form-urlencoded', 'Accept': 'text/plain'}
#        con2  =   httplib.HTTPConnection("localhost:8080")
#        con2.request("DELETE","/citype",params,headers)
#        con2.close()

       return ci_type_json

def GET(self):
       all_col = ('value','description','type_fid','ci_fid','owner','family_id','time','change_log','citype_name','ci_name','ciat_name','value_type') 
       ci_input = web.input()
       condition = " "
       for col in range(len(all_col)):
           col_name = all_col[col]
           value = ci_input.get(col_name,None)
           if value <> None:
               if col_name == 'time' :
                   condition = condition + "a.starttime <= '" + value + "' and a.endtime > '" + value + "' and b.starttime <= '" + value + "' and b.endtime > '" + value + "' and c.starttime <= '" + value + "' and c.endtime > '" + value + "' and d.starttime <= '" + value + "' and d.endtime > '" + value + "' and "
               elif col_name == 'citype_name':
                   condition = condition + "d.name = '" + value + "' and "
               elif col_name == 'ci_name':
                   condition = condition + "b.name = '" + value + "' and "
               elif col_name == 'ciat_name':
                   condition = condition + "c.name = '" + value + "' and "
               elif col_name == 'value_type':
                   condition = condition + "c.value_type = '" + value + "' and "
               else :
                   condition = condition + "a." + col_name + " = '" + value + "' and "
           if value == None and col_name == 'time':
               condition = condition + "a.endtime = '" + ENDTIME + "' and b.endtime = '" + ENDTIME + "' and c.endtime = '" + ENDTIME + "' and d.endtime = '" + ENDTIME + "' and "

       v_sql = "select distinct d.name citype_name, b.name ci_name, c.name ciat_name, c.value_type, a.value, convert(a.description,'utf8') description, a.type_fid, a.ci_fid, a.owner, a.family_id, a.change_log from t_ci_attribute a, t_ci b, t_ci_attribute_type c, t_ci_type d where " + condition + "a.type_fid = c.family_id and a.ci_fid = b.family_id and b.type_fid = d.family_id and c.ci_type_fid = d.family_id "
       ci_recs = db.query(v_sql)
       ci_as_dict = []
       for ci in ci_recs:
           ci_as_dict.append(ci)
       ci_json = json.dumps(ci_as_dict, indent = 4,ensure_ascii=False, separators = (',',':')).decode("GB2312")
#        import sys,httplib, urllib 
#        params = urllib.urlencode({'fid':'FCAD00000002','change_log':'test'})
#        headers = {'Content-type': 'application/x-www-form-urlencoded', 'Accept': 'text/plain'}
#        con2  =   httplib.HTTPConnection("localhost:8080") 
#        con2.request("DELETE","/ciattr",params,headers) 
#        con2.close()

       return ci_json