falcon的api

一、判断session是否有效

# -*- coding: UTF-8 -*-
#!/usr/bin/env python
# Created by Administrator on 2017/12/15
import json
import requestsuser = 'admin'
sig = '1a7521abad6c11e9b4b3000c29a13d5d'
domain = 'http://172.20.16.5:8080'
api_token = '{"name":"' + user + '", "sig":"' + sig + '"}'
directiry = "/api/v1/user/auth_session"
print directiry
falcon_header = {"Apitoken": api_token,"X-Forwarded-For": "127.0.0.1","Content-Type": "application/json","name": user,"sig": sig}params = {'url': domain + directiry,'headers': falcon_header,'timeout': 30
}
res1 = requests.get(**params)
data1 = json.loads(res1.text)
print(data1)

View Code

2、获取session

# -*- coding: UTF-8 -*-
# !/usr/bin/env python
import json
import time
import requestsip = '172.20.16.5'
user = 'admin'
api_port = '8080'  # api对应端口为8080
login_path = '/api/v1/user/login'
login_url = 'http://' + ip + ':' + api_port + login_path
eventcases_path = "/api/v1/alarm/eventcases"
eventcases_url = 'http://' + ip + ':' + api_port + eventcases_path
s = requests.Session()
login_res = s.post(url=login_url,data={'name': 'admin', 'password': 'password'})login_data = json.loads(login_res.text)
# print(login_data['sig'])
sig = login_data['sig']
# print(sig)
api_token = '{"name":"' + user + '", "sig":"' + sig + '"}'# get event cases by api
falcon_header = {"Apitoken": api_token,"X-Forwarded-For": "127.0.0.1","Content-Type": "application/json","name": user,"sig": sig}params = {'url': eventcases_url,'headers': falcon_header,'timeout': 30
}payload = {# "startTime": 1466956800,"endTime": int(time.time()),"status": "PROBLEM","process_status": "ignored,unresolved",}
print params['headers']['sig']

View Code

https://blog.csdn.net/jb19900111/article/details/85064362

二、深入完善 falcon_api

1）获取基本的用户session，请求头信息，以及用户所在的id

# -*- coding: UTF-8 -*-
# !/usr/bin/env python
import json, time
import requestsurl = "http://172.20.16.5:8080"
user = 'admin'
password = 'password'
hostname = 'Yuanjin'class falcon_base():def __init__(self,user,password):self.user = userself.password = passworddef get_sig(self):login_url = url + "/api/v1/user/login"s = requests.Session()login_res = s.post(url=login_url,data={'name': self.user, 'password': self.password})login_data = json.loads(login_res.text)return login_data['sig']def get_falcon_header(self):sig = self.get_sig()api_token = '{"name":"' + user + '", "sig":"' + sig + '"}'falcon_header = {"Apitoken": api_token,"X-Forwarded-For": "127.0.0.1","Content-Type": "application/json","name": self.user,"sig": sig}return falcon_headerdef get_user_id(self):user_url = url + "/api/v1/graph/endpoint?q=%s"%(hostname)params = {'url': user_url,'headers': self.get_falcon_header(),'timeout': 30}res = requests.get(**params)data = json.loads(res.text)user_id = data[0]["id"]return user_id

falcon_base 类

2）获取监控指标。需引入基本信息

class falcon_info():def __init__(self,user,password):self.user = userself.password = passwordself.falconbase = falcon_base(self.user,self.password)def get_graph(self):#  得到具体监控项falcon_header = self.falconbase.get_falcon_header()uid = self.falconbase.get_user_id()graph_url = url + '/api/v1/graph/endpoint_counter?eid=%s' %(uid)params = {'url': graph_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)counters = [counter["counter"] for counter in data]return countersdef get_graph_history(self):# 得到指定监控项的历史记录falcon_header = self.falconbase.get_falcon_header()graph_history_url = url + "/api/v1/graph/history"end_time = int(time.time())  # 必须要整形start_time = end_time - 1800  # 30分钟counters = self.get_graph()params = {'url': graph_history_url,'headers': falcon_header,'timeout': 30}payload = {"step": 60,"start_time": start_time,"hostnames": [hostname, ],"end_time": end_time,"counters": counters,"consol_fun": "AVERAGE"}params['data'] = json.dumps(payload)res = requests.post(**params)data = json.loads(res.text)print datadata = dict([(iter["counter"], iter["Values"]) for iter in data])for key in data:values = data[key]data[key] = [{"timestamp": time.strftime('%H:%M', time.localtime(k["timestamp"])), "value": k["value"]} fork invalues if k["value"]]data["in_ip"] = hostnameprint(data)if __name__ == '__main__':falconinfo = falcon_info(user,password)falconinfo.get_graph_history()

falcon_info 类

更新

# -*- coding: UTF-8 -*-import json, time
import requests
import falcon_baseclass falcon_info():def __init__(self, user, password):self.user = userself.password = passwordself.falconbase = falcon_base.falcon_base(self.user, self.password)def get_graph(self):#  得到具体监控项falcon_header = self.falconbase.get_falcon_header()uid = self.falconbase.get_user_id()graph_url = falcon_base.url + '/api/v1/graph/endpoint_counter?eid=%s' % (uid)params = {'url': graph_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)counters = [counter["counter"] for counter in data]return countersdef get_graph_history(self):# 得到指定监控项的历史记录falcon_header = self.falconbase.get_falcon_header()graph_history_url = falcon_base.url + "/api/v1/graph/history"end_time = int(time.time())  # 必须要整形start_time = end_time - 1800  # 30分钟counters = self.get_graph()params = {'url': graph_history_url,'headers': falcon_header,'timeout': 30}payload = {"step": 60,"start_time": start_time,"hostnames": [falcon_base.hostname, ],"end_time": end_time,"counters": counters,"consol_fun": "AVERAGE"}params['data'] = json.dumps(payload)res = requests.post(**params)data = json.loads(res.text)data = dict([(iter["counter"], iter["Values"]) for iter in data])for key in data:values = data[key]data[key] = [{"timestamp": time.strftime('%H:%M', time.localtime(k["timestamp"])), "value": k["value"]} fork invalues if k["value"]]data["in_ip"] = falcon_base.hostnameprint(data)def get_user_info(self):# 获取用户信息falcon_header = self.falconbase.get_falcon_header()alarm_url = falcon_base.url + '/api/v1/user/users'params = {'url': alarm_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_template(self):# 获取模板信息falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_screens(self):falcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/dashboard/screens'print strategy_urlparams = {'url': strategy_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_alarm(self):falcon_header = self.falconbase.get_falcon_header()alarm_url = falcon_base.url + '/api/v1/alarm/event_note'# end_time = int(time.time())  # 必须要整形# start_time = end_time - 1800  # 30分钟params = {'url': alarm_url,'headers': falcon_header,'timeout': 30,}res = requests.get(**params)data = json.loads(res.text)print datadef eventcases(self):# 获取每天的报警信息eventcases_url = falcon_base.url + '/api/v1/alarm/eventcases'falcon_header = self.falconbase.get_falcon_header()s = requests.Session()oneday = 86400params = {'url': eventcases_url,'headers': falcon_header,'timeout': 30}payload = {"startTime": int(time.time() - oneday*2),"endTime": int(time.time()),"status": "PROBLEM","process_status": "ignored,unresolved",}params['data'] = json.dumps(payload)eventcases_res = s.post(**params)eventcases_data = json.loads(eventcases_res.text)for proble in eventcases_data:print proble['endpoint'],proble['timestamp'],proble['metric'],proble['note']if __name__ == '__main__':falconinfo = falcon_info(falcon_base.user, falcon_base.password)falconinfo.eventcases()

View Code

3）更新

# -*- coding: UTF-8 -*-import json, time
import requests
import falcon_baseclass falcon_info():def __init__(self, user, password):self.user = userself.password = passwordself.falconbase = falcon_base.falcon_base(self.user, self.password)def get_graph(self):#  得到具体监控项falcon_header = self.falconbase.get_falcon_header()uid = self.falconbase.get_user_id()graph_url = falcon_base.url + '/api/v1/graph/endpoint_counter?eid=%s' % (uid)params = {'url': graph_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)counters = [counter["counter"] for counter in data]return countersdef get_graph_history(self):# 得到指定监控项的历史记录falcon_header = self.falconbase.get_falcon_header()graph_history_url = falcon_base.url + "/api/v1/graph/history"end_time = int(time.time())  # 必须要整形start_time = end_time - 1800  # 30分钟counters = self.get_graph()params = {'url': graph_history_url,'headers': falcon_header,'timeout': 30}payload = {"step": 60,"start_time": start_time,"hostnames": [falcon_base.hostname, ],"end_time": end_time,"counters": counters,"consol_fun": "AVERAGE"}params['data'] = json.dumps(payload)res = requests.post(**params)data = json.loads(res.text)data = dict([(iter["counter"], iter["Values"]) for iter in data])for key in data:values = data[key]data[key] = [{"timestamp": time.strftime('%H:%M', time.localtime(k["timestamp"])), "value": k["value"]} fork invalues if k["value"]]data["in_ip"] = falcon_base.hostnameprint(data)def get_user_info(self):# 获取用户信息falcon_header = self.falconbase.get_falcon_header()alarm_url = falcon_base.url + '/api/v1/user/users'params = {'url': alarm_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_template(self):# 获取模板列表falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)# print datatemplates_list = []for  i in data['templates']:templates = {}uid =  i['template']['id']action_id =  i['template']['action_id']template = i['template']['tpl_name']templates['uid'] = uidtemplates['template'] = templatetemplates['action_id'] = action_idtemplates_list.append(templates)# print '模板列表',templates_listreturn templates_listdef get_team(self):# 获取用户组falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/team'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datateams = {}for i in data:team_id = i['team']['id']team_name = i['team']['name']# teams['team_id']= team_idteams[team_id]= team_name# team_list.append(teams)print teamsreturn teamsdef get_template_id(self):# 根据模板id 获取模板详细信息id = 1falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template/%s' %(id)params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_screens(self):falcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/dashboard/screens'print strategy_urlparams = {'url': strategy_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_alarm(self):falcon_header = self.falconbase.get_falcon_header()oneday = 86400alarm_url = falcon_base.url + '/api/v1/alarm/event_note'print alarm_urlparams = {'url': alarm_url,'headers': falcon_header,'timeout': 30,# "startTime": int(time.time() - oneday * 2),# "endTime": int(time.time()),}res = requests.get(**params)print resdata = json.loads(res.text)print datadef eventcases(self):# 获取每天的报警信息eventcases_url = falcon_base.url + '/api/v1/alarm/eventcases'falcon_header = self.falconbase.get_falcon_header()s = requests.Session()oneday = 86400params = {'url': eventcases_url,'headers': falcon_header,'timeout': 30}payload = {"startTime": int(time.time() - oneday*5),"endTime": int(time.time()),"status": "PROBLEM","process_status": "ignored,unresolved",}params['data'] = json.dumps(payload)eventcases_res = s.post(**params)eventcases_data = json.loads(eventcases_res.text)# print eventcases_dataproble_list = []template_count = {}for proble in eventcases_data:proble_count = {}template_id = proble['template_id']metric = proble['metric']if template_id not in proble_count:proble_count['template_id'] = template_idif template_id not in template_count:template_count[template_id] = 1else:template_count[template_id] += 1if metric in proble_count:proble_count[metric] += 1else:proble_count[metric] = 1proble_list.append(proble_count)print proble_listprint template_countreturn proble_list,template_countdef info(self):templates_list = self.get_template()proble_list,template_count = self.eventcases()team_list = self.get_team()action_info = {}print team_listfor i in template_count:    # 循环报警的模板 idfor j in templates_list:if j['uid'] == i:       # 根据模板id 找到对应的 模板信息template = j['template']    # 确定了告警的模板action_id =  j['action_id']actions =  team_list[action_id]  # 确定了报警的模板对应的接收组action_info['template'] = templateaction_info['actions'] = actionsaction_info['count'] = template_count[i]print action_infoif __name__ == '__main__':falconinfo = falcon_info(falcon_base.user, falcon_base.password)falconinfo.eventcases()

View Code

4）添加模板

# -*- coding: UTF-8 -*-import json, time
import requests
import falcon_basetimestr = str(time.strftime('%Y_%m_%d'))
'''
第一步：先在原来的环境获取对应模板的监控项目，写入文件。需要给指定模板的id
第二步：在新环境创建相应的模板监控项目。需要指定对应模板的id详细过程：
在falcon_base.py指定ip，用户名，密码，来获取相应的token老环境中执行函数  get_template_id()   获取到的 template.json文件新环境中执行函数 create_strategy()  创建监控项目特别强调：注意：id
'''
class falcon_template():def __init__(self, user, password):self.user = userself.password = passwordself.falconbase = falcon_base.falcon_base(self.user, self.password)def get_graph(self):#  得到具体监控项falcon_header = self.falconbase.get_falcon_header()uid = self.falconbase.get_user_id()graph_url = falcon_base.url + '/api/v1/graph/endpoint_counter?eid=%s' % (uid)params = {'url': graph_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)counters = [counter["counter"] for counter in data]return countersdef get_graph_history(self):# 得到指定监控项的历史记录falcon_header = self.falconbase.get_falcon_header()graph_history_url = falcon_base.url + "/api/v1/graph/history"end_time = int(time.time())  # 必须要整形start_time = end_time - 1800  # 30分钟counters = self.get_graph()params = {'url': graph_history_url,'headers': falcon_header,'timeout': 30}payload = {"step": 60,"start_time": start_time,"hostnames": [falcon_base.hostname, ],"end_time": end_time,"counters": counters,"consol_fun": "AVERAGE"}params['data'] = json.dumps(payload)res = requests.post(**params)data = json.loads(res.text)data = dict([(iter["counter"], iter["Values"]) for iter in data])for key in data:values = data[key]data[key] = [{"timestamp": time.strftime('%H:%M', time.localtime(k["timestamp"])), "value": k["value"]} fork invalues if k["value"]]data["in_ip"] = falcon_base.hostnameprint(data)def get_user_info(self):# 获取用户信息falcon_header = self.falconbase.get_falcon_header()alarm_url = falcon_base.url + '/api/v1/user/users'params = {'url': alarm_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_template(self):# 获取模板列表falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef create_template(self):# 创建模板falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template's = requests.Session()params = {'url': template_url,'headers': falcon_header,'timeout': 30}payload = {"parent_id": 0, # 继承现有Template"name": "Test_template" # 创建模板的名字}params['data'] = json.dumps(payload)template = s.post(**params)print template.textdef get_team(self):# 获取用户组falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/team'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datateams = {}for i in data:team_id = i['team']['id']team_name = i['team']['name']# teams['team_id']= team_idteams[team_id]= team_name# team_list.append(teams)print teamsreturn teamsdef get_template_id(self):# 根据模板id 获取模板详细信息id = 1falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template/%s' %(id)params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.content)res = json.dumps(data)with open('template.json', mode='w') as f:f.write(res)print resdef get_screens(self):falcon_header = self.falconbase.get_falcon_header()screens_url = falcon_base.url + '/api/v1/dashboard/screens'print screens_urlparams = {'url': screens_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_alarm(self):falcon_header = self.falconbase.get_falcon_header()oneday = 86400alarm_url = falcon_base.url + '/api/v1/alarm/event_note'print alarm_urlparams = {'url': alarm_url,'headers': falcon_header,'timeout': 30,# "startTime": int(time.time() - oneday * 2),# "endTime": int(time.time()),}res = requests.get(**params)print resdata = json.loads(res.text)print datadef get_strategy(self):# 获取模板监控项目id = 4falcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/strategy/%s'%idparams = {'url': strategy_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef open_strategy(self):with open('template.json','r') as f:res=f.read()dic=json.loads(res)strategy = dic['stratges']return  strategydef create_strategy(self):strategy_dic = self.open_strategy()# print strategy_dicfalcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/strategy's = requests.Session()params = {'url': strategy_url,'headers': falcon_header,'timeout': 30}for strategy in strategy_dic:payload = {"tpl_id": 10,          # 对应模板的id"tags": strategy['tags'],"run_end": strategy['run_end'],"run_begin": strategy['run_begin'],"right_value": strategy['right_value'],"priority": strategy['priority'],"op": strategy['op'],"note": strategy['note'],"metric": strategy['metric'],"max_step": strategy['max_step'],"func": strategy['func']}params['data'] = json.dumps(payload)template = s.post(**params)print template.textif __name__ == '__main__':falconinfo = falcon_template(falcon_base.user, falcon_base.password)falconinfo.create_strategy()

上面代码存在bug。right_value小于0或者为小数时无法添加。下面修改为固定值后，再做修改

# -*- coding: UTF-8 -*-import json, time
import requests
import falcon_basetimestr = str(time.strftime('%Y_%m_%d'))
'''
第一步：先在原来的环境获取对应模板的监控项目，写入文件。需要给指定模板的id
第二步：在新环境创建相应的模板监控项目。需要指定对应模板的id详细过程：
在falcon_base.py指定ip，用户名，密码，来获取相应的token老环境中执行函数  get_template_id()   获取到的 template.json文件新环境中执行函数 create_strategy()  创建监控项目特别强调：注意：id 和 "right_value" 负数，小数创建失败，类型 为 字符串的整数'''
class falcon_template():def __init__(self, user, password):self.user = userself.password = passwordself.falconbase = falcon_base.falcon_base(self.user, self.password)def get_graph(self):#  得到具体监控项falcon_header = self.falconbase.get_falcon_header()uid = self.falconbase.get_user_id()graph_url = falcon_base.url + '/api/v1/graph/endpoint_counter?eid=%s' % (uid)params = {'url': graph_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)counters = [counter["counter"] for counter in data]return countersdef get_graph_history(self):# 得到指定监控项的历史记录falcon_header = self.falconbase.get_falcon_header()graph_history_url = falcon_base.url + "/api/v1/graph/history"end_time = int(time.time())  # 必须要整形start_time = end_time - 1800  # 30分钟counters = self.get_graph()params = {'url': graph_history_url,'headers': falcon_header,'timeout': 30}payload = {"step": 60,"start_time": start_time,"hostnames": [falcon_base.hostname, ],"end_time": end_time,"counters": counters,"consol_fun": "AVERAGE"}params['data'] = json.dumps(payload)res = requests.post(**params)data = json.loads(res.text)data = dict([(iter["counter"], iter["Values"]) for iter in data])for key in data:values = data[key]data[key] = [{"timestamp": time.strftime('%H:%M', time.localtime(k["timestamp"])), "value": k["value"]} fork invalues if k["value"]]data["in_ip"] = falcon_base.hostnameprint(data)def get_user_info(self):# 获取用户信息falcon_header = self.falconbase.get_falcon_header()alarm_url = falcon_base.url + '/api/v1/user/users'params = {'url': alarm_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_template(self):# 获取模板列表falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef create_template(self):# 创建模板falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template's = requests.Session()params = {'url': template_url,'headers': falcon_header,'timeout': 30}payload = {"parent_id": 0, # 继承现有Template"name": "Test_template" # 创建模板的名字}params['data'] = json.dumps(payload)template = s.post(**params)print template.textdef get_team(self):# 获取用户组falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/team'params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datateams = {}for i in data:team_id = i['team']['id']team_name = i['team']['name']# teams['team_id']= team_idteams[team_id]= team_name# team_list.append(teams)print teamsreturn teamsdef get_template_id(self):# 根据模板id 获取模板详细信息id = 1falcon_header = self.falconbase.get_falcon_header()template_url = falcon_base.url + '/api/v1/template/%s' %(id)params = {'url': template_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.content)res = json.dumps(data)with open('template.json', mode='w') as f:f.write(res)print resdef get_screens(self):falcon_header = self.falconbase.get_falcon_header()screens_url = falcon_base.url + '/api/v1/dashboard/screens'print screens_urlparams = {'url': screens_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef get_alarm(self):falcon_header = self.falconbase.get_falcon_header()oneday = 86400alarm_url = falcon_base.url + '/api/v1/alarm/event_note'print alarm_urlparams = {'url': alarm_url,'headers': falcon_header,'timeout': 30,# "startTime": int(time.time() - oneday * 2),# "endTime": int(time.time()),}res = requests.get(**params)print resdata = json.loads(res.text)print datadef get_strategy(self):# 获取模板监控项目id = 4falcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/strategy/%s'%idparams = {'url': strategy_url,'headers': falcon_header,'timeout': 30}res = requests.get(**params)data = json.loads(res.text)print datadef open_strategy(self):with open('template2.json','r') as f:res=f.read()dic=json.loads(res)strategy = dic['stratges']return  strategydef create_strategy(self):strategy_dic = self.open_strategy()falcon_header = self.falconbase.get_falcon_header()strategy_url = falcon_base.url + '/api/v1/strategy's = requests.Session()params = {'url': strategy_url,'headers': falcon_header,'timeout': 30}for strategy in strategy_dic:unicode_value = strategy['right_value']str_value = unicode_value.encode("utf-8")float_value = float(str_value)if not float_value.is_integer():str_value = '999'if float_value < 0:str_value = '999'payload = {"tpl_id": 14,  # 对应模板的id"tags": strategy['tags'],"run_end": strategy['run_end'],"run_begin": strategy['run_begin'],"right_value": str_value,"priority": strategy['priority'],"op": strategy['op'],"note": strategy['note'],"metric": strategy['metric'],"max_step": strategy['max_step'],"func": strategy['func']}params['data'] = json.dumps(payload)template = s.post(**params)print template.textif __name__ == '__main__':falconinfo = falcon_template(falcon_base.user, falcon_base.password)falconinfo.create_strategy()

View Code

转载于:https://www.cnblogs.com/linu/p/11233790.html

falcon的api相关推荐

Open-falcon技术系列文章——安装部署open-falcon
Open-falcon技术系列文章--安装部署open-falcon 安装部署open-falcon 文章目录 Open-falcon技术系列文章--安装部署open-falcon 前言一.通过yu ...
开源运维监控open-falcon的快速部署
在目前流行的三大开源运维监控工具Zabbix.Nagios.Open-Falcon当中,Open-Falcon属于后起之秀,由小米团队提供,可以从运营级别(基本配置即可),以及应用级别(二次开发,通过 ...
Python Web框架比较，选择对的框架让开发更便捷
Python程序员有很多很好的选择来创建Web应用程序和API:Django,Weppy,Bottle和Flask引领潮流. 如果正在开发一个Web应用程序并且已经选择使用Python作为构建它的语言 ...
python高性能web框架排行榜_目前最受欢迎的12个Python web框架，你用过几个？
1.Django Django是一个开源的Web应用框架,由Python写成,支持许多数据库引擎,可以让Web开发变得迅速和可扩展,并会不断的版本更新以匹配Python最新版本,如果是新手程序员,可以 ...
falcon框架_如何使用Falcon构建RESTful API
falcon框架介绍 (Introduction) RESTful APIs are a major component of any well-architected stack, and Pyt ...
分离硬件和代码、稳定 API，PyTorch Lightning 1.0.0 版本正式发布
机器之心报道机器之心编辑部还记得那个看起来像 Keras 的轻量版 PyTorch 框架 Lightning 吗?它终于出了 1.0.0 版本,并增添了很多新功能,在度量.优化.日志记录.数据流. ...
api网关选型_如何轻松打造百亿流量API网关？看这一篇就够了（下）
如何轻松打造百亿流量API网关?看这一篇就够了(上) 上篇整体描述了网关的背景,涉及职能.分类.定位环节,本篇进入本文的重点,将会具体谈下百亿级流量API网关的演进过程. 准备好瓜子花生小板凳开始积累 ...
Apache Falcon数据集管理和数据处理平台
2019独角兽企业重金招聘Python工程师标准>>> jopen 4年前发布 | 67K 次阅读分布式/云计算/大数据 Apache Falcon Apache Falcon 是 ...
Falcon 快速开始
介绍当您的目标是构建快速.可扩展的REST风格API微服务时, Falcon 是个不错的选择. 这是一个可靠的.高性能的Python Web框架,用于构建大规模应用后端和微服务.Falcon鼓励RE ...

falcon的api

falcon的api相关推荐

最新文章

热门文章