Skip to content

Add enhanced_search_issues API for Jira Cloud using new search/jql endpoint #2326

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
3 changes: 3 additions & 0 deletions .readthedocs.yml
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ build:
apt_packages:
- libkrb5-dev

sphinx:
configuration: docs/conf.py

python:
install:
- requirements: constraints.txt
Expand Down
193 changes: 193 additions & 0 deletions jira/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,7 @@ def __init__(
_maxResults: int = 0,
_total: int | None = None,
_isLast: bool | None = None,
_nextPageToken: str | None = None,
) -> None:
"""Results List.

Expand All @@ -234,6 +235,7 @@ def __init__(
_maxResults (int): Max results per page. Defaults to 0.
_total (Optional[int]): Total results from query. Defaults to 0.
_isLast (Optional[bool]): True to mark this page is the last page? (Default: ``None``).
_nextPageToken (Optional[str]): Token for fetching the next page of results. Defaults to None.
see `The official API docs <https://developer.atlassian.com/cloud/jira/platform/rest/v3/intro/#expansion:~:text=for%20all%20operations.-,isLast,-indicates%20whether%20the>`_
"""
if iterable is not None:
Expand All @@ -249,6 +251,7 @@ def __init__(

self.iterable: list[ResourceType] = list(iterable) if iterable else []
self.current = self.startAt
self.nextPageToken = _nextPageToken

def __next__(self) -> ResourceType: # type:ignore[misc]
self.current += 1
Expand Down Expand Up @@ -816,6 +819,7 @@ def json_params() -> dict[str, Any]:
resource = self._get_json(
request_path, params=page_params, base=base, use_post=use_post
)

next_items_page = self._get_items_from_page(item_type, items_key, resource)
items = next_items_page

Expand Down Expand Up @@ -908,6 +912,58 @@ def json_params() -> dict[str, Any]:
[item_type(self._options, self._session, resource)], 0, 1, 1, True
)

@cloud_api
def _fetch_pages_searchToken(
self,
item_type: type[ResourceType],
items_key: str | None,
request_path: str,
maxResults: int = 50,
params: dict[str, Any] | None = None,
base: str = JIRA_BASE_URL,
use_post: bool = False,
) -> ResultList[ResourceType]:
"""Fetch from a paginated API endpoint using `nextPageToken`.

Args:
item_type (Type[Resource]): Type of single item. Returns a `ResultList` of such items.
items_key (Optional[str]): Path to the items in JSON returned from the server.
request_path (str): Path in the request URL.
maxResults (int): Maximum number of items to return per page. (Default: 50)
params (Dict[str, Any]): Parameters to be sent with the request.
base (str): Base URL for the requests.
use_post (bool): Whether to use POST instead of GET.

Returns:
ResultList: List of fetched items.
"""
DEFAULT_BATCH = 100 # Max batch size per request
fetch_all = maxResults in (0, False) # If False/0, fetch everything

page_params = (params or {}).copy() # Ensure params isn't modified
page_params["maxResults"] = DEFAULT_BATCH if fetch_all else maxResults

# Use caller-provided nextPageToken if present
nextPageToken: str | None = page_params.get("nextPageToken")
items: list[ResourceType] = []

while True:
# Ensure nextPageToken is set in params if it exists
if nextPageToken:
page_params["nextPageToken"] = nextPageToken
else:
page_params.pop("nextPageToken", None)

response = self._get_json(
request_path, params=page_params, base=base, use_post=use_post
)
items.extend(self._get_items_from_page(item_type, items_key, response))
nextPageToken = response.get("nextPageToken")
if not fetch_all or not nextPageToken:
break

return ResultList(items, _nextPageToken=nextPageToken)

def _get_items_from_page(
self,
item_type: type[ResourceType],
Expand Down Expand Up @@ -3547,6 +3603,22 @@ def search_issues(
elif fields is None:
fields = ["*all"]

if self._is_cloud:
if startAt == 0:
return self.enhanced_search_issues(
jql_str=jql_str,
maxResults=maxResults,
fields=fields,
expand=expand,
properties=properties,
json_result=json_result,
use_post=use_post,
)
else:
raise JIRAError(
"The `search` API is deprecated in Jira Cloud. Use `enhanced_search_issues` method instead."
)

# this will translate JQL field names to REST API Name
# most people do know the JQL names so this will help them use the API easier
untranslate = {} # use to add friendly aliases when we get the results back
Expand Down Expand Up @@ -3600,6 +3672,127 @@ def search_issues(

return issues

@cloud_api
def enhanced_search_issues(
self,
jql_str: str,
nextPageToken: str | None = None,
maxResults: int = 50,
fields: str | list[str] | None = "*all",
expand: str | None = None,
reconcileIssues: list[int] | None = None,
properties: str | None = None,
*,
json_result: bool = False,
use_post: bool = False,
) -> dict[str, Any] | ResultList[Issue]:
"""Get a :class:`~jira.client.ResultList` of issue Resources matching a JQL search string.

Args:
jql_str (str): The JQL search string.
nextPageToken (Optional[str]): Token for paginated results.
maxResults (int): Maximum number of issues to return.
Total number of results is available in the ``total`` attribute of the returned :class:`ResultList`.
If maxResults evaluates to False, it will try to get all issues in batches. (Default: ``50``)
fields (Optional[Union[str, List[str]]]): comma-separated string or list of issue fields to include in the results.
Default is to include all fields.
expand (Optional[str]): extra information to fetch inside each resource.
reconcileIssues (Optional[List[int]]): List of issue IDs to reconcile.
properties (Optional[str]): extra properties to fetch inside each result
json_result (bool): True to return a JSON response. When set to False a :class:`ResultList` will be returned. (Default: ``False``)
use_post (bool): True to use POST endpoint to fetch issues.

Returns:
Union[Dict, ResultList]: JSON Dict if ``json_result=True``, otherwise a `ResultList`.
"""
if isinstance(fields, str):
fields = fields.split(",")
elif fields is None:
fields = ["*all"]

# this will translate JQL field names to REST API Name
# most people do know the JQL names so this will help them use the API easier
untranslate = {} # use to add friendly aliases when we get the results back
if self._fields_cache:
for i, field in enumerate(fields):
if field in self._fields_cache:
untranslate[self._fields_cache[field]] = fields[i]
fields[i] = self._fields_cache[field]

search_params: dict[str, Any] = {
"jql": jql_str,
"fields": fields,
"expand": expand,
"properties": properties,
"reconcileIssues": reconcileIssues or [],
}
if nextPageToken:
search_params["nextPageToken"] = nextPageToken

if json_result:
if not maxResults:
warnings.warn(
"All issues cannot be fetched at once, when json_result parameter is set",
Warning,
)
else:
search_params["maxResults"] = maxResults
r_json: dict[str, Any] = self._get_json(
"search/jql", params=search_params, use_post=use_post
)
return r_json

issues = self._fetch_pages_searchToken(
item_type=Issue,
items_key="issues",
request_path="search/jql",
maxResults=maxResults,
params=search_params,
use_post=use_post,
)

if untranslate:
iss: Issue
for iss in issues:
for k, v in untranslate.items():
if iss.raw:
if k in iss.raw.get("fields", {}):
iss.raw["fields"][v] = iss.raw["fields"][k]

return issues

@cloud_api
def approximate_issue_count(
self,
jql_str: str,
*,
json_result: bool = False,
) -> int | dict[str, Any]:
"""Get an approximate count of issues matching a JQL search string.

Args:
jql_str (str): The JQL search string.
json_result (bool): If True, returns the full JSON response. Defaults to False.

Returns:
int | dict[str, Any]: The issue count if json_result is False, else the raw JSON response.
"""
if not self._is_cloud:
raise ValueError(
"The 'approximate-count' API is only available for Jira Cloud."
)

search_params = {"jql": jql_str}

response_json: dict[str, Any] = self._get_json(
"search/approximate-count", params=search_params, use_post=True
)

if json_result:
return response_json

return response_json.get("count", 0)

# Security levels
def security_level(self, id: str) -> SecurityLevel:
"""Get a security level Resource.
Expand Down
Loading