-
-
Notifications
You must be signed in to change notification settings - Fork 853
/
resources.py
1708 lines (1411 loc) · 55.5 KB
/
resources.py
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
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Jira resource definitions.
This module implements the Resource classes that translate JSON from Jira REST
resources into usable objects.
"""
from __future__ import annotations
import json
import logging
import re
import time
from typing import TYPE_CHECKING, Any, Dict, List, Type, cast
from requests import Response
from requests.structures import CaseInsensitiveDict
from jira.resilientsession import ResilientSession, parse_errors
from jira.utils import json_loads, remove_empty_attributes, threaded_requests
if TYPE_CHECKING:
from jira.client import JIRA
AnyLike = Any
else:
class AnyLike:
"""Dummy subclass of base object class for when type checker is not running."""
pass
__all__ = (
"Resource",
"Issue",
"Comment",
"Project",
"Attachment",
"Component",
"Dashboard",
"DashboardItemProperty",
"DashboardItemPropertyKey",
"Filter",
"DashboardGadget",
"Votes",
"PermissionScheme",
"Watchers",
"Worklog",
"IssueLink",
"IssueLinkType",
"IssueProperty",
"IssueSecurityLevelScheme",
"IssueType",
"IssueTypeScheme",
"NotificationScheme",
"Priority",
"PriorityScheme",
"Version",
"WorkflowScheme",
"Role",
"Resolution",
"SecurityLevel",
"Status",
"User",
"Group",
"CustomFieldOption",
"RemoteLink",
"Customer",
"ServiceDesk",
"RequestType",
"resource_class_map",
)
logging.getLogger("jira").addHandler(logging.NullHandler())
class Resource:
"""Models a URL-addressable resource in the Jira REST API.
All Resource objects provide the following:
``find()`` -- get a resource from the server and load it into the current object (though clients should use the methods in the JIRA class instead of this method directly)
``update()`` -- changes the value of this resource on the server and returns a new resource object for it
``delete()`` -- deletes this resource from the server
``self`` -- the URL of this resource on the server
``raw`` -- dict of properties parsed out of the JSON response from the server
Subclasses will implement ``update()`` and ``delete()`` as appropriate for the specific resource.
All Resources have a resource path of the form:
* ``issue``
* ``project/{0}``
* ``issue/{0}/votes``
* ``issue/{0}/comment/{1}``
where the bracketed numerals are placeholders for ID values that are filled in from the ``ids`` parameter to ``find()``.
"""
JIRA_BASE_URL = "{server}/rest/{rest_path}/{rest_api_version}/{path}"
# A prioritized list of the keys in self.raw most likely to contain a
# human readable name or identifier, or that offer other key information.
_READABLE_IDS = (
"displayName",
"key",
"name",
"accountId",
"filename",
"value",
"scope",
"votes",
"id",
"mimeType",
"closed",
)
# A list of properties that should uniquely identify a Resource object.
# Each of these properties should be hashable, usually strings
_HASH_IDS = (
"self",
"type",
"key",
"id",
"name",
)
def __init__(
self,
resource: str,
options: dict[str, Any],
session: ResilientSession,
base_url: str = JIRA_BASE_URL,
):
"""Initializes a generic resource.
Args:
resource (str): The name of the resource.
options (Dict[str,str]): Options for the new resource
session (ResilientSession): Session used for the resource.
base_url (Optional[str]): The Base Jira url.
"""
self._resource = resource
self._options = options
self._session = session
self._base_url = base_url
# Explicitly define as None, so we know when a resource has actually been loaded
self.raw: dict[str, Any] | None = None
def __str__(self) -> str:
"""Return the first value we find that is likely to be human-readable.
Returns:
str
"""
if self.raw:
for name in self._READABLE_IDS:
if name in self.raw:
pretty_name = str(self.raw[name])
# Include any child to support nested select fields.
if hasattr(self, "child"):
pretty_name += " - " + str(self.child)
return pretty_name
# If all else fails, use repr to make sure we get something.
return repr(self)
def __repr__(self) -> str:
"""Identify the class and include any and all relevant values.
Returns:
str
"""
names: list[str] = []
if self.raw:
for name in self._READABLE_IDS:
if name in self.raw:
names.append(name + "=" + repr(self.raw[name]))
if not names:
return f"<JIRA {self.__class__.__name__} at {id(self)}>"
return f"<JIRA {self.__class__.__name__}: {', '.join(names)}>"
def __getattr__(self, item: str) -> Any:
"""Allow access of attributes via names.
Args:
item (str): Attribute Name
Raises:
AttributeError: When attribute does not exist.
Returns:
Any: Attribute value.
"""
try:
return self[item] # type: ignore
except Exception as e:
if hasattr(self, "raw") and self.raw is not None and item in self.raw:
return self.raw[item]
else:
raise AttributeError(
f"{self.__class__!r} object has no attribute {item!r} ({e})"
)
def __getstate__(self) -> dict[str, Any]:
"""Pickling the resource."""
return vars(self)
def __setstate__(self, raw_pickled: dict[str, Any]):
"""Unpickling of the resource."""
# https://stackoverflow.com/a/50888571/7724187
vars(self).update(raw_pickled)
def __hash__(self) -> int:
"""Hash calculation.
We try to find unique identifier like properties to form our hash object.
Technically 'self', if present, is the unique URL to the object, and should be sufficient to generate a unique hash.
"""
hash_list = []
for a in self._HASH_IDS:
if hasattr(self, a):
hash_list.append(getattr(self, a))
if hash_list:
return hash(tuple(hash_list))
else:
raise TypeError(f"'{self.__class__}' is not hashable")
def __eq__(self, other: Any) -> bool:
"""Default equality test.
Checks the types look about right and that the relevant attributes that uniquely identify a resource are equal.
"""
return isinstance(other, self.__class__) and all(
[
getattr(self, a) == getattr(other, a)
for a in self._HASH_IDS
if hasattr(self, a)
]
)
def find(
self,
id: tuple[str, ...] | int | str,
params: dict[str, str] | None = None,
):
"""Finds a resource based on the input parameters.
Args:
id (Union[Tuple[str, str], int, str]): id
params (Optional[Dict[str, str]]): params
"""
if params is None:
params = {}
if isinstance(id, tuple):
path = self._resource.format(*id)
else:
path = self._resource.format(id)
url = self._get_url(path)
self._find_by_url(url, params)
def _find_by_url(
self,
url: str,
params: dict[str, str] | None = None,
):
"""Finds a resource on the specified url.
The resource is loaded with the JSON data returned by doing a
request on the specified url.
Args:
url (str): url
params (Optional[Dict[str, str]]): params
"""
self._load(url, params=params)
def _get_url(self, path: str) -> str:
"""Gets the url for the specified path.
Args:
path (str): str
Returns:
str
"""
options = self._options.copy()
options.update({"path": path})
return self._base_url.format(**options)
def update(
self,
fields: dict[str, Any] | None = None,
async_: bool | None = None,
jira: JIRA = None,
notify: bool = True,
**kwargs: Any,
):
"""Update this resource on the server.
Keyword arguments are marshalled into a dict before being sent. If this resource doesn't support ``PUT``, a :py:exc:`.JIRAError`
will be raised; subclasses that specialize this method will only raise errors in case of user error.
Args:
fields (Optional[Dict[str, Any]]): Fields which should be updated for the object.
async_ (Optional[bool]): True to add the request to the queue, so it can be executed later using async_run()
jira (jira.client.JIRA): Instance of Jira Client
notify (bool): True to notify watchers about the update, sets parameter notifyUsers. (Default: ``True``).
Admin or project admin permissions are required to disable the notification.
kwargs (Any): extra arguments to the PUT request.
"""
if async_ is None:
async_: bool = self._options["async"] # type: ignore # redefinition
data = {}
if fields is not None:
data.update(fields)
data.update(kwargs)
if not notify:
querystring = "?notifyUsers=false"
else:
querystring = ""
r = self._session.put(self.self + querystring, data=json.dumps(data))
if "autofix" in self._options and r.status_code == 400:
user = None
error_list = parse_errors(r)
logging.error(error_list)
if (
"The reporter specified is not a user." in error_list
and "reporter" not in data["fields"]
):
logging.warning(
"autofix: setting reporter to '%s' and retrying the update."
% self._options["autofix"]
)
data["fields"]["reporter"] = {"name": self._options["autofix"]}
if (
"Issues must be assigned." in error_list
and "assignee" not in data["fields"]
):
logging.warning(
"autofix: setting assignee to '{}' for {} and retrying the update.".format(
self._options["autofix"], self.key
)
)
data["fields"]["assignee"] = {"name": self._options["autofix"]}
if (
"Issue type is a sub-task but parent issue key or id not specified."
in error_list
):
logging.warning(
"autofix: trying to fix sub-task without parent by converting to it to bug"
)
data["fields"]["issuetype"] = {"name": "Bug"}
if (
"The summary is invalid because it contains newline characters."
in error_list
):
logging.warning("autofix: trying to fix newline in summary")
data["fields"]["summary"] = self.fields.summary.replace("/n", "")
for error in error_list:
if re.search(
r"^User '(.*)' was not found in the system\.", error, re.U
):
m = re.search(
r"^User '(.*)' was not found in the system\.", error, re.U
)
if m:
user = m.groups()[0]
else:
raise NotImplementedError()
if re.search(r"^User '(.*)' does not exist\.", error):
m = re.search(r"^User '(.*)' does not exist\.", error)
if m:
user = m.groups()[0]
else:
raise NotImplementedError()
if user and jira:
logging.warning(
"Trying to add missing orphan user '%s' in order to complete the previous failed operation."
% user
)
jira.add_user(user, "noreply@example.com", 10100, active=False)
# if 'assignee' not in data['fields']:
# logging.warning("autofix: setting assignee to '%s' and retrying the update." % self._options['autofix'])
# data['fields']['assignee'] = {'name': self._options['autofix']}
# EXPERIMENTAL --->
if async_: # FIXME: no async
if not hasattr(self._session, "_async_jobs"):
self._session._async_jobs = set() # type: ignore
self._session._async_jobs.add( # type: ignore
threaded_requests.put( # type: ignore
self.self, data=json.dumps(data)
)
)
else:
r = self._session.put(self.self, data=json.dumps(data))
time.sleep(self._options["delay_reload"])
self._load(self.self)
def delete(self, params: dict[str, Any] | None = None) -> Response | None:
"""Delete this resource from the server, passing the specified query parameters.
If this resource doesn't support ``DELETE``, a :py:exc:`.JIRAError` will be raised; subclasses that specialize this method will
only raise errors in case of user error.
Args:
params: Parameters for the delete request.
Returns:
Optional[Response]: Returns None if async
"""
if self._options["async"]:
# FIXME: mypy doesn't think this should work
if not hasattr(self._session, "_async_jobs"):
self._session._async_jobs = set() # type: ignore
self._session._async_jobs.add( # type: ignore
threaded_requests.delete(url=self.self, params=params) # type: ignore
)
return None
else:
return self._session.delete(url=self.self, params=params)
def _load(
self,
url: str,
headers=CaseInsensitiveDict(),
params: dict[str, str] | None = None,
path: str | None = None,
):
"""Load a resource.
Args:
url (str): url
headers (Optional[CaseInsensitiveDict]): headers. Defaults to CaseInsensitiveDict().
params (Optional[Dict[str,str]]): params to get request. Defaults to None.
path (Optional[str]): field to get. Defaults to None.
Raises:
ValueError: If json cannot be loaded
"""
r = self._session.get(url, headers=headers, params=params)
try:
j = json_loads(r)
except ValueError as e:
logging.error(f"{e}:\n{r.text}")
raise e
if path:
j = j[path]
self._parse_raw(j)
def _parse_raw(self, raw: dict[str, Any]):
"""Parse a raw dictionary to create a resource.
Args:
raw (Dict[str, Any])
"""
self.raw = raw
if not raw:
raise NotImplementedError(f"We cannot instantiate empty resources: {raw}")
dict2resource(raw, self, self._options, self._session)
def _default_headers(self, user_headers):
# result = dict(user_headers)
# result['accept'] = 'application/json'
return CaseInsensitiveDict(
self._options["headers"].items() + user_headers.items()
)
class Attachment(Resource):
"""An issue attachment."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "attachment/{0}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def get(self):
"""Return the file content as a string."""
r = self._session.get(self.content, headers={"Accept": "*/*"})
return r.content
def iter_content(self, chunk_size=1024):
"""Return the file content as an iterable stream."""
r = self._session.get(self.content, stream=True)
return r.iter_content(chunk_size)
class Component(Resource):
"""A project component."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "component/{0}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def delete(self, moveIssuesTo: str | None = None): # type: ignore[override]
"""Delete this component from the server.
Args:
moveIssuesTo: the name of the component to which to move any issues this component is applied
"""
params = {}
if moveIssuesTo is not None:
params["moveIssuesTo"] = moveIssuesTo
super().delete(params)
class CustomFieldOption(Resource):
"""An existing option for a custom issue field."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "customFieldOption/{0}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
class Dashboard(Resource):
"""A Jira dashboard."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "dashboard/{0}", options, session)
if raw:
self._parse_raw(raw)
self.gadgets: list[DashboardGadget] = []
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
class DashboardItemPropertyKey(Resource):
"""A jira dashboard item property key."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "dashboard/{0}/items/{1}/properties", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
class DashboardItemProperty(Resource):
"""A jira dashboard item."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(
self, "dashboard/{0}/items/{1}/properties/{2}", options, session
)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def update( # type: ignore[override] # incompatible supertype ignored
self, dashboard_id: str, item_id: str, value: dict[str, Any]
) -> DashboardItemProperty:
"""Update this resource on the server.
Keyword arguments are marshalled into a dict before being sent. If this resource doesn't support ``PUT``, a :py:exc:`.JIRAError`
will be raised; subclasses that specialize this method will only raise errors in case of user error.
Args:
dashboard_id (str): The ``id`` if the dashboard.
item_id (str): The id of the dashboard item (``DashboardGadget``) to target.
value (dict[str, Any]): The value of the targeted property key.
Returns:
DashboardItemProperty
"""
options = self._options.copy()
options["path"] = (
f"dashboard/{dashboard_id}/items/{item_id}/properties/{self.key}"
)
self.raw["value"].update(value)
self._session.put(self.JIRA_BASE_URL.format(**options), self.raw["value"])
return DashboardItemProperty(self._options, self._session, raw=self.raw)
def delete(self, dashboard_id: str, item_id: str) -> Response: # type: ignore[override] # incompatible supertype ignored
"""Delete dashboard item property.
Args:
dashboard_id (str): The ``id`` of the dashboard.
item_id (str): The ``id`` of the dashboard item (``DashboardGadget``).
Returns:
Response
"""
options = self._options.copy()
options["path"] = (
f"dashboard/{dashboard_id}/items/{item_id}/properties/{self.key}"
)
return self._session.delete(self.JIRA_BASE_URL.format(**options))
class DashboardGadget(Resource):
"""A jira dashboard gadget."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "dashboard/{0}/gadget/{1}", options, session)
if raw:
self._parse_raw(raw)
self.item_properties: list[DashboardItemProperty] = []
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def update( # type: ignore[override] # incompatible supertype ignored
self,
dashboard_id: str,
color: str | None = None,
position: dict[str, Any] | None = None,
title: str | None = None,
) -> DashboardGadget:
"""Update this resource on the server.
Keyword arguments are marshalled into a dict before being sent. If this resource doesn't support ``PUT``, a :py:exc:`.JIRAError`
will be raised; subclasses that specialize this method will only raise errors in case of user error.
Args:
dashboard_id (str): The ``id`` of the dashboard to add the gadget to `required`.
color (str): The color of the gadget, should be one of: blue, red, yellow,
green, cyan, purple, gray, or white.
ignore_uri_and_module_key_validation (bool): Whether to ignore the
validation of the module key and URI. For example, when a gadget is created
that is part of an application that is not installed.
position (dict[str, int]): A dictionary containing position information like -
`{"column": 0, "row", 1}`.
title (str): The title of the gadget.
Returns:
``DashboardGadget``
"""
data = remove_empty_attributes(
{"color": color, "position": position, "title": title}
)
options = self._options.copy()
options["path"] = f"dashboard/{dashboard_id}/gadget/{self.id}"
self._session.put(self.JIRA_BASE_URL.format(**options), json=data)
options["path"] = f"dashboard/{dashboard_id}/gadget"
return next(
DashboardGadget(self._options, self._session, raw=gadget)
for gadget in self._session.get(
self.JIRA_BASE_URL.format(**options)
).json()["gadgets"]
if gadget["id"] == self.id
)
def delete(self, dashboard_id: str) -> Response: # type: ignore[override] # incompatible supertype ignored
"""Delete gadget from dashboard.
Args:
dashboard_id (str): The ``id`` of the dashboard.
Returns:
Response
"""
options = self._options.copy()
options["path"] = f"dashboard/{dashboard_id}/gadget/{self.id}"
return self._session.delete(self.JIRA_BASE_URL.format(**options))
class Field(Resource):
"""An issue field.
A field cannot be fetched from the Jira API individually, but paginated lists of fields are returned by some endpoints.
"""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "field/{0}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
class Filter(Resource):
"""An issue navigator filter."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "filter/{0}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
class Issue(Resource):
"""A Jira issue."""
class _IssueFields(AnyLike):
class _Comment:
def __init__(self) -> None:
self.comments: list[Comment] = []
class _Worklog:
def __init__(self) -> None:
self.worklogs: list[Worklog] = []
def __init__(self):
self.assignee: UnknownResource | None = None
self.attachment: list[Attachment] = []
self.comment = self._Comment()
self.created: str
self.description: str | None = None
self.duedate: str | None = None
self.issuelinks: list[IssueLink] = []
self.issuetype: IssueType
self.labels: list[str] = []
self.priority: Priority
self.project: Project
self.reporter: UnknownResource
self.resolution: Resolution | None = None
self.security: SecurityLevel | None = None
self.status: Status
self.statuscategorychangedate: str | None = None
self.summary: str
self.timetracking: TimeTracking
self.versions: list[Version] = []
self.votes: Votes
self.watchers: Watchers
self.worklog = self._Worklog()
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "issue/{0}", options, session)
self.fields: Issue._IssueFields
self.id: str
self.key: str
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def update( # type: ignore[override] # incompatible supertype ignored
self,
fields: dict[str, Any] = None,
update: dict[str, Any] = None,
async_: bool = None,
jira: JIRA = None,
notify: bool = True,
**fieldargs,
):
"""Update this issue on the server.
Each keyword argument (other than the predefined ones) is treated as a field name and the argument's value is treated as
the intended value for that field -- if the fields argument is used, all other keyword arguments will be ignored.
Jira projects may contain many issue types. Some issue screens have different requirements for fields in an issue.
This information is available through the :py:meth:`.JIRA.editmeta` method.
Further examples are available here: https://developer.atlassian.com/display/JIRADEV/JIRA+REST+API+Example+-+Edit+issues
Args:
fields (Dict[str,Any]): a dict containing field names and the values to use
update (Dict[str,Any]): a dict containing update the operations to apply
async_ (Optional[bool]): True to add the request to the queue, so it can be executed later using async_run() (Default: ``None``))
jira (Optional[jira.client.JIRA]): JIRA instance.
notify (bool): True to notify watchers about the update, sets parameter notifyUsers. (Default: ``True``).
Admin or project admin permissions are required to disable the notification.
fieldargs (dict): keyword arguments will generally be merged into fields, except lists, which will be merged into updates
"""
data = {}
if fields is not None:
fields_dict = fields
else:
fields_dict = {}
data["fields"] = fields_dict
if update is not None:
update_dict = update
else:
update_dict = {}
data["update"] = update_dict
for field in sorted(fieldargs.keys()):
value = fieldargs[field]
# apply some heuristics to make certain changes easier
if isinstance(value, str):
if field == "assignee" or field == "reporter":
fields_dict[field] = {"name": value}
elif field == "comment":
if "comment" not in update_dict:
update_dict["comment"] = []
update_dict["comment"].append({"add": {"body": value}})
else:
fields_dict[field] = value
elif isinstance(value, list):
if field not in update_dict:
update_dict[field] = []
update_dict[field].extend(value)
else:
fields_dict[field] = value
super().update(async_=async_, jira=jira, notify=notify, fields=data)
def get_field(self, field_name: str) -> Any:
"""Obtain the (parsed) value from the Issue's field.
Args:
field_name (str): The name of the field to get
Raises:
AttributeError: If the field does not exist or if the field starts with an ``_``
Returns:
Any: Returns the parsed data stored in the field. For example, "project" would be of class :py:class:`Project`
"""
if field_name.startswith("_"):
raise AttributeError(
f"An issue field_name cannot start with underscore (_): {field_name}",
field_name,
)
else:
return getattr(self.fields, field_name)
def add_field_value(self, field: str, value: str):
"""Add a value to a field that supports multiple values, without resetting the existing values.
This should work with: labels, multiple checkbox lists, multiple select
Args:
field (str): The field name
value (str): The field's value
"""
super().update(fields={"update": {field: [{"add": value}]}})
def delete(self, deleteSubtasks=False):
"""Delete this issue from the server.
Args:
deleteSubtasks (bool): True to also delete subtasks. If any are present the Issue won't be deleted (Default: ``True``)
"""
super().delete(params={"deleteSubtasks": deleteSubtasks})
def permalink(self):
"""Get the URL of the issue, the browsable one not the REST one.
Returns:
str: URL of the issue
"""
return f"{self._options['server']}/browse/{self.key}"
class Comment(Resource):
"""An issue comment."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "issue/{0}/comment/{1}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def update( # type: ignore[override]
# The above ignore is added because we've added new parameters and order of
# parameters is different.
# Will need to be solved in a major version bump.
self,
fields: dict[str, Any] | None = None,
async_: bool | None = None,
jira: JIRA = None,
body: str = "",
visibility: dict[str, str] | None = None,
is_internal: bool = False,
notify: bool = True,
):
"""Update a comment.
Keyword arguments are marshalled into a dict before being sent.
Args:
fields (Optional[Dict[str, Any]]): DEPRECATED => a comment doesn't have fields
async_ (Optional[bool]): True to add the request to the queue, so it can be executed later using async_run() (Default: ``None``))
jira (jira.client.JIRA): Instance of Jira Client
visibility (Optional[Dict[str, str]]): a dict containing two entries: "type" and "value".
"type" is 'role' (or 'group' if the Jira server has configured comment visibility for groups)
"value" is the name of the role (or group) to which viewing of this comment will be restricted.
body (str): New text of the comment
is_internal (bool): True to mark the comment as 'Internal' in Jira Service Desk (Default: ``False``)
notify (bool): True to notify watchers about the update, sets parameter notifyUsers. (Default: ``True``).
Admin or project admin permissions are required to disable the notification.
"""
data: dict[str, Any] = {}
if body:
data["body"] = body
if visibility:
data["visibility"] = visibility
if is_internal:
data["properties"] = [
{"key": "sd.public.comment", "value": {"internal": is_internal}}
]
super().update(async_=async_, jira=jira, notify=notify, fields=data)
class RemoteLink(Resource):
"""A link to a remote application from an issue."""
def __init__(
self,
options: dict[str, str],
session: ResilientSession,
raw: dict[str, Any] = None,
):
Resource.__init__(self, "issue/{0}/remotelink/{1}", options, session)
if raw:
self._parse_raw(raw)
self.raw: dict[str, Any] = cast(Dict[str, Any], self.raw)
def update(self, object, globalId=None, application=None, relationship=None):
"""Update a RemoteLink. 'object' is required.
For definitions of the allowable fields for 'object' and the keyword arguments 'globalId', 'application' and 'relationship',
see https://developer.atlassian.com/display/JIRADEV/JIRA+REST+API+for+Remote+Issue+Links.
Args:
object: the link details to add (see the above link for details)
globalId: unique ID for the link (see the above link for details)
application: application information for the link (see the above link for details)
relationship: relationship description for the link (see the above link for details)
"""
data = {"object": object}
if globalId is not None:
data["globalId"] = globalId
if application is not None:
data["application"] = application
if relationship is not None:
data["relationship"] = relationship
super().update(**data)
class Votes(Resource):
"""Vote information on an issue."""
def __init__(
self,
options: dict[str, str],