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
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
"""
This module implements the Jobs class, which is intended to be a container-like
interface for all of the jobs defined on a single Jenkins server.
"""
from __future__ import annotations
from typing import Iterator
import logging
import time
from jenkinsapi.job import Job
from jenkinsapi.custom_exceptions import JenkinsAPIException, UnknownJob
log = logging.getLogger(__name__)
class Jobs(object):
"""
This class provides a container-like API which gives
access to all jobs defined on the Jenkins server. It behaves
like a dict in which keys are Job-names and values are actual
jenkinsapi.Job objects.
"""
def __init__(self, jenkins: "Jenkins") -> None:
self.jenkins = jenkins
self._data = []
def _del_data(self, job_name: str) -> None:
if not self._data:
return
for num, job_data in enumerate(self._data):
if job_data["name"] == job_name:
del self._data[num]
return
def __len__(self) -> int:
return len(self.keys())
def poll(self, tree="jobs[name,color,url]"):
return self.jenkins.poll(tree=tree)
def __delitem__(self, job_name: str) -> None:
"""
Delete a job by name
:param str job_name: name of a existing job
:raises JenkinsAPIException: When job is not deleted
"""
if job_name in self:
try:
delete_job_url = self[job_name].get_delete_url()
self.jenkins.requester.post_and_confirm_status(
delete_job_url, data="some random bytes..."
)
self._del_data(job_name)
except JenkinsAPIException:
# Sometimes jenkins throws NPE when removing job
# It removes job ok, but it is good to be sure
# so we re-try if job was not deleted
if job_name in self:
delete_job_url = self[job_name].get_delete_url()
self.jenkins.requester.post_and_confirm_status(
delete_job_url, data="some random bytes..."
)
self._del_data(job_name)
def __setitem__(self, key: str, value: str) -> "Job":
"""
Create Job
:param str key: Job name
:param str value: XML configuration of the job
.. code-block:: python
api = Jenkins('http://localhost:8080/')
new_job = api.jobs['my_new_job'] = config_xml
"""
return self.create(key, value)
def __getitem__(self, job_name: str) -> "Job":
if job_name in self:
job_data = [
job_row
for job_row in self._data
if job_row["name"] == job_name
or Job.get_full_name_from_url_and_baseurl(
job_row["url"], self.jenkins.baseurl
)
== job_name
][0]
return Job(job_data["url"], job_data["name"], self.jenkins)
else:
raise UnknownJob(job_name)
def iteritems(self) -> Iterator[str, "Job"]:
"""
Iterate over the names & objects for all jobs
"""
for job in self.itervalues():
if job.name != job.get_full_name():
yield job.get_full_name(), job
else:
yield job.name, job
def __contains__(self, job_name: str) -> bool:
"""
True if job_name exists in Jenkins
"""
return job_name in self.keys()
def iterkeys(self) -> Iterator[str]:
"""
Iterate over the names of all available jobs
"""
if not self._data:
self._data = self.poll().get("jobs", [])
for row in self._data:
if row["name"] != Job.get_full_name_from_url_and_baseurl(
row["url"], self.jenkins.baseurl
):
yield Job.get_full_name_from_url_and_baseurl(
row["url"], self.jenkins.baseurl
)
else:
yield row["name"]
def itervalues(self) -> Iterator["Job"]:
"""
Iterate over all available jobs
"""
if not self._data:
self._data = self.poll().get("jobs", [])
for row in self._data:
yield Job(row["url"], row["name"], self.jenkins)
def keys(self) -> list[str]:
"""
Return a list of the names of all jobs
"""
return list(self.iterkeys())
def create(self, job_name: str, config: str | bytes) -> "Job":
"""
Create a job
:param str jobname: Name of new job
:param str config: XML configuration of new job
:returns Job: new Job object
"""
if job_name in self:
return self[job_name]
if not config:
raise JenkinsAPIException("Job XML config cannot be empty")
params = {"name": job_name}
if isinstance(config, bytes):
config = config.decode("utf-8")
self.jenkins.requester.post_xml_and_confirm_status(
self.jenkins.get_create_url(), data=config, params=params
)
# Reset to get it refreshed from Jenkins
self._data = []
return self[job_name]
def create_multibranch_pipeline(
self, job_name: str, config: str, block: bool = True, delay: int = 60
) -> list["Job"]:
"""
Create a multibranch pipeline job
:param str jobname: Name of new job
:param str config: XML configuration of new job
:param block: block until scan is finished?
:param delay: max delay to wait for scan to finish (seconds)
:returns list of new Jobs after scan
"""
if not config:
raise JenkinsAPIException("Job XML config cannot be empty")
params = {"name": job_name}
if isinstance(config, bytes):
config = config.decode("utf-8")
self.jenkins.requester.post_xml_and_confirm_status(
self.jenkins.get_create_url(), data=config, params=params
)
# Reset to get it refreshed from Jenkins
self._data = []
# Launch a first scan / indexing to discover the branches...
self.jenkins.requester.post_and_confirm_status(
"{}/job/{}/build".format(self.jenkins.baseurl, job_name),
data="",
valid=[200, 302], # expect 302 without redirects
allow_redirects=False,
)
start_time = time.time()
# redirect-url does not work with indexing;
# so the only workaround found is to parse the console output
# until scan has finished.
scan_finished = False
while not scan_finished and block and time.time() < start_time + delay:
indexing_console_text = self.jenkins.requester.get_url(
"{}/job/{}/indexing/consoleText".format(
self.jenkins.baseurl, job_name
)
)
if (
indexing_console_text.text.strip()
.split("\n")[-1]
.startswith("Finished:")
):
scan_finished = True
time.sleep(1)
# now search for all jobs created; those who start with job_name + '/'
jobs = []
for name in self.jenkins.get_jobs_list():
if name.startswith(job_name + "/"):
jobs.append(self[name])
return jobs
def copy(self, job_name: str, new_job_name: str) -> "Job":
"""
Copy a job
:param str job_name: Name of an existing job
:param new_job_name: Name of new job
:returns Job: new Job object
"""
params = {"name": new_job_name, "mode": "copy", "from": job_name}
self.jenkins.requester.post_and_confirm_status(
self.jenkins.get_create_url(), params=params, data=""
)
self._data = []
return self[new_job_name]
def rename(self, job_name: str, new_job_name: str) -> "Job":
"""
Rename a job
:param str job_name: Name of an existing job
:param str new_job_name: Name of new job
:returns Job: new Job object
"""
params = {"newName": new_job_name}
rename_job_url = self[job_name].get_rename_url()
self.jenkins.requester.post_and_confirm_status(
rename_job_url, params=params, data=""
)
self._data = []
return self[new_job_name]
def build(self, job_name: str, params=None, **kwargs) -> "QueueItem":
"""
Executes build of a job
:param str job_name: Job name
:param dict params: Job parameters
:param kwargs: Parameters for Job.invoke() function
:returns QueueItem: Object to track build progress
"""
if params:
assert isinstance(params, dict)
return self[job_name].invoke(build_params=params, **kwargs)
return self[job_name].invoke(**kwargs)
|
