adding type hints

Author: bandophahita

atlassian/jira.py

@@ -4,7 +4,7 @@
 import logging
 import os
 import re
-from typing import TYPE_CHECKING, Any, BinaryIO, Literal, cast
+from typing import TYPE_CHECKING, Any, BinaryIO, Literal, cast, Iterable
 from warnings import warn
 
 from deprecated import deprecated
@@ -145,7 +145,7 @@ def get_permissions(
 
         return self.get(url, params=params)
 
-    def get_all_permissions(self):
+    def get_all_permissions(self) -> dict | None:
         """
         Returns all permissions that are present in the Jira instance -
         Global, Project and the global ones added by plugins
@@ -159,7 +159,9 @@ def get_all_permissions(self):
     Reference: https://docs.atlassian.com/software/jira/docs/api/REST/8.5.0/#api/2/application-properties
     """
 
-    def get_property(self, key: T_id | None = None, permission_level: str | None = None, key_filter: str | None = None):
+    def get_property(
+        self, key: T_id | None = None, permission_level: str | None = None, key_filter: str | None = None
+    ) -> dict | None:
         """
         Returns an application property
         :param key: str
@@ -180,7 +182,7 @@ def get_property(self, key: T_id | None = None, permission_level: str | None = N
 
         return self.get(url, params=params)
 
-    def set_property(self, property_id: T_id, value: str):
+    def set_property(self, property_id: T_id, value: str) -> dict | None:
         """
         Modify an application property via PUT. The "value" field present in the PUT will override the existing value.
         :param property_id:
@@ -207,15 +209,15 @@ def get_advanced_settings(self) -> dict | None:
     Reference: https://docs.atlassian.com/software/jira/docs/api/REST/8.5.0/#api/2/applicationrole
     """
 
-    def get_all_application_roles(self):
+    def get_all_application_roles(self) -> dict | None:
         """
         Returns all ApplicationRoles in the system
         :return:
         """
         url = self.resource_url("applicationrole")
         return self.get(url) or {}
 
-    def get_application_role(self, role_key: str):
+    def get_application_role(self, role_key: str) -> dict | None:
         """
         Returns the ApplicationRole with passed key if it exists
         :param role_key: str
@@ -242,7 +244,7 @@ def get_attachments_ids_from_issue(self, issue: T_id) -> list[dict[str, str]]:
             list_attachments_id.append({"filename": attachment["filename"], "attachment_id": attachment["id"]})
         return list_attachments_id
 
-    def get_attachment(self, attachment_id: T_id):
+    def get_attachment(self, attachment_id: T_id) -> dict | None:
         """
         Returns the meta-data for an attachment, including the URI of the actual attached file
         :param attachment_id: int
@@ -301,7 +303,7 @@ def download_attachments_from_issue(self, issue: T_id, path: str | None = None,
         except Exception as e:
             raise e
 
-    def get_attachment_content(self, attachment_id: T_id):
+    def get_attachment_content(self, attachment_id: T_id) -> dict | None:
         """
         Returns the content for an attachment
         :param attachment_id: int
@@ -311,7 +313,7 @@ def get_attachment_content(self, attachment_id: T_id):
         url = f"{base_url}/content/{attachment_id}"
         return self.get(url, not_json_response=True)
 
-    def remove_attachment(self, attachment_id: T_id):
+    def remove_attachment(self, attachment_id: T_id) -> dict | None:
         """
         Remove an attachment from an issue
         :param attachment_id: int
@@ -321,7 +323,7 @@ def remove_attachment(self, attachment_id: T_id):
         url = f"{base_url}/{attachment_id}"
         return self.delete(url)
 
-    def get_attachment_meta(self):
+    def get_attachment_meta(self) -> dict | None:
         """
         Returns the meta information for an attachments,
         specifically if they are enabled and the maximum upload size allowed
@@ -330,7 +332,7 @@ def get_attachment_meta(self):
         url = self.resource_url("attachment/meta")
         return self.get(url)
 
-    def get_attachment_expand_human(self, attachment_id: T_id):
+    def get_attachment_expand_human(self, attachment_id: T_id) -> dict | None:
         """
         Returns the information for an expandable attachment in human-readable format
         :param attachment_id: int
@@ -340,7 +342,7 @@ def get_attachment_expand_human(self, attachment_id: T_id):
         url = f"{base_url}/{attachment_id}/expand/human"
         return self.get(url)
 
-    def get_attachment_expand_raw(self, attachment_id: T_id):
+    def get_attachment_expand_raw(self, attachment_id: T_id) -> dict | None:
         """
         Returns the information for an expandable attachment in raw format
         :param attachment_id: int
@@ -362,7 +364,7 @@ def get_audit_records(
         filter: str | None = None,
         from_date: str | None = None,
         to_date: str | None = None,
-    ):
+    ) -> dict | None:
         """
         Returns auditing records filtered using provided parameters
         :param offset: the number of record from which search starts
@@ -392,7 +394,7 @@ def get_audit_records(
         url = self.resource_url("auditing/record")
         return self.get(url, params=params) or {}
 
-    def post_audit_record(self, audit_record: dict | str):
+    def post_audit_record(self, audit_record: dict | str) -> dict | None:
         """
         Store a record in Audit Log
         :param audit_record: json with compat https://docs.atlassian.com/jira/REST/schema/audit-record#
@@ -406,7 +408,7 @@ def post_audit_record(self, audit_record: dict | str):
     Reference: https://docs.atlassian.com/software/jira/docs/api/REST/8.5.0/#api/2/avatar
     """
 
-    def get_all_system_avatars(self, avatar_type: str = "user"):
+    def get_all_system_avatars(self, avatar_type: str = "user") -> dict | None:
         """
         Returns all system avatars of the given type.
         :param avatar_type:
@@ -422,11 +424,11 @@ def get_all_system_avatars(self, avatar_type: str = "user"):
     Reference: https://docs.atlassian.com/software/jira/docs/api/REST/8.5.0/#api/2/cluster
     """
 
-    def get_cluster_all_nodes(self):
+    def get_cluster_all_nodes(self) -> dict | None:
         url = self.resource_url("cluster/nodes")
         return self.get(url)
 
-    def delete_cluster_node(self, node_id: T_id):
+    def delete_cluster_node(self, node_id: T_id) -> dict | None:
         """
         Delete the node from the cluster if state of node is OFFLINE
         :param node_id: str
@@ -436,7 +438,7 @@ def delete_cluster_node(self, node_id: T_id):
         url = f"{base_url}/{node_id}"
         return self.delete(url)
 
-    def set_node_to_offline(self, node_id: T_id):
+    def set_node_to_offline(self, node_id: T_id) -> dict | None:
         """
         Change the node's state to offline if the node is reporting as active, but is not alive
         :param node_id: str
@@ -446,14 +448,15 @@ def set_node_to_offline(self, node_id: T_id):
         url = f"{base_url}/{node_id}/offline"
         return self.put(url)
 
-    def get_cluster_alive_nodes(self):
+    def get_cluster_alive_nodes(self) -> list:
         """
         Get cluster nodes where alive = True
         :return: list of node dicts
         """
-        return [_ for _ in self.get_cluster_all_nodes() if _["alive"]]
+        nodes = self.get_cluster_all_nodes()
+        return [_ for _ in nodes.values() if _["alive"]] if nodes else []
 
-    def request_current_index_from_node(self, node_id: T_id):
+    def request_current_index_from_node(self, node_id: T_id) -> dict | None:
         """
         Request current index from node (the request is processed asynchronously)
         :return:
@@ -467,7 +470,7 @@ def request_current_index_from_node(self, node_id: T_id):
     Reference: https://confluence.atlassian.com/support/create-a-support-zip-using-the-rest-api-in-data-center-applications-952054641.html
     """
 
-    def generate_support_zip_on_nodes(self, node_ids: list):
+    def generate_support_zip_on_nodes(self, node_ids: list) -> dict | None:
         """
         Generate a support zip on targeted nodes of a cluster
         :param node_ids: list
@@ -477,7 +480,7 @@ def generate_support_zip_on_nodes(self, node_ids: list):
         url = "/rest/troubleshooting/latest/support-zip/cluster"
         return self.post(url, data=data)
 
-    def check_support_zip_status(self, cluster_task_id: T_id):
+    def check_support_zip_status(self, cluster_task_id: T_id) -> dict | None:
         """
         Check status of support zip creation task
         :param cluster_task_id: str
@@ -486,7 +489,7 @@ def check_support_zip_status(self, cluster_task_id: T_id):
         url = f"/rest/troubleshooting/latest/support-zip/status/cluster/{cluster_task_id}"
         return self.get(url)
 
-    def download_support_zip(self, file_name: str):
+    def download_support_zip(self, file_name: str) -> bytes:
         """
         Download created support zip file
         :param file_name: str
@@ -500,12 +503,12 @@ def download_support_zip(self, file_name: str):
     Reference: https://docs.atlassian.com/software/jira/docs/api/REST/8.5.0/#api/2/cluster/zdu
     """
 
-    def get_cluster_zdu_state(self):
+    def get_cluster_zdu_state(self) -> dict | None:
         url = self.resource_url("cluster/zdu/state")
         return self.get(url)
 
     # Issue Comments
-    def issue_get_comments(self, issue_id: T_id):
+    def issue_get_comments(self, issue_id: T_id) -> dict | None:
         """
         Get Comments on an Issue.
         :param issue_id: Issue ID
@@ -516,7 +519,7 @@ def issue_get_comments(self, issue_id: T_id):
         url = f"{base_url}/{issue_id}/comment"
         return self.get(url)
 
-    def issues_get_comments_by_id(self, *args):
+    def issues_get_comments_by_id(self, *args: int) -> dict | None:
         """
         Get Comments on Multiple Issues
         :param *args: int Issue ID's
@@ -847,7 +850,7 @@ def get_filter(self, filter_id: T_id):
         url = f"{base_url}/{filter_id}"
         return self.get(url)
 
-    def update_filter(self, filter_id: T_id, jql: str, **kwargs):
+    def update_filter(self, filter_id: T_id, jql: str, **kwargs: Any):
         """
         :param filter_id: int
         :param jql: str
@@ -1412,7 +1415,7 @@ def issue_field_value_append(self, issue_id_or_key: str, field: str, value: str,
             params=params,
         )
 
-    def get_issue_labels(self, issue_key: str):
+    def get_issue_labels(self, issue_key: str) -> dict | None:
         """
         Get issue labels.
         :param issue_key:
@@ -1748,7 +1751,7 @@ def scrap_regex_from_issue(self, issue: str, regex: str):
         except HTTPError as e:
             if e.response.status_code == 404:
                 # Raise ApiError as the documented reason is ambiguous
-                log.error("couldn't find issue: ", issue["key"])
+                log.error("couldn't find issue: ", issue)
                 raise ApiNotFoundError(
                     "There is no content with the given issue ud,"
                     "or the calling user does not have permission to view the issue",
@@ -1890,7 +1893,7 @@ def update_issue_remote_link_by_id(
         url = f"{base_url}/{issue_key}/remotelink/{link_id}"
         return self.put(url, data=data)
 
-    def delete_issue_remote_link_by_id(self, issue_key: str, link_id: T_id):
+    def delete_issue_remote_link_by_id(self, issue_key: str, link_id: T_id) -> dict | None:
         """
         Deletes Remote Link on Issue
         :param issue_key: str
@@ -1900,27 +1903,23 @@ def delete_issue_remote_link_by_id(self, issue_key: str, link_id: T_id):
         url = f"{base_url}/{issue_key}/remotelink/{link_id}"
         return self.delete(url)
 
-    def get_issue_transitions(self, issue_key: str):
+    def get_issue_transitions(self, issue_key: str) -> list[dict]:
         if self.advanced_mode:
-            return [
-                {
-                    "name": transition["name"],
-                    "id": int(transition["id"]),
-                    "to": transition["to"]["name"],
-                }
-                for transition in (self.get_issue_transitions_full(issue_key).json() or {}).get("transitions")
-            ]
+            resp = cast(Response, self.get_issue_transitions_full(issue_key))
+            d: dict[str, list] = resp.json() or {}
         else:
-            return [
-                {
-                    "name": transition["name"],
-                    "id": int(transition["id"]),
-                    "to": transition["to"]["name"],
-                }
-                for transition in (self.get_issue_transitions_full(issue_key) or {}).get("transitions")
-            ]
+            d = self.get_issue_transitions_full(issue_key) or {}
+
+        return [
+            {
+                "name": transition["name"],
+                "id": int(transition["id"]),
+                "to": transition["to"]["name"],
+            }
+            for transition in cast(list[dict], d.get("transitions"))
+        ]
 
-    def issue_transition(self, issue_key: str, status: str):
+    def issue_transition(self, issue_key: str, status: str) -> dict | None:
         return self.set_issue_status(issue_key, status)
 
     def set_issue_status(
@@ -1979,7 +1978,7 @@ def get_issue_status(self, issue_key: str):
         fields = [("fields",), ("status",), ("name",)]
         return self._get_response_content(url, fields=fields) or {}
 
-    def get_issue_status_id(self, issue_key: str):
+    def get_issue_status_id(self, issue_key: str) -> str:
         base_url = self.resource_url("issue")
         url = f"{base_url}/{issue_key}?fields=status"
         fields = [("fields",), ("status",), ("id",)]

atlassian/typehints.py

@@ -5,5 +5,3 @@
 _Data: TypeAlias = (
     dict | str
 )
-
-