From 1de348d3263282cba5ae0732f77c9df4613a062c Mon Sep 17 00:00:00 2001 From: LuisFigueroaG Date: Sun, 18 Oct 2026 21:15:05 -0300 Subject: [PATCH] [Confluence] ConfluenceV2: add cql() The Server and legacy Cloud clients have cql(), and the CQL section of the docs uses it, but ConfluenceV2 only had iter_cql() and cql_all(), so confluence.cql(...) failed with "'ConfluenceCloud' object has no attribute 'cql'". Confluence() returns the same client for API gateway URLs (scoped tokens), so those callers hit the error too. cql() returns one page of results from the v1 rest/api/search endpoint, which search() and iter_cql() already use, and accepts limit, expand and the cursor from a previous response. --- atlassian/confluence/cloud/cloud.py | 33 +++++++++++++++++++++++++++++ docs/confluence.rst | 10 ++++++--- tests/test_confluence_v2.py | 23 ++++++++++++++++++++ 3 files changed, 63 insertions(+), 3 deletions(-) diff --git a/atlassian/confluence/cloud/cloud.py b/atlassian/confluence/cloud/cloud.py index 4696155fb..1ce8dc96b 100644 --- a/atlassian/confluence/cloud/cloud.py +++ b/atlassian/confluence/cloud/cloud.py @@ -628,6 +628,39 @@ def search( log.error(f"Failed to perform search: {e}") raise + def cql( + self, + cql: str, + limit: int = 25, + expand: Optional[str] = None, + cursor: Optional[str] = None, + ) -> Dict[str, Any]: + """ + Return one page of CQL search results. + + Confluence Cloud offers CQL search only on the v1 API, so this call is made + against the v1 search endpoint. Use :meth:`iter_cql` to follow the cursor + through every page. + + Args: + cql: Confluence Query Language (CQL) expression + limit: (optional) Maximum number of results to return per request. Default: 25 + expand: (optional) Properties to expand on each result, e.g. 'content.body.storage' + cursor: (optional) Cursor from the ``_links.next`` of a previous response + + Returns: + Dictionary with search results + + Raises: + HTTPError: If the API call fails + """ + params: Dict[str, Any] = {"cql": cql, "limit": limit} + if expand: + params["expand"] = expand + if cursor: + params["cursor"] = cursor + return self.get(self.get_endpoint("search"), params=params) + def iter_cql( self, cql: Optional[str] = None, diff --git a/docs/confluence.rst b/docs/confluence.rst index bb32f0972..d404b35f2 100644 --- a/docs/confluence.rst +++ b/docs/confluence.rst @@ -967,12 +967,16 @@ CQL ``iter_cql()`` and ``cql_all()`` work for Cloud and Server/Data Center. Cloud follows its cursor link, while Server/Data Center retains its offset-based -pagination semantics. The Cloud V2 client (``ConfluenceV2``) exposes the same -pair. CQL search has no v2 endpoint, so it is served by ``rest/api/search`` -there as well: +pagination semantics. The Cloud V2 client (``ConfluenceV2``) exposes ``cql()`` +and the same pair. CQL search has no v2 endpoint, so it is served by +``rest/api/search`` there as well: .. code-block:: python + response = confluence_v2.cql( + 'space.id = "123456" and type = page', limit=25, expand="content.body.storage" + ) + for result in confluence_v2.iter_cql('space.id = "123456" and type = page', limit=25): process(result) diff --git a/tests/test_confluence_v2.py b/tests/test_confluence_v2.py index f0360ef1c..bb7c3fcc5 100644 --- a/tests/test_confluence_v2.py +++ b/tests/test_confluence_v2.py @@ -512,6 +512,29 @@ def test_search_with_cql(self, mock_get): ) self.assertEqual(response, mock_response) + @patch("atlassian.confluence.cloud.ConfluenceCloud.get") + def test_cql(self, mock_get): + mock_response = {"results": [{"content": {"id": "123"}}], "_links": {}} + mock_get.return_value = mock_response + + response = self.confluence_v2.cql('space = "TEAM" and type = page') + + mock_get.assert_called_once_with( + "rest/api/search", params={"cql": 'space = "TEAM" and type = page', "limit": 25} + ) + self.assertEqual(response, mock_response) + + @patch("atlassian.confluence.cloud.ConfluenceCloud.get") + def test_cql_with_expand_and_cursor(self, mock_get): + mock_get.return_value = {"results": []} + + self.confluence_v2.cql("type = page", limit=10, expand="content.body.storage", cursor="abc") + + mock_get.assert_called_once_with( + "rest/api/search", + params={"cql": "type = page", "limit": 10, "expand": "content.body.storage", "cursor": "abc"}, + ) + def test_search_no_query_or_cql(self): # Test missing both query and cql with self.assertRaises(ValueError):