diff --git a/.readthedocs.yml b/.readthedocs.yml index 58cb9081e..ed8caa57a 100644 --- a/.readthedocs.yml +++ b/.readthedocs.yml @@ -10,6 +10,9 @@ build: apt_packages: - libkrb5-dev +sphinx: + configuration: docs/conf.py + python: install: - requirements: constraints.txt diff --git a/jira/client.py b/jira/client.py index 36dc2fea9..ad16d4bfb 100644 --- a/jira/client.py +++ b/jira/client.py @@ -225,6 +225,7 @@ def __init__( _maxResults: int = 0, _total: int | None = None, _isLast: bool | None = None, + _nextPageToken: str | None = None, ) -> None: """Results List. @@ -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 `_ """ if iterable is not None: @@ -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 @@ -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 @@ -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], @@ -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 @@ -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.