Skip to content

Instantly share code, notes, and snippets.

@sky-joker sky-joker/zbxapi.py
Last active Jun 11, 2019

Embed
What would you like to do?
Zabbix APIをpythonで実行するテンプレート
#!/usr/bin/env python3
from collections import defaultdict
import requests
import json
import atexit
class zbxapi:
"""
Zabbix APIを処理するクラス
"""
def __init__(self):
self.url = ''
self.user = ''
self.passwd = ''
self.token = ''
def login(self):
"""
ログインメソッド
ログインに成功したらTokenを取得する
:rtype: str
:return: Zabbix API Auth Token
"""
params = multi_dimension_dict(1)
params['user'] = self.user
params['password'] = self.passwd
# Tokenを取得
r = self.post('user.login', params)
self.token = json.loads(r)['result']
def logout(self):
"""
ログアウトメソッド
:rtype: str
:return: ログアウトの成功(True)・失敗(False)
"""
params = multi_dimension_dict(1)
r = self.post('user.logout', params)
return json.loads(r)['result']
def get(self, method, params):
"""
Zabbixから情報を取得するメソッド
:type method: str
:param method: Zabbix APIメソッド
:type params: dict
:param params: 取得したい情報のパラメーター
:rtype: json
:return: レスポンスボディ
"""
data = multi_dimension_dict(1)
data['jsonrpc'] = '2.0'
data['method'] = method
data['params'] = params
data['id'] = 1
data['auth'] = self.token
r = requests.get(self.url,
data=json.dumps(data),
headers=get_headers(),
verify=False)
if (r.status_code == 200):
return r.text
def post(self, method, params):
"""
Zabbixへ情報を送るメソッド
:type method: str
:param method: Zabbix APIメソッド
:type params: dict
:param params: 送信する情報のパラメーター
:rtype: json
:return: レスポンスボディ
"""
data = multi_dimension_dict(1)
data['jsonrpc'] = '2.0'
data['method'] = method
data['params'] = params
data['id'] = 1
if(self.token): data['auth'] = self.token
r = requests.post(self.url,
data=json.dumps(data),
headers=get_headers(),
verify=False)
if(r.status_code == 200):
return(r.text)
def multi_dimension_dict(dimension, callable_obj=int):
"""
pythonで多次元連想配列を使う関数
参照元: http://materia.jp/blog/20121119.html
"""
nodes = defaultdict(callable_obj)
for i in range(dimension-1):
p = nodes.copy()
nodes = defaultdict(lambda : defaultdict(p.default_factory))
return nodes
def get_headers():
"""
ヘッダー情報
:rtype: dict
:return: Zabbix APIを実行するためのヘッダー
"""
headers = {
'Content-Type':'application/json-rpc'
}
return headers
if __name__ == "__main__":
# Login
zapi = zbxapi()
zapi.url = 'http://IP or FQDN/zabbix/api_jsonrpc.php'
zapi.user = 'admin'
zapi.passwd = 'zabbix'
zapi.login()
atexit.register(zapi.logout)
# ここから処理を書いていく
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.