[Confluence] ConfluenceV2: add cql()

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,

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)
 

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):