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
|
"""
Module for jenkinsapi nodes
"""
from __future__ import annotations
from typing import Iterator
import logging
from urllib.parse import urlencode
from jenkinsapi.node import Node
from jenkinsapi.jenkinsbase import JenkinsBase
from jenkinsapi.custom_exceptions import JenkinsAPIException
from jenkinsapi.custom_exceptions import UnknownNode
from jenkinsapi.custom_exceptions import PostRequired
log: logging.Logger = logging.getLogger(__name__)
class Nodes(JenkinsBase):
"""
Class to hold information on a collection of nodes
"""
def __init__(self, baseurl: str, jenkins_obj: "Jenkins") -> None:
"""
Handy access to all of the nodes on your Jenkins server
"""
self.jenkins = jenkins_obj
JenkinsBase.__init__(
self,
(
baseurl.rstrip("/")
if "/computer" in baseurl
else baseurl.rstrip("/") + "/computer"
),
)
def get_jenkins_obj(self) -> "Jenkins":
return self.jenkins
def __str__(self) -> str:
return "Nodes @ %s" % self.baseurl
def __contains__(self, node_name: str) -> bool:
return node_name in self.keys()
def iterkeys(self) -> Iterator[str]:
"""
Return an iterator over the container's node names.
Using iterkeys() while creating nodes may raise a RuntimeError
or fail to iterate over all entries.
"""
for item in self._data["computer"]:
yield item["displayName"]
def keys(self) -> list[str]:
"""
Return a copy of the container's list of node names.
"""
return list(self.iterkeys())
def _make_node(self, nodename) -> Node:
"""
Creates an instance of Node for the given nodename.
This function assumes the returned node exists.
"""
if nodename.lower() == "built-in node":
nodeurl = "%s/(%s)" % (self.baseurl, "built-in")
else:
nodeurl = "%s/%s" % (self.baseurl, nodename)
return Node(self.jenkins, nodeurl, nodename, node_dict={})
def iteritems(self) -> Iterator[tuple[str, Node]]:
"""
Return an iterator over the container's (name, node) pairs.
Using iteritems() while creating nodes may raise a RuntimeError or
fail to iterate over all entries.
"""
for item in self._data["computer"]:
nodename = item["displayName"]
try:
yield nodename, self._make_node(nodename)
except Exception:
raise JenkinsAPIException("Unable to iterate nodes")
def items(self) -> list[tuple[str, Node]]:
"""
Return a copy of the container's list of (name, node) pairs.
"""
return list(self.iteritems())
def itervalues(self) -> Iterator[Node]:
"""
Return an iterator over the container's nodes.
Using itervalues() while creating nodes may raise a RuntimeError
or fail to iterate over all entries.
"""
for item in self._data["computer"]:
try:
yield self._make_node(item["displayName"])
except Exception:
raise JenkinsAPIException("Unable to iterate nodes")
def values(self) -> list[Node]:
"""
Return a copy of the container's list of nodes.
"""
return list(self.itervalues())
def __getitem__(self, nodename: str) -> Node:
if nodename in self:
return self._make_node(nodename)
raise UnknownNode(nodename)
def __len__(self) -> int:
return len(self.keys())
def __delitem__(self, item: str) -> None:
if item in self and item != "Built-In Node":
url = "%s/doDelete" % self[item].baseurl
try:
self.jenkins.requester.get_and_confirm_status(url)
except PostRequired:
# Latest Jenkins requires POST here. GET kept for compatibility
self.jenkins.requester.post_and_confirm_status(url, data={})
self.poll()
else:
if item != "Built-In Node":
raise UnknownNode("Node %s does not exist" % item)
log.info("Requests to remove built-in node ignored")
def __setitem__(self, name: str, node_dict: dict):
if not isinstance(node_dict, dict):
raise ValueError('"node_dict" parameter must be a Node dict')
if name not in self:
self.create_node(name, node_dict)
self.poll()
def create_node(self, name: str, node_dict: dict) -> Node:
"""
Create a new slave node
:param str name: name of slave
:param dict node_dict: node dict (See Node class)
:return: node obj
"""
if name in self:
return self[name]
node = Node(
jenkins_obj=self.jenkins,
baseurl="",
nodename=name,
node_dict=node_dict,
poll=False,
)
url = "%s/computer/doCreateItem?%s" % (
self.jenkins.baseurl,
urlencode(node.get_node_attributes()),
)
data = {"json": urlencode(node.get_node_attributes())}
self.jenkins.requester.post_and_confirm_status(url, data=data)
self.poll()
return self[name]
def create_node_with_config(self, name: str, config: dict) -> Node | None:
"""
Create a new slave node with specific configuration.
Config should be resemble the output of node.get_node_attributes()
:param str name: name of slave
:param dict config: Node attributes for Jenkins API request
to create node
(See function output Node.get_node_attributes())
:return: node obj
"""
if name in self:
return self[name]
if not isinstance(config, dict):
return None
url = "%s/computer/doCreateItem?%s" % (
self.jenkins.baseurl,
urlencode(config),
)
data = {"json": urlencode(config)}
self.jenkins.requester.post_and_confirm_status(url, data=data)
self.poll()
return self[name]
|
