Support revisions

joppy/api.py

@@ -3,6 +3,7 @@
 import copy
 import json
 import logging
+import time
 from typing import (
     Any,
     Callable,
@@ -244,6 +245,54 @@ def modify_resource(self, id_: str, **data: dt.JoplinTypes) -> None:
         self.put(f"/resources/{id_}", data=data)
 
 
+class Revision(ApiBase):
+    """
+    Revisions are supported since Joplin 2.13.2.
+    See: https://github.com/laurent22/joplin/releases/tag/v2.13.2
+    """
+
+    def add_revision(
+        self, item_id: str, item_type: dt.ItemType, **data: dt.JoplinTypes
+    ) -> str:
+        """Add a revision to an item."""
+        now_unix_seconds = int(time.time())
+        return str(
+            self.post(
+                "/revisions",
+                data={
+                    # There seem to be some undocumented required fields.
+                    "item_id": item_id,
+                    "item_type": item_type.value,
+                    "item_updated_time": now_unix_seconds,
+                    "item_created_time": now_unix_seconds,
+                    **data,
+                },
+            ).json()["id"]
+        )
+
+    def delete_revision(self, id_: str) -> None:
+        """Delete a revision."""
+        self.delete(f"/revisions/{id_}")
+
+    def get_revision(self, id_: str, **query: dt.JoplinTypes) -> dt.RevisionData:
+        """Get the revision with the given ID."""
+        response = dt.RevisionData(**self.get(f"/revisions/{id_}", query=query).json())
+        return response
+
+    def get_revisions(self, **query: dt.JoplinTypes) -> dt.DataList[dt.RevisionData]:
+        """
+        Get revisions, paginated. To get all revisions (unpaginated), use
+        "get_all_revisions()".
+        """
+        response = self.get("/revisions", query=query).json()
+        response["items"] = [dt.RevisionData(**item) for item in response["items"]]
+        return dt.DataList[dt.RevisionData](**response)
+
+    def modify_revision(self, id_: str, **data: dt.JoplinTypes) -> None:
+        """Modify a revision."""
+        self.put(f"/revisions/{id_}", data=data)
+
+
 class Search(ApiBase):
     def search(
         self, **query: dt.JoplinTypes
@@ -319,7 +368,7 @@ def modify_tag(self, id_: str, **data: dt.JoplinTypes) -> None:
         self.put(f"/tags/{id_}", data=data)
 
 
-class Api(Event, Note, Notebook, Ping, Resource, Search, Tag):
+class Api(Event, Note, Notebook, Ping, Resource, Revision, Search, Tag):
     """
     Collects all basic API functions and contains a few more useful methods.
     This should be the only class accessed from the users.
@@ -364,6 +413,12 @@ def delete_all_resources(self) -> None:
             assert resource.id is not None
             self.delete_resource(resource.id)
 
+    def delete_all_revisions(self) -> None:
+        """Delete all revisions."""
+        for revision in self.get_all_revisions():
+            assert revision.id is not None
+            self.delete_revision(revision.id)
+
     def delete_all_tags(self) -> None:
         """Delete all tags."""
         for tag in self.get_all_tags():
@@ -401,6 +456,10 @@ def get_all_resources(self, **query: dt.JoplinTypes) -> List[dt.ResourceData]:
         """Get all resources, unpaginated."""
         return self._unpaginate(self.get_resources, **query)
 
+    def get_all_revisions(self, **query: dt.JoplinTypes) -> List[dt.RevisionData]:
+        """Get all revisions, unpaginated."""
+        return self._unpaginate(self.get_revisions, **query)
+
     def get_all_tags(self, **query: dt.JoplinTypes) -> List[dt.TagData]:
         """Get all tags, unpaginated."""
         return self._unpaginate(self.get_tags, **query)

joppy/data_types.py

@@ -223,6 +223,28 @@ def default_fields() -> Set[str]:
         return {"id", "title"}
 
 
+@dataclass
+class RevisionData(BaseData):
+    """https://joplinapp.org/help/api/references/rest_api/#revisions"""
+
+    id: Optional[str] = None
+    parent_id: Optional[str] = None
+    item_type: Optional[ItemType] = None
+    item_id: Optional[str] = None
+    item_updated_time: Optional[datetime] = None
+    title_diff: Optional[str] = None
+    body_diff: Optional[str] = None
+    metadata_diff: Optional[str] = None
+    encryption_cipher_text: Optional[str] = None
+    encryption_applied: Optional[bool] = None
+    updated_time: Optional[datetime] = None
+    created_time: Optional[datetime] = None
+
+    @staticmethod
+    def default_fields() -> Set[str]:
+        return {"id"}
+
+
 @dataclass
 class TagData(BaseData):
     """https://joplinapp.org/api/references/rest_api/#tags"""
@@ -262,7 +284,9 @@ def default_fields() -> Set[str]:
         return {"id", "item_type", "item_id", "type", "created_time"}
 
 
-T = TypeVar("T", EventData, NoteData, NotebookData, ResourceData, TagData, str)
+T = TypeVar(
+    "T", EventData, NoteData, NotebookData, ResourceData, RevisionData, TagData, str
+)
 
 
 @dataclass

test/test_api.py

@@ -73,10 +73,12 @@ def setUp(self):
         logging.debug("Test: %s", self.id())
 
         self.api = Api(token=API_TOKEN)
-        # Note: Notes get deleted automatically.
+        # Notes get deleted automatically.
         self.api.delete_all_notebooks()
         self.api.delete_all_resources()
         self.api.delete_all_tags()
+        # Delete revisions last to cover all previous deletions.
+        self.api.delete_all_revisions()
 
     @staticmethod
     def get_random_id() -> str:
@@ -525,6 +527,62 @@ def test_check_property_title(self, filename):
         self.assertEqual(resource.title, title)
 
 
+class Revision(TestBase):
+    def test_add(self):
+        """Add a revision."""
+        self.api.add_notebook()  # notebook for the note
+        note_id = self.api.add_note()  # note that gets a new revision
+        id_ = self.api.add_revision(note_id, dt.ItemType.NOTE)
+
+        revisions = self.api.get_revisions().items
+        self.assertEqual(len(revisions), 1)
+        self.assertEqual(revisions[0].id, id_)
+
+    def test_get_revision(self):
+        """Get a specific revision."""
+        self.api.add_notebook()  # notebook for the note
+        note_id = self.api.add_note()  # note that gets a new revision
+        id_ = self.api.add_revision(note_id, dt.ItemType.NOTE)
+        revision = self.api.get_revision(id_=id_)
+        self.assertEqual(revision.assigned_fields(), revision.default_fields())
+        self.assertEqual(revision.type_, dt.ItemType.REVISION)
+
+    def test_get_revisions(self):
+        """Get all revisions."""
+        self.api.add_notebook()  # notebook for the note
+        note_id = self.api.add_note()  # note that gets a new revision
+        self.api.add_revision(note_id, dt.ItemType.NOTE)
+        revisions = self.api.get_revisions()
+        self.assertEqual(len(revisions.items), 1)
+        self.assertFalse(revisions.has_more)
+        for revision in revisions.items:
+            self.assertEqual(revision.assigned_fields(), revision.default_fields())
+
+    def test_get_all_revisions(self):
+        """Get all revisions, unpaginated."""
+        # Small limit and count to create/remove as less as possible items.
+        count, limit = random.randint(1, 10), random.randint(1, 10)
+        self.api.add_notebook()  # notebook for the note
+        note_id = self.api.add_note()  # note that gets a new revision
+        for _ in range(count):
+            self.api.add_revision(note_id, dt.ItemType.NOTE)
+        self.assertEqual(
+            len(self.api.get_revisions(limit=limit).items), min(limit, count)
+        )
+        self.assertEqual(len(self.api.get_all_revisions(limit=limit)), count)
+
+    def test_get_revisions_valid_properties(self):
+        """Try to get specific properties of a revision."""
+        self.api.add_notebook()  # notebook for the note
+        note_id = self.api.add_note()  # note that gets a new revision
+        self.api.add_revision(note_id, dt.ItemType.NOTE)
+        property_combinations = self.get_combinations(dt.RevisionData.fields())
+        for properties in property_combinations:
+            revisions = self.api.get_revisions(fields=",".join(properties))
+            for revision in revisions.items:
+                self.assertEqual(revision.assigned_fields(), set(properties))
+
+
 # TODO: Add more tests for the search parameter.
 class Search(TestBase):
     def test_empty(self):