1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
# Authors:
# @author: David Blaisonneau <david.blaisonneau@orange.com>
# @author: Arnaud Morin <arnaud1.morin@orange.com>
import json
import requests
from requests_futures.sessions import FuturesSession
from pprint import pformat
class Api:
"""
Api class
Class to deal with the foreman API v2
"""
def __init__(self, password, login='admin', ip='127.0.0.1', printErrors=False):
""" Function __init__
Init the API with the connection params
@param password: authentication password
@param password: authentication login - default is admin
@param ip: api ip - default is localhost
@return RETURN: self
"""
self.base_url = 'http://{}/api/v2/'.format(ip)
self.headers = {'Accept': 'version=2',
'Content-Type': 'application/json; charset=UTF-8'}
self.auth = (login, password)
self.errorMsg = ''
self.printErrors = printErrors
def list(self, obj, filter=False, only_id=False, limit=20):
""" Function list
Get the list of an object
@param obj: object name ('hosts', 'puppetclasses'...)
@param filter: filter for objects
@param only_id: boolean to only return dict with name/id
@return RETURN: the list of the object
"""
self.url = '{}{}/?per_page={}'.format(self.base_url, obj, limit)
if filter:
self.url += '&search={}'.format(filter)
self.resp = requests.get(url=self.url, auth=self.auth,
headers=self.headers)
if only_id:
if self.__process_resp__(obj) is False:
return False
if type(self.res['results']) is list:
return dict((x['name'], x['id']) for x in self.res['results'])
elif type(self.res['results']) is dict:
r = {}
for v in self.res['results'].values():
for vv in v:
r[vv['name']] = vv['id']
return r
else:
return False
else:
return self.__process_resp__(obj)
def get(self, obj, id, sub_object=None):
""" Function get
Get an object by id
@param obj: object name ('hosts', 'puppetclasses'...)
@param id: the id of the object (name or id)
@return RETURN: the targeted object
"""
self.url = '{}{}/{}'.format(self.base_url, obj, id)
if sub_object:
self.url += '/' + sub_object
self.resp = requests.get(url=self.url, auth=self.auth,
headers=self.headers)
if self.__process_resp__(obj):
return self.res
return False
def get_id_by_name(self, obj, name):
""" Function get_id_by_name
Get the id of an object
@param obj: object name ('hosts', 'puppetclasses'...)
@param id: the id of the object (name or id)
@return RETURN: the targeted object
"""
list = self.list(obj, filter='name = "{}"'.format(name),
only_id=True, limit=1)
return list[name] if name in list.keys() else False
def set(self, obj, id, payload, action='', async=False):
""" Function set
Set an object by id
@param obj: object name ('hosts', 'puppetclasses'...)
@param id: the id of the object (name or id)
@param action: specific action of an object ('power'...)
@param payload: the dict of the payload
@param async: should this request be async, if true use
return.result() to get the response
@return RETURN: the server response
"""
self.url = '{}{}/{}'.format(self.base_url, obj, id)
if action:
self.url += '/{}'.format(action)
self.payload = json.dumps(payload)
if async:
session = FuturesSession()
return session.put(url=self.url, auth=self.auth,
headers=self.headers, data=self.payload)
else:
self.resp = requests.put(url=self.url, auth=self.auth,
headers=self.headers, data=self.payload)
if self.__process_resp__(obj):
return self.res
return False
def create(self, obj, payload, async=False):
""" Function create
Create an new object
@param obj: object name ('hosts', 'puppetclasses'...)
@param payload: the dict of the payload
@param async: should this request be async, if true use
return.result() to get the response
@return RETURN: the server response
"""
self.url = self.base_url + obj
self.payload = json.dumps(payload)
if async:
session = FuturesSession()
return session.post(url=self.url, auth=self.auth,
headers=self.headers, data=self.payload)
else:
self.resp = requests.post(url=self.url, auth=self.auth,
headers=self.headers,
data=self.payload)
return self.__process_resp__(obj)
def delete(self, obj, id):
""" Function delete
Delete an object by id
@param obj: object name ('hosts', 'puppetclasses'...)
@param id: the id of the object (name or id)
@return RETURN: the server response
"""
self.url = '{}{}/{}'.format(self.base_url, obj, id)
self.resp = requests.delete(url=self.url,
auth=self.auth,
headers=self.headers, )
return self.__process_resp__(obj)
def __process_resp__(self, obj):
""" Function __process_resp__
Process the response sent by the server and store the result
@param obj: object name ('hosts', 'puppetclasses'...)
@return RETURN: the server response
"""
self.last_obj = obj
if self.resp.status_code > 299:
self.errorMsg = ">> Error {} for object '{}'".format(self.resp.status_code,
self.last_obj)
try:
self.ret = json.loads(self.resp.text)
self.errorMsg += pformat(self.ret[list(self.ret.keys())[0]])
except:
self.ret = self.resp.text
self.errorMsg += self.ret
if self.printErrors:
print(self.errorMsg)
return False
self.res = json.loads(self.resp.text)
if 'results' in self.res.keys():
return self.res['results']
return self.res
def __str__(self):
ret = pformat(self.base_url) + "\n"
ret += pformat(self.headers) + "\n"
ret += pformat(self.auth) + "\n"
return ret
|
