[Confluence] ConfluenceCloud: add create_page and update_page

atlassian/confluence/base.py

@@ -49,6 +49,20 @@ def _sub_url(self, url):
         """
         return self.url_joiner(self.url, url)
 
+    @staticmethod
+    def _create_body(body, representation):
+        if representation not in [
+            "atlas_doc_format",
+            "editor",
+            "export_view",
+            "view",
+            "storage",
+            "wiki",
+        ]:
+            raise ValueError("Wrong value for representation, it should be either wiki or storage")
+
+        return {representation: {"value": body, "representation": representation}}
+
     def get_page_id_by_url(self, page_url):
         """Resolve a Confluence page URL to its page ID.
 

atlassian/confluence/cloud/__init__.py

@@ -7,6 +7,7 @@
 from urllib.parse import quote
 from .base import ConfluenceCloudBase
 from requests import HTTPError
+from atlassian import utils
 from atlassian.errors import (
     ApiError,
     ApiNotFoundError,
@@ -322,6 +323,98 @@ def get_page_by_id(self, page_id, **kwargs):
         """Get a page (or other content) by ID from the Cloud V1 content API."""
         return self.get(f"rest/api/content/{page_id}", **kwargs)
 
+    def create_page(
+        self,
+        space,
+        title,
+        body,
+        parent_id=None,
+        type="page",
+        representation="storage",
+        editor=None,
+        full_width=False,
+        status="current",
+        version_comment=None,
+    ):
+        """Create a page through the Cloud V1 content API.
+
+        Takes the same arguments as the Server ``create_page``, so ``space``
+        is the space key. ``ConfluenceV2.create_page`` creates pages by space
+        ID through the V2 API instead.
+        """
+        width = "full-width" if full_width else "fixed-width"
+        data = {
+            "type": type,
+            "title": title,
+            "status": status,
+            "space": {"key": space},
+            "body": self._create_body(body, representation),
+            "metadata": {
+                "properties": {
+                    "content-appearance-draft": {"value": width},
+                    "content-appearance-published": {"value": width},
+                }
+            },
+        }
+        if parent_id:
+            data["ancestors"] = [{"type": type, "id": parent_id}]
+        if version_comment is not None:
+            data["version"] = {"message": version_comment}
+        if editor in ("v1", "v2"):
+            data["metadata"]["properties"]["editor"] = {"value": editor}
+        return self.post("rest/api/content", data=data)
+
+    def update_page(
+        self,
+        page_id,
+        title,
+        body=None,
+        parent_id=None,
+        type="page",
+        representation="storage",
+        minor_edit=False,
+        version_comment=None,
+        always_update=False,
+        full_width=False,
+    ):
+        """Update a page through the Cloud V1 content API.
+
+        Takes the same arguments as the Server ``update_page``. The next
+        version number is read from the page. Unless ``always_update`` is set,
+        the page is returned without an update when its title and storage
+        body already match.
+        """
+        page = self.get_page_by_id(page_id, params={"expand": "body.storage,version"})
+        if not always_update and body is not None and page.get("title") == title:
+            current_body = ((page.get("body") or {}).get("storage") or {}).get("value")
+            if (
+                current_body is not None
+                and utils.symbol_normalizer(current_body).strip().lower() == body.strip().lower()
+            ):
+                log.info("Content of %s is exactly the same", page_id)
+                return page
+
+        width = "full-width" if full_width else "fixed-width"
+        data = {
+            "id": page_id,
+            "type": type,
+            "title": title,
+            "version": {"number": page["version"]["number"] + 1, "minorEdit": minor_edit},
+            "metadata": {
+                "properties": {
+                    "content-appearance-draft": {"value": width},
+                    "content-appearance-published": {"value": width},
+                }
+            },
+        }
+        if body is not None:
+            data["body"] = self._create_body(body, representation)
+        if parent_id:
+            data["ancestors"] = [{"type": "page", "id": parent_id}]
+        if version_comment:
+            data["version"]["message"] = version_comment
+        return self.put(f"rest/api/content/{page_id}", data=data, params={"status": "current"})
+
     def get_page_child_count(self, page_id, type="page"):
         """Return the number of direct children of ``type`` without listing them."""
         page = self.get_page_by_id(page_id, expand=f"children.{type}")

atlassian/confluence/server/__init__.py

@@ -144,20 +144,6 @@ def request(
             allow_redirects=allow_redirects,
         )
 
-    @staticmethod
-    def _create_body(body, representation):
-        if representation not in [
-            "atlas_doc_format",
-            "editor",
-            "export_view",
-            "view",
-            "storage",
-            "wiki",
-        ]:
-            raise ValueError("Wrong value for representation, it should be either wiki or storage")
-
-        return {representation: {"value": body, "representation": representation}}
-
     # Content Management
     def get_content(self, content_id, **kwargs):
         """Get content by ID."""

docs/confluence.rst

@@ -291,6 +291,8 @@ Page actions
     confluence.update_page(page_id, title, body, parent_id=None, type='page', representation='storage', minor_edit=False, full_width=False)
 
     # Server/Data Center updates use the configured REST API root and send a JSON body.
+    # On Cloud, ``ConfluenceCloud`` (and ``Confluence`` with a Cloud URL) take the
+    # same create_page/update_page arguments and use ``/wiki/rest/api/content``.
 
     # Get every contributor for a collaboratively edited page revision.
     contributors = confluence.get_page_version_contributors(page_id, version_number)

tests/confluence/test_confluence_cloud.py

@@ -7,7 +7,7 @@
 import logging
 from unittest.mock import patch
 
-from atlassian.confluence import ConfluenceCloud
+from atlassian.confluence import Confluence, ConfluenceCloud
 from atlassian.confluence.cloud import Cloud
 from atlassian.confluence.cloud.cloud import ConfluenceCloud as ConfluenceCloudV2
 from atlassian.errors import ApiError
@@ -256,6 +256,78 @@ def test_update_content(self, mock_put, confluence_cloud):
         mock_put.assert_called_once_with("rest/api/content/123", data=content_data, **{})
         assert result == {"id": "123", "title": "Updated Page"}
 
+    @patch.object(ConfluenceCloud, "post")
+    def test_create_page(self, mock_post, confluence_cloud):
+        mock_post.return_value = {"id": "456", "title": "New Page"}
+
+        result = confluence_cloud.create_page("TEST", "New Page", "<p>Hello</p>", parent_id="123")
+
+        mock_post.assert_called_once_with(
+            "rest/api/content",
+            data={
+                "type": "page",
+                "title": "New Page",
+                "status": "current",
+                "space": {"key": "TEST"},
+                "body": {"storage": {"value": "<p>Hello</p>", "representation": "storage"}},
+                "metadata": {
+                    "properties": {
+                        "content-appearance-draft": {"value": "fixed-width"},
+                        "content-appearance-published": {"value": "fixed-width"},
+                    }
+                },
+                "ancestors": [{"type": "page", "id": "123"}],
+            },
+        )
+        assert result == {"id": "456", "title": "New Page"}
+
+    @patch.object(ConfluenceCloud, "put")
+    @patch.object(ConfluenceCloud, "get")
+    def test_update_page_through_confluence_with_cloud_url(self, mock_get, mock_put):
+        confluence = Confluence(url="https://test.atlassian.net", username="user", password="token")
+        mock_get.return_value = {
+            "id": "123",
+            "title": "Old Title",
+            "version": {"number": 4},
+            "body": {"storage": {"value": "<p>Old</p>"}},
+        }
+        mock_put.return_value = {"id": "123", "version": {"number": 5}}
+
+        result = confluence.update_page("123", "New Title", "<p>New</p>", parent_id="9", version_comment="Sync")
+
+        mock_get.assert_called_once_with("rest/api/content/123", params={"expand": "body.storage,version"})
+        mock_put.assert_called_once_with(
+            "rest/api/content/123",
+            data={
+                "id": "123",
+                "type": "page",
+                "title": "New Title",
+                "version": {"number": 5, "minorEdit": False, "message": "Sync"},
+                "metadata": {
+                    "properties": {
+                        "content-appearance-draft": {"value": "fixed-width"},
+                        "content-appearance-published": {"value": "fixed-width"},
+                    }
+                },
+                "body": {"storage": {"value": "<p>New</p>", "representation": "storage"}},
+                "ancestors": [{"type": "page", "id": "9"}],
+            },
+            params={"status": "current"},
+        )
+        assert result == {"id": "123", "version": {"number": 5}}
+
+    @patch.object(ConfluenceCloud, "put")
+    @patch.object(ConfluenceCloud, "get")
+    def test_update_page_skips_unchanged_content(self, mock_get, mock_put, confluence_cloud):
+        page = {"id": "123", "title": "Title", "version": {"number": 4}, "body": {"storage": {"value": "<p>Same</p>"}}}
+        mock_get.return_value = page
+
+        assert confluence_cloud.update_page("123", "Title", "<p>Same</p>") == page
+        mock_put.assert_not_called()
+
+        confluence_cloud.update_page("123", "Title", "<p>Same</p>", always_update=True)
+        assert mock_put.call_args.kwargs["data"]["version"]["number"] == 5
+
     @patch.object(ConfluenceCloud, "delete")
     def test_delete_content(self, mock_delete, confluence_cloud):
         """Test delete_content method."""