Skip to content

datarobot_genai.drtools.core.clients.jira

jira

JiraClient

Client for interacting with Jira API using OAuth access token.

At the moment of creating this client, official Jira SDK is not supporting async.

Source code in datarobot_genai/drtools/core/clients/jira.py 
 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
class JiraClient:
    """
    Client for interacting with Jira API using OAuth access token.

    At the moment of creating this client, official Jira SDK is not supporting async.
    """

    def __init__(self, auth: str | AtlassianAuth) -> None:
        """
        Initialize Jira client with OAuth Bearer or API token Basic auth.

        Args:
            auth: :class:`AtlassianAuth` or a raw OAuth access token string (Bearer)
        """
        self._auth = auth if isinstance(auth, AtlassianAuth) else AtlassianAuth.oauth_bearer(auth)
        self._client = httpx.AsyncClient(
            headers=self._auth.http_headers(),
            timeout=30.0,
        )
        self._cloud_id: str | None = self._auth.cloud_id or None

    async def _get_cloud_id(self) -> str:
        """
        Get the cloud ID for the authenticated Atlassian Jira instance.

        According to Atlassian OAuth 2.0 documentation, API calls should use:
        https://api.atlassian.com/ex/jira/{cloudId}/rest/api/3/...

        Returns
        -------
            Cloud ID string

        Raises
        ------
            ValueError: If cloud ID cannot be retrieved
        """
        if self._cloud_id:
            return self._cloud_id

        self._cloud_id = await get_atlassian_cloud_id(
            self._client, service_type="jira", auth=self._auth
        )
        return self._cloud_id

    async def _get_full_url(self, path: str) -> str:
        """Return URL for Jira API."""
        cloud_id = await self._get_cloud_id()
        return f"{ATLASSIAN_API_BASE}/ex/jira/{cloud_id}/rest/api/3/{path}"

    async def search_jira_issues(self, jql_query: str, max_results: int) -> list[Issue]:
        """
        Search Jira issues using JQL (Jira Query Language).

        Args:
            jql_query: JQL Query
            max_results: Maximum number of issues to return

        Returns
        -------
            List of Jira issues

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url("search/jql")
        response = await self._client.post(
            url,
            json={
                "jql": jql_query,
                "fields": list(RESPONSE_JIRA_ISSUE_FIELDS),
                "maxResults": max_results,
            },
        )

        response.raise_for_status()
        raw_issues = response.json().get("issues", [])
        issues = [Issue(**issue) for issue in raw_issues]
        return issues

    async def get_jira_issue(self, issue_key: str) -> Issue:
        """
        Get a Jira issue by its key.

        Args:
            issue_key: The key of the Jira issue, e.g., 'PROJ-123'

        Returns
        -------
            Jira issue

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url(f"issue/{issue_key}")
        response = await self._client.get(url, params={"fields": RESPONSE_JIRA_ISSUE_FIELDS_STR})

        if response.status_code == HTTPStatus.NOT_FOUND:
            raise ValueError(f"{issue_key} not found")

        response.raise_for_status()
        issue = Issue(**response.json())
        return issue

    async def get_jira_issue_types(self, project_key: str) -> dict[str, str]:
        """
        Get Jira issue types possible for given project.

        Args:
            project_key: The key of the Jira project, e.g., 'PROJ'

        Returns
        -------
            Dictionary where key is the issue type name and value is the issue type ID

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url(f"issue/createmeta/{project_key}/issuetypes")
        response = await self._client.get(url)

        response.raise_for_status()
        jsoned = response.json()
        issue_types = {
            issue_type["untranslatedName"]: issue_type["id"]
            for issue_type in jsoned.get("issueTypes", [])
        }
        return issue_types

    async def create_jira_issue(
        self, project_key: str, summary: str, issue_type_id: str, description: str | None
    ) -> str:
        """
        Create Jira issue.

        Args:
            project_key: The key of the Jira project, e.g., 'PROJ'
            summary: Summary of Jira issue (title), e.g., 'Fix bug abc'
            issue_type_id: ID type of Jira issue, e.g., "1"
            description: Optional description of Jira issue

        Returns
        -------
            Jira issue key

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url("issue")
        payload = {
            "fields": {
                "project": {"key": project_key},
                "summary": summary,
                "issuetype": {"id": issue_type_id},
            }
        }

        if description:
            payload["fields"]["description"] = {
                "content": [
                    {"content": [{"text": description, "type": "text"}], "type": "paragraph"}
                ],
                "type": "doc",
                "version": 1,
            }

        response = await self._client.post(url, json=payload)

        response.raise_for_status()
        jsoned = response.json()
        return jsoned["key"]

    async def update_jira_issue(self, issue_key: str, fields: dict[str, Any]) -> list[str]:
        """
        Update Jira issue.

        Args:
            issue_key: The key of the Jira issue, e.g., 'PROJ-123'
            fields: A dictionary of field names and their new values
                e.g., {'description': 'New content'}

        Returns
        -------
            List of updated fields

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url(f"issue/{issue_key}")
        payload = {"fields": fields}

        response = await self._client.put(url, json=payload)

        response.raise_for_status()
        return list(fields.keys())

    async def get_available_jira_transitions(self, issue_key: str) -> dict[str, str]:
        """
        Get Available Jira Transitions.

        Args:
            issue_key: The key of the Jira issue, e.g., 'PROJ-123'

        Returns
        -------
            Dictionary where key is the transition name and value is the transition ID

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url(f"issue/{issue_key}/transitions")
        response = await self._client.get(url)
        response.raise_for_status()
        jsoned = response.json()
        transitions = {
            transition["name"]: transition["id"] for transition in jsoned.get("transitions", [])
        }
        return transitions

    async def transition_jira_issue(self, issue_key: str, transition_id: str) -> None:
        """
        Transition Jira issue.

        Args:
            issue_key: The key of the Jira issue, e.g., 'PROJ-123'
            transition_id: Id of target transitionm e.g. '123'.
                Can be obtained from `get_available_jira_transitions`.

        Returns
        -------
            Nothing

        Raises
        ------
            httpx.HTTPStatusError: If the API request fails
        """
        url = await self._get_full_url(f"issue/{issue_key}")
        payload = {"transition": {"id": transition_id}}

        response = await self._client.post(url, json=payload)

        response.raise_for_status()

    async def __aenter__(self) -> "JiraClient":
        """Async context manager entry."""
        return self

    async def __aexit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
    ) -> None:
        """Async context manager exit."""
        await self._client.aclose()

__init__ 

__init__(auth: str | AtlassianAuth) -> None

Initialize Jira client with OAuth Bearer or API token Basic auth.

Parameters:

Name Type Description Default
auth str | AtlassianAuth

:class:AtlassianAuth or a raw OAuth access token string (Bearer)

 required
Source code in datarobot_genai/drtools/core/clients/jira.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __init__(self, auth: str | AtlassianAuth) -> None:
    """
    Initialize Jira client with OAuth Bearer or API token Basic auth.

    Args:
        auth: :class:`AtlassianAuth` or a raw OAuth access token string (Bearer)
    """
    self._auth = auth if isinstance(auth, AtlassianAuth) else AtlassianAuth.oauth_bearer(auth)
    self._client = httpx.AsyncClient(
        headers=self._auth.http_headers(),
        timeout=30.0,
    )
    self._cloud_id: str | None = self._auth.cloud_id or None

search_jira_issues async 

search_jira_issues(jql_query: str, max_results: int) -> list[Issue]

Search Jira issues using JQL (Jira Query Language).

Parameters:

Name Type Description Default
jql_query str

JQL Query

 required
max_results int

Maximum number of issues to return

 required
Returns
List of Jira issues
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
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
async def search_jira_issues(self, jql_query: str, max_results: int) -> list[Issue]:
    """
    Search Jira issues using JQL (Jira Query Language).

    Args:
        jql_query: JQL Query
        max_results: Maximum number of issues to return

    Returns
    -------
        List of Jira issues

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url("search/jql")
    response = await self._client.post(
        url,
        json={
            "jql": jql_query,
            "fields": list(RESPONSE_JIRA_ISSUE_FIELDS),
            "maxResults": max_results,
        },
    )

    response.raise_for_status()
    raw_issues = response.json().get("issues", [])
    issues = [Issue(**issue) for issue in raw_issues]
    return issues

get_jira_issue async 

get_jira_issue(issue_key: str) -> Issue

Get a Jira issue by its key.

Parameters:

Name Type Description Default
issue_key str

The key of the Jira issue, e.g., 'PROJ-123'

 required
Returns
Jira issue
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async def get_jira_issue(self, issue_key: str) -> Issue:
    """
    Get a Jira issue by its key.

    Args:
        issue_key: The key of the Jira issue, e.g., 'PROJ-123'

    Returns
    -------
        Jira issue

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url(f"issue/{issue_key}")
    response = await self._client.get(url, params={"fields": RESPONSE_JIRA_ISSUE_FIELDS_STR})

    if response.status_code == HTTPStatus.NOT_FOUND:
        raise ValueError(f"{issue_key} not found")

    response.raise_for_status()
    issue = Issue(**response.json())
    return issue

get_jira_issue_types async 

get_jira_issue_types(project_key: str) -> dict[str, str]

Get Jira issue types possible for given project.

Parameters:

Name Type Description Default
project_key str

The key of the Jira project, e.g., 'PROJ'

 required
Returns
Dictionary where key is the issue type name and value is the issue type ID
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
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
async def get_jira_issue_types(self, project_key: str) -> dict[str, str]:
    """
    Get Jira issue types possible for given project.

    Args:
        project_key: The key of the Jira project, e.g., 'PROJ'

    Returns
    -------
        Dictionary where key is the issue type name and value is the issue type ID

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url(f"issue/createmeta/{project_key}/issuetypes")
    response = await self._client.get(url)

    response.raise_for_status()
    jsoned = response.json()
    issue_types = {
        issue_type["untranslatedName"]: issue_type["id"]
        for issue_type in jsoned.get("issueTypes", [])
    }
    return issue_types

create_jira_issue async 

create_jira_issue(project_key: str, summary: str, issue_type_id: str, description: str | None) -> str

Create Jira issue.

Parameters:

Name Type Description Default
project_key str

The key of the Jira project, e.g., 'PROJ'

 required
summary str

Summary of Jira issue (title), e.g., 'Fix bug abc'

 required
issue_type_id str

ID type of Jira issue, e.g., "1"

 required
description str | None

Optional description of Jira issue

 required
Returns
Jira issue key
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
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
async def create_jira_issue(
    self, project_key: str, summary: str, issue_type_id: str, description: str | None
) -> str:
    """
    Create Jira issue.

    Args:
        project_key: The key of the Jira project, e.g., 'PROJ'
        summary: Summary of Jira issue (title), e.g., 'Fix bug abc'
        issue_type_id: ID type of Jira issue, e.g., "1"
        description: Optional description of Jira issue

    Returns
    -------
        Jira issue key

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url("issue")
    payload = {
        "fields": {
            "project": {"key": project_key},
            "summary": summary,
            "issuetype": {"id": issue_type_id},
        }
    }

    if description:
        payload["fields"]["description"] = {
            "content": [
                {"content": [{"text": description, "type": "text"}], "type": "paragraph"}
            ],
            "type": "doc",
            "version": 1,
        }

    response = await self._client.post(url, json=payload)

    response.raise_for_status()
    jsoned = response.json()
    return jsoned["key"]

update_jira_issue async 

update_jira_issue(issue_key: str, fields: dict[str, Any]) -> list[str]

Update Jira issue.

Parameters:

Name Type Description Default
issue_key str

The key of the Jira issue, e.g., 'PROJ-123'

 required
fields dict[str, Any]

A dictionary of field names and their new values e.g., {'description': 'New content'}

 required
Returns
List of updated fields
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
async def update_jira_issue(self, issue_key: str, fields: dict[str, Any]) -> list[str]:
    """
    Update Jira issue.

    Args:
        issue_key: The key of the Jira issue, e.g., 'PROJ-123'
        fields: A dictionary of field names and their new values
            e.g., {'description': 'New content'}

    Returns
    -------
        List of updated fields

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url(f"issue/{issue_key}")
    payload = {"fields": fields}

    response = await self._client.put(url, json=payload)

    response.raise_for_status()
    return list(fields.keys())

get_available_jira_transitions async 

get_available_jira_transitions(issue_key: str) -> dict[str, str]

Get Available Jira Transitions.

Parameters:

Name Type Description Default
issue_key str

The key of the Jira issue, e.g., 'PROJ-123'

 required
Returns
Dictionary where key is the transition name and value is the transition ID
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
async def get_available_jira_transitions(self, issue_key: str) -> dict[str, str]:
    """
    Get Available Jira Transitions.

    Args:
        issue_key: The key of the Jira issue, e.g., 'PROJ-123'

    Returns
    -------
        Dictionary where key is the transition name and value is the transition ID

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url(f"issue/{issue_key}/transitions")
    response = await self._client.get(url)
    response.raise_for_status()
    jsoned = response.json()
    transitions = {
        transition["name"]: transition["id"] for transition in jsoned.get("transitions", [])
    }
    return transitions

transition_jira_issue async 

transition_jira_issue(issue_key: str, transition_id: str) -> None

Transition Jira issue.

Parameters:

Name Type Description Default
issue_key str

The key of the Jira issue, e.g., 'PROJ-123'

 required
transition_id str

Id of target transitionm e.g. '123'. Can be obtained from get_available_jira_transitions.

 required
Returns
Nothing
Raises
httpx.HTTPStatusError: If the API request fails
Source code in datarobot_genai/drtools/core/clients/jira.py 
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
async def transition_jira_issue(self, issue_key: str, transition_id: str) -> None:
    """
    Transition Jira issue.

    Args:
        issue_key: The key of the Jira issue, e.g., 'PROJ-123'
        transition_id: Id of target transitionm e.g. '123'.
            Can be obtained from `get_available_jira_transitions`.

    Returns
    -------
        Nothing

    Raises
    ------
        httpx.HTTPStatusError: If the API request fails
    """
    url = await self._get_full_url(f"issue/{issue_key}")
    payload = {"transition": {"id": transition_id}}

    response = await self._client.post(url, json=payload)

    response.raise_for_status()

__aenter__ async 

__aenter__() -> JiraClient

Async context manager entry.

Source code in datarobot_genai/drtools/core/clients/jira.py 
333
334
335
async def __aenter__(self) -> "JiraClient":
    """Async context manager entry."""
    return self

__aexit__ async 

__aexit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None

Async context manager exit.

Source code in datarobot_genai/drtools/core/clients/jira.py 
337
338
339
340
341
async def __aexit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
) -> None:
    """Async context manager exit."""
    await self._client.aclose()