這篇文章主要介紹python中如何使用saltstack api，文中介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！
重慶分公司
重慶分公司
# -*- coding:utf-8 -*-
import sys
import json
import pycurl
from io import BytesIO
class PyCurl(object):
def __init__(self, url, **kwargs):
# 傳入url地址
self.url = url
# 取出header相關(guān)信息
self.header = kwargs.get("header", None)
# 創(chuàng)建一個(gè)curl對(duì)象
self.curl = pycurl.Curl()
# setopt 來設(shè)置一些請求選項(xiàng)
# 指定請求的URL
self.curl.setopt(self.curl.URL, self.url)
# 設(shè)置代理瀏覽器
self.curl.setopt(self.curl.HEADER, False)
# 設(shè)置請求方式
self.curl.setopt(self.curl.POST, True)
# 設(shè)置https方式
self.curl.setopt(pycurl.SSL_VERIFYPEER, 0)
self.curl.setopt(pycurl.SSL_VERIFYHOST, 0)
# 判斷header是否存在
if self.header:
# 設(shè)置模擬瀏覽器
self.curl.setopt(self.curl.HTTPHEADER, self.header)
def request(self, data=None, timeout=None):
# 判斷對(duì)象類型 是否為 str
if isinstance(data, str):
# 將數(shù)據(jù)提交
self.curl.setopt(pycurl.POSTFIELDS, data)
header_buf = BytesIO()
body_buf = BytesIO()
# 強(qiáng)制獲取新的連接，即替代緩存中的連接
self.curl.setopt(self.curl.FRESH_CONNECT, True)
# 完成交互后強(qiáng)制斷開連接，不重用
self.curl.setopt(self.curl.FORBID_REUSE, True)
if str(timeout).isdigit() and timeout > 0:
# 設(shè)置timeout超時(shí)時(shí)間
self.curl.setopt(self.curl.TIMEOUT, timeout)
# 將返回的HTTP HEADER定向到回調(diào)函數(shù)header_buf
self.curl.setopt(self.curl.HEADERFUNCTION, header_buf.write)
# 將返回的內(nèi)容定向到回調(diào)函數(shù)body_buf
self.curl.setopt(self.curl.WRITEFUNCTION, body_buf.write)
try:
# 服務(wù)器返回信息
self.curl.perform()
except pycurl.error:
return False
# 狀態(tài)碼
http_code = self.curl.getinfo(self.curl.HTTP_CODE)
# 關(guān)閉連接
self.curl.close()
# 返回狀態(tài)碼 header body
return {"http_code": http_code, "header": header_buf.getvalue(), "body": body_buf.getvalue(), "url": self.url}
class SaltApi(object):
def __init__(self, **kwargs):
# 設(shè)置超時(shí)時(shí)間
self.timeout = kwargs.get("timeout", 300)
# 設(shè)置頭信息
self.header = kwargs.get("header", ["Content-Type:application/json"])
# 獲取url
self.__url = "https://192.168.44.20:8888"
# 獲取
self.__username = "saltapi"
self.__password = "password"
# token id 獲取
def token_id(self):
obj = {'eauth': 'pam', 'username': self.__username, 'password': self.__password}
result = self.post(prefix="/login/", **obj)
if result:
try:
self.__token_id = result['return'][0]['token']
except KeyError:
raise KeyError
print('##',self.__token_id,'##')
return self.__token_id
def post(self, prefix="/", token=None, **data):
# url拼接
url = self.__url + prefix
print (data)
# 實(shí)例化
self.header.append(str(token))
curl = PyCurl(url, header=self.header)
# 發(fā)起請求
result = curl.request(data=json.dumps(data), timeout=self.timeout)
# 判斷值
if not result:
return result
# 判斷狀態(tài)碼是否等于200
if result["http_code"] != 200:
self.response = "response code %s".format(result["info"]["http_code"])
return self.response
result = json.loads(result["body"].decode())
# 判斷是否有error
if "error" in result and result["error"]:
self.response = "%s(%s)" % (result["error"]["data"], result["error"]["code"])
return self.response
# 返回正確的數(shù)據(jù)
return result
def all_key(self):
'''
獲取所有的minion_key
'''
token = 'X-Auth-Token:%s' % self.token_id()
obj = {'client': 'wheel', 'fun': 'key.list_all'}
content = self.post(token=token, **obj)
# 取出認(rèn)證已經(jīng)通過的
minions = content['return'][0]['data']['return']['minions']
# print('已認(rèn)證',minions)
# 取出未通過認(rèn)證的
minions_pre = content['return'][0]['data']['return']['minions_pre']
# print('未認(rèn)證',minions_pre)
return minions, minions_pre
def accept_key(self, node_name):
'''
如果你想認(rèn)證某個(gè)主機(jī) 那么調(diào)用此方法
'''
token = 'X-Auth-Token:%s' % self.token_id()
obj = {'client': 'wheel', 'fun': 'key.accept', 'match': node_name}
content = self.post(token=token, **obj)
print (content)
ret = content['return'][0]['data']['success']
return ret
# 刪除認(rèn)證方法
def delete_key(self, node_name):
obj = {'client': 'wheel', 'fun': 'key.delete', 'match': node_name}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
ret = content['return'][0]['data']['success']
return ret
'''
tgt_type --
The type of tgt. Allowed values:
glob - Bash glob completion - Default
pcre - Perl style regular expression
list - Python list of hosts
grain - Match based on a grain comparison
grain_pcre - Grain comparison with a regex
pillar - Pillar data comparison
pillar_pcre - Pillar data comparison with a regex
nodegroup - Match on nodegroup
range - Use a Range server for matching
compound - Pass a compound match string
ipcidr - Match based on Subnet (CIDR notation) or IPv4 address.
Changed in version 2017.7.0: Renamed from expr_form to tgt_type
可以針對(duì)各種類型進(jìn)行執(zhí)行響應(yīng)的命令
sapi = SaltApi()
ret = sapi.host_remote_func(tgt="ba*",fun='test.ping',tgt_type='pcre')
print('%s\n%s'%('pcre',ret))
ret = sapi.host_remote_func(tgt="bamboo,Min-Centos1",fun='test.ping',tgt_type='list')
print('%s\n%s' % ('list', ret))
ret = sapi.host_remote_func(tgt="*",fun='test.ping')
print('%s\n%s' % ('glob', ret))
ret = sapi.host_remote_func(tgt="os:CentOS",fun='test.ping',tgt_type='grain')
print('%s\n%s' % ('grain', ret))
ret = sapi.host_remote_func(tgt="L@Min-Centos1 or G@os:Ubuntu",fun='test.ping',tgt_type='compound')
print('%s\n%s' % ('compound', ret))
ret = sapi.host_remote_func(tgt="192.168.44.0/24",fun='test.ping',tgt_type='ipcidr')
'''
# 針對(duì)主機(jī)遠(yuǎn)程執(zhí)行模塊
# def host_remote_func(self, tgt, fun,expr_from='glob'):
def host_remote_func(self, tgt, fun, tgt_type='glob'):
''' tgt是主機(jī) fun是模塊
寫上模塊名 返回 可以用來調(diào)用基本的資產(chǎn)
例如 curl -k https://ip地址:8080/ \
> -H "Accept: application/x-yaml" \
> -H "X-Auth-Token:b50e90485615309de0d83132cece2906f6193e43" \
> -d client='local' \
> -d tgt='*' \
> -d fun='test.ping' 要執(zhí)行的模塊
return:
- iZ28r91y66hZ: true
node2.minion: true
'''
obj = {'client': 'local', 'tgt': tgt, 'fun': fun, 'tgt_type': tgt_type}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
ret = content['return'][0]
return ret
def group_remote_func(self, tgt, fun):
obj = {'client': 'local', 'tgt': tgt, 'fun': fun, 'expr_form': 'nodegroup'}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
print (content)
ret = content['return'][0]
return ret
def host_remote_execution_module(self, tgt, fun, arg, tgt_type='glob'):
'執(zhí)行fun 傳入傳入?yún)?shù)arg '
obj = {'client': 'local', 'tgt': tgt, 'fun': fun, 'arg': arg, 'tgt_type': tgt_type}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
ret = content['return'][0]
return ret
# print(salt_aa.host_remote_execution_module('*', 'cmd.run', 'ifconfig'))
# 基于分組來執(zhí)行
def group_remote_execution_module(self, tgt, fun, arg):
'''
根據(jù)分組來執(zhí)行
tgt =
'''
obj = {'client': 'local', 'tgt': tgt, 'fun': fun, 'arg': arg, 'expr_form': 'nodegroup'}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]
return jid
def host_sls(self, tgt, arg, tgt_type='glob'):
'''主機(jī)進(jìn)行sls'''
obj = {'client': 'local', 'tgt': tgt, 'fun': 'state.sls', 'arg': arg, 'tgt_type': tgt_type}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
return content
def group_sls(self, tgt, arg):
''' 分組進(jìn)行sls '''
obj = {'client': 'local', 'tgt': tgt, 'fun': 'state.sls', 'arg': arg, 'expr_form': 'nodegroup'}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]['jid']
return jid
def host_sls_async(self, tgt, arg, tgt_type='glob'):
'''主機(jī)異步sls '''
obj = {'client': 'local_async', 'tgt': tgt, 'fun': 'state.sls', 'arg': arg, 'tgt_type': tgt_type}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]['jid']
return jid
def group_sls_async(self, tgt, arg):
'''分組異步sls '''
obj = {'client': 'local_async', 'tgt': tgt, 'fun': 'state.sls', 'arg': arg, 'expr_form': 'nodegroup'}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]['jid']
return jid
def server_group_pillar(self, tgt, arg, **kwargs):
'''分組進(jìn)行sls and pillar'''
obj = {'client': 'local', 'tgt': tgt, 'fun': 'state.sls', 'arg': arg, 'expr_form': 'nodegroup',
'kwarg': kwargs}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]
print (jid)
def server_hosts_pillar(self, tgt, arg, **kwargs):
'''針對(duì)主機(jī)執(zhí)行sls and pillar '''
obj = {"client": "local", "tgt": tgt, "fun": "state.sls", "arg": arg, "kwarg": kwargs}
token = 'X-Auth-Token:%s' % self.token_id()
content = self.post(token=token, **obj)
jid = content['return'][0]
print content
return jid
def jobs_all_list(self):
'''打印所有jid緩存'''
token = 'X-Auth-Token:%s' % self.token_id()
obj = {"client": "runner", "fun": "jobs.list_jobs"}
content = self.post(token=token, **obj)
print (content)
def jobs_jid_status(self, jid):
'''查看jid運(yùn)行狀態(tài)'''
token = 'X-Auth-Token:%s' % self.token_id()
obj = {"client": "runner", "fun": "jobs.lookup_jid", "jid": jid}
content = self.post(token=token, **obj)
print (content)
return content
# if __name__ == '__main__':
# sa = SaltApi()
# print (sa.host_remote_func("bamboo", 'test.ping', 'list'))以上是“python中如何使用saltstack api”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對(duì)大家有幫助，更多相關(guān)知識(shí)，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
