Skip to content

jira_api

JiraAPI

Source code in src/rich_jira_release_notes/libs/jira/jira_api.py 
 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
class JiraAPI:
    def __init__(self, base_url: str, credentials: JiraCredentialsModel) -> None:
        self.base_url = base_url
        self.credentials = credentials

    @staticmethod
    def _build_field_maps_from_names(
        names: dict | None, selected_fields: list[str]
    ) -> dict[str, str]:
        if not isinstance(names, dict):
            return {}

        return {
            display_name: field_key
            for field_key, display_name in names.items()
            if display_name in selected_fields
        }

    @staticmethod
    def _get_system_field_key(field_name: str) -> str | None:
        return {
            "summary": "summary",
            "description": "description",
        }.get(field_name.lower())

    def _search_field_id(
        self, field_name: str, headers: dict[str, str], auth: HTTPBasicAuth
    ) -> str | None:
        field_search_url = f"{self.base_url}/rest/api/3/field/search"
        params: RequestParams = {"query": field_name, "maxResults": 50}
        response = requests.request(
            "GET",
            field_search_url,
            headers=headers,
            params=params,
            auth=auth,
        )
        response.raise_for_status()
        values = response.json().get("values", [])

        for value in values:
            if value.get("name", "").lower() == field_name.lower():
                return value["id"]
        return None

    def _resolve_field_maps(
        self,
        data: dict,
        fields: list[str],
        headers: dict[str, str],
        auth: HTTPBasicAuth,
    ) -> dict[str, str]:
        field_maps = self._build_field_maps_from_names(data.get("names"), fields)

        missing_fields = [field for field in fields if field not in field_maps]
        for field_name in missing_fields:
            system_key = self._get_system_field_key(field_name)
            if system_key is not None:
                field_maps[field_name] = system_key
                continue

            field_id = self._search_field_id(field_name, headers, auth)
            if field_id is not None:
                field_maps[field_name] = field_id

        return field_maps

    def get_issues(self, jql_query: str, fields: list[str]) -> list:
        """Get issues from Jira utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-search/#api-rest-api-3-search-jql-get

        Args:
            jql_query (str): JQL query to search for issues

        Returns:
            list: List of issues
        """
        url = f"{self.base_url}/rest/api/3/search/jql"
        auth = HTTPBasicAuth(self.credentials.username, self.credentials.token)

        headers = {"Accept": "application/json"}

        query: RequestParams = {
            "jql": jql_query,
            "fields": "*all",
            "fieldsByKeys": "true",
            "expand": "renderedFields,names",
        }

        response = requests.request(
            "GET", url, headers=headers, params=query, auth=auth
        )
        response.raise_for_status()

        data = json.loads(response.text)

        # Resolve Jira internal field names to clear text representation of desired fields
        field_maps = self._resolve_field_maps(data, fields, headers, auth)

        # Extract desired fields from issues
        result = []
        for issue in data["issues"]:
            entry = {
                "id": issue["id"],
                "key": issue["key"],
                "type": issue["fields"]["issuetype"]["name"],
                "fields": {},
            }
            for field_key, field_value in field_maps.items():
                if (
                    field_value in issue["renderedFields"]
                    and issue["renderedFields"][field_value] is not None
                ):
                    entry["fields"][field_key] = JiraField(
                        value=issue["renderedFields"][field_value],
                        type=JiraFieldType.RICH_TEXT,
                    )
                elif (
                    field_value in issue["fields"]
                    and issue["fields"][field_value] is not None
                ):
                    if isinstance(issue["fields"][field_value], list):
                        entry["fields"][field_key] = JiraField(
                            value=[
                                entry["value"] for entry in issue["fields"][field_value]
                            ],
                            type=JiraFieldType.CHECKBOX,
                        )
                    else:
                        entry["fields"][field_key] = JiraField(
                            value=issue["fields"][field_value], type=JiraFieldType.TEXT
                        )
            result.append(entry)
        return [JiraIssue(**issue) for issue in result]

    def download_attachment(self, url: str, output_path: str) -> None:
        """Download an attachment from Jira using its URL utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-attachments#api-group-issue-attachments

        Args:
            url (str): URL of attachment
            output_path (str): Output path to save attachment to
        """
        auth = HTTPBasicAuth(self.credentials.username, self.credentials.token)

        headers = {"Accept": "application/json"}

        response = requests.request("GET", url, headers=headers, auth=auth)

        with open(output_path, "wb") as f:
            f.write(response.content)

download_attachment(url, output_path)

Download an attachment from Jira using its URL utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-attachments#api-group-issue-attachments

Parameters:

Name Type Description Default
url str

URL of attachment

 required
output_path str

Output path to save attachment to

 required
Source code in src/rich_jira_release_notes/libs/jira/jira_api.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def download_attachment(self, url: str, output_path: str) -> None:
    """Download an attachment from Jira using its URL utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-attachments#api-group-issue-attachments

    Args:
        url (str): URL of attachment
        output_path (str): Output path to save attachment to
    """
    auth = HTTPBasicAuth(self.credentials.username, self.credentials.token)

    headers = {"Accept": "application/json"}

    response = requests.request("GET", url, headers=headers, auth=auth)

    with open(output_path, "wb") as f:
        f.write(response.content)

get_issues(jql_query, fields)

Get issues from Jira utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-search/#api-rest-api-3-search-jql-get

Parameters:

Name Type Description Default
jql_query str

JQL query to search for issues

 required

Returns:

Name Type Description
list list

List of issues
Source code in src/rich_jira_release_notes/libs/jira/jira_api.py 
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
def get_issues(self, jql_query: str, fields: list[str]) -> list:
    """Get issues from Jira utilizing https://developer.atlassian.com/cloud/jira/platform/rest/v3/api-group-issue-search/#api-rest-api-3-search-jql-get

    Args:
        jql_query (str): JQL query to search for issues

    Returns:
        list: List of issues
    """
    url = f"{self.base_url}/rest/api/3/search/jql"
    auth = HTTPBasicAuth(self.credentials.username, self.credentials.token)

    headers = {"Accept": "application/json"}

    query: RequestParams = {
        "jql": jql_query,
        "fields": "*all",
        "fieldsByKeys": "true",
        "expand": "renderedFields,names",
    }

    response = requests.request(
        "GET", url, headers=headers, params=query, auth=auth
    )
    response.raise_for_status()

    data = json.loads(response.text)

    # Resolve Jira internal field names to clear text representation of desired fields
    field_maps = self._resolve_field_maps(data, fields, headers, auth)

    # Extract desired fields from issues
    result = []
    for issue in data["issues"]:
        entry = {
            "id": issue["id"],
            "key": issue["key"],
            "type": issue["fields"]["issuetype"]["name"],
            "fields": {},
        }
        for field_key, field_value in field_maps.items():
            if (
                field_value in issue["renderedFields"]
                and issue["renderedFields"][field_value] is not None
            ):
                entry["fields"][field_key] = JiraField(
                    value=issue["renderedFields"][field_value],
                    type=JiraFieldType.RICH_TEXT,
                )
            elif (
                field_value in issue["fields"]
                and issue["fields"][field_value] is not None
            ):
                if isinstance(issue["fields"][field_value], list):
                    entry["fields"][field_key] = JiraField(
                        value=[
                            entry["value"] for entry in issue["fields"][field_value]
                        ],
                        type=JiraFieldType.CHECKBOX,
                    )
                else:
                    entry["fields"][field_key] = JiraField(
                        value=issue["fields"][field_value], type=JiraFieldType.TEXT
                    )
        result.append(entry)
    return [JiraIssue(**issue) for issue in result]