diff --git a/joppy/api.py b/joppy/api.py index fa9f524..5979f89 100644 --- a/joppy/api.py +++ b/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) diff --git a/joppy/data_types.py b/joppy/data_types.py index 9276317..05712a3 100644 --- a/joppy/data_types.py +++ b/joppy/data_types.py @@ -233,6 +233,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""" @@ -273,7 +295,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 diff --git a/test/test_api.py b/test/test_api.py index 089ca40..98be1b6 100644 --- a/test/test_api.py +++ b/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: @@ -531,6 +533,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):