add support for the server API

README.md

@@ -1,6 +1,6 @@
 # joppy
 
-Python interface for the [Joplin data API](https://joplinapp.org/api/references/rest_api/).
+Python interface for the [Joplin data API](https://joplinapp.org/api/references/rest_api/) (client) and the Joplin server API.
 
 [![build](https://github.com/marph91/joppy/actions/workflows/build.yml/badge.svg)](https://github.com/marph91/joppy/actions/workflows/build.yml)
 [![lint](https://github.com/marph91/joppy/actions/workflows/lint.yml/badge.svg)](https://github.com/marph91/joppy/actions/workflows/lint.yml)
@@ -41,6 +41,8 @@ For details, consult the [implementation](joppy/api.py), [joplin documentation](
 
 ## :bulb: Example snippets
 
+### Client API
+
 Start joplin and [get your API token](https://joplinapp.org/api/references/rest_api/#authorisation). Click to expand the examples.
 
 <details>
@@ -218,7 +220,24 @@ for resource in api.get_all_resources():
 
 </details>
 
-For more usage examples, check the example scripts or [tests](test/test_api.py).
+For more usage examples, check the example scripts or [tests](test/test_client_api.py).
+
+### Server API
+
+The server API should work similarly to the client API in most cases. **Be aware that the server API is experimental and may break at any time. Make sure you have a backup and know how to restore it.**
+
+```python
+from joppy.server_api import ServerApi
+
+# Create a new Api instance.
+api = ServerApi(user="admin@localhost", password="admin", url="http://localhost:22300")
+
+# Add a notebook.
+notebook_id = api.add_notebook(title="My first notebook")
+
+# Add a note in the previously created notebook.
+note_id = api.add_note(title="My first note", body="With some content", parent_id=notebook_id)
+```
 
 ## :newspaper: Examples
 

joppy/client_api.py

@@ -6,7 +6,6 @@
 import time
 from typing import (
     Any,
-    Callable,
     cast,
     Dict,
     List,
@@ -18,6 +17,7 @@
 import requests
 
 import joppy.data_types as dt
+from joppy import tools
 
 
 # Use a global session object for better performance.
@@ -33,12 +33,12 @@
 
 
 ##############################################################################
-# Base wrapper that manages the requests to the REST API.
+# Base wrapper that manages the requests to the client REST API.
 ##############################################################################
 
 
 class ApiBase:
-    """Contains the basic requests of the REST API."""
+    """Contains the basic requests of the client REST API."""
 
     def __init__(self, token: str, url: str = "http://localhost:41184") -> None:
         self.url = url
@@ -429,47 +429,32 @@ def delete_all_tags(self) -> None:
             assert tag.id is not None
             self.delete_tag(tag.id)
 
-    @staticmethod
-    def _unpaginate(
-        func: Callable[..., dt.DataList[dt.T]], **query: dt.JoplinTypes
-    ) -> List[dt.T]:
-        """Calls an Joplin endpoint until it's response doesn't contain more data."""
-        response = func(**query)
-        items = response.items
-        page = 1  # pages are one based
-        while response.has_more:
-            page += 1
-            query["page"] = page
-            response = func(**query)
-            items.extend(response.items)
-        return items
-
     def get_all_events(self, **query: dt.JoplinTypes) -> List[dt.EventData]:
         """Get all events, unpaginated."""
-        return self._unpaginate(self.get_events, **query)
+        return tools._unpaginate(self.get_events, **query)
 
     def get_all_notes(self, **query: dt.JoplinTypes) -> List[dt.NoteData]:
         """Get all notes, unpaginated."""
-        return self._unpaginate(self.get_notes, **query)
+        return tools._unpaginate(self.get_notes, **query)
 
     def get_all_notebooks(self, **query: dt.JoplinTypes) -> List[dt.NotebookData]:
         """Get all notebooks, unpaginated."""
-        return self._unpaginate(self.get_notebooks, **query)
+        return tools._unpaginate(self.get_notebooks, **query)
 
     def get_all_resources(self, **query: dt.JoplinTypes) -> List[dt.ResourceData]:
         """Get all resources, unpaginated."""
-        return self._unpaginate(self.get_resources, **query)
+        return tools._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)
+        return tools._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)
+        return tools._unpaginate(self.get_tags, **query)
 
     def search_all(
         self, **query: dt.JoplinTypes
     ) -> List[Union[dt.NoteData, dt.NotebookData, dt.ResourceData, dt.TagData]]:
         """Issue a search and get all results, unpaginated."""
-        return self._unpaginate(self.search, **query)  # type: ignore
+        return tools._unpaginate(self.search, **query)  # type: ignore

joppy/data_types.py

@@ -12,6 +12,7 @@
     Set,
     TypeVar,
 )
+import uuid
 
 
 # Datatypes used by the Joplin API. Needed for arbitrary kwargs.
@@ -20,14 +21,14 @@
 JoplinKwargs = MutableMapping[str, JoplinTypes]
 
 
-class EventChangeType(enum.Enum):
+class EventChangeType(enum.IntEnum):
     # https://joplinapp.org/api/references/rest_api/#properties-4
     CREATED = 1
     UPDATED = 2
     DELETED = 3
 
 
-class ItemType(enum.Enum):
+class ItemType(enum.IntEnum):
     # https://joplinapp.org/api/references/rest_api/#item-type-ids
     NOTE = 1
     FOLDER = 2
@@ -47,7 +48,7 @@ class ItemType(enum.Enum):
     COMMAND = 16
 
 
-class MarkupLanguage(enum.Enum):
+class MarkupLanguage(enum.IntEnum):
     # https://discourse.joplinapp.org/t/api-body-vs-body-html/11697/4
     MARKDOWN = 1
     HTML = 2
@@ -93,34 +94,44 @@ def __post_init__(self) -> None:
                 "todo_due",
                 "todo_completed",
             ):
-                casted_value = (
-                    None if value == 0 else datetime.fromtimestamp(value / 1000.0)
-                )
-                setattr(self, field_.name, casted_value)
+                try:
+                    value_int = int(value)
+                    casted_value = (
+                        None
+                        if value_int == 0
+                        else datetime.fromtimestamp(value_int / 1000.0)
+                    )
+                    setattr(self, field_.name, casted_value)
+                except ValueError:
+                    # TODO: This is not spec conform.
+                    casted_value = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
+                    setattr(self, field_.name, casted_value)
             elif field_.name in (
                 "is_conflict",
                 "is_todo",
                 "encryption_applied",
                 "is_shared",
                 "encryption_blob_encrypted",
             ):
-                setattr(self, field_.name, bool(value))
+                setattr(self, field_.name, bool(int(value)))
             elif field_.name == "latitude":
-                if not (-90 <= value <= 90):
+                setattr(self, field_.name, float(value))
+                if not (-90 <= float(value) <= 90):
                     raise ValueError("Invalid latitude:", value)
             elif field_.name == "longitude":
-                if not (-180 <= value <= 180):
+                setattr(self, field_.name, float(value))
+                if not (-180 <= float(value) <= 180):
                     raise ValueError("Invalid longitude:", value)
             elif field_.name == "markup_language":
-                setattr(self, field_.name, MarkupLanguage(value))
+                setattr(self, field_.name, MarkupLanguage(int(value)))
             # elif field_.name == "order":
             # elif field_.name == "crop_rect":
             # elif field_.name == "icon":
             # elif field_.name == "filename":  # "file_extension"
             elif field_.name in ("item_type", "type_"):
-                setattr(self, field_.name, ItemType(value))
+                setattr(self, field_.name, ItemType(int(value)))
             elif field_.name == "type":
-                setattr(self, field_.name, EventChangeType(value))
+                setattr(self, field_.name, EventChangeType(int(value)))
 
     def assigned_fields(self) -> Set[str]:
         # Exclude "type_" for convenience.
@@ -139,7 +150,7 @@ def fields(cls) -> Set[str]:
     def default_fields() -> Set[str]:
         return {"id", "parent_id", "title"}
 
-    def __repr__(self) -> str:
+    def __str__(self) -> str:
         # show only fields with values
         not_none_fields = ", ".join(
             f"{field.name}={getattr(self, field.name)}"
@@ -188,6 +199,44 @@ class NoteData(BaseData):
     image_data_url: Optional[str] = None
     crop_rect: Optional[str] = None
 
+    def serialize(self) -> str:
+        lines = []
+        if self.title is not None:
+            lines.extend([self.title, ""])
+        if self.body is not None:
+            lines.extend([self.body, ""])
+        for field_ in fields(self):
+            match field_.name:
+                case "id":
+                    # ID is always required
+                    if self.id is None:
+                        self.id = uuid.uuid4().hex
+                    lines.append(f"{field_.name}: {self.id}")
+                case "markup_language":
+                    # required to get an editable note
+                    if self.markup_language is None:
+                        self.markup_language = MarkupLanguage.MARKDOWN
+                    lines.append(f"{field_.name}: {self.markup_language}")
+                case "source_application":
+                    if self.source_application is None:
+                        self.source_application = "joppy"
+                    lines.append(f"{field_.name}: {self.source_application}")
+                case "title" | "body":
+                    pass  # handled before
+                case "type_":
+                    self.item_type = ItemType.NOTE
+                    lines.append(f"{field_.name}: {self.item_type}")
+                case "updated_time":
+                    # required, even if empty
+                    value_raw = getattr(self, field_.name)
+                    value = "" if value_raw is None else value_raw
+                    lines.append(f"{field_.name}: {value}")
+                case _:
+                    value_raw = getattr(self, field_.name)
+                    if value_raw is not None:
+                        lines.append(f"{field_.name}: {value_raw}")
+        return "\n".join(lines)
+
 
 @dataclass
 class NotebookData(BaseData):
@@ -209,6 +258,33 @@ class NotebookData(BaseData):
     user_data: Optional[str] = None
     deleted_time: Optional[datetime] = None
 
+    def serialize(self) -> str:
+        lines = []
+        if self.title is not None:
+            lines.extend([self.title, ""])
+        for field_ in fields(self):
+            match field_.name:
+                case "id":
+                    # ID is always required
+                    if self.id is None:
+                        self.id = uuid.uuid4().hex
+                    lines.append(f"{field_.name}: {self.id}")
+                case "title":
+                    pass  # handled before
+                case "type_":
+                    self.item_type = ItemType.FOLDER
+                    lines.append(f"{field_.name}: {self.item_type}")
+                case "updated_time":
+                    # required, even if empty
+                    value_raw = getattr(self, field_.name)
+                    value = "" if value_raw is None else value_raw
+                    lines.append(f"{field_.name}: {value}")
+                case _:
+                    value_raw = getattr(self, field_.name)
+                    if value_raw is not None:
+                        lines.append(f"{field_.name}: {value_raw}")
+        return "\n".join(lines)
+
 
 @dataclass
 class ResourceData(BaseData):
@@ -280,6 +356,72 @@ class TagData(BaseData):
     parent_id: Optional[str] = None
     user_data: Optional[str] = None
 
+    def serialize(self) -> str:
+        lines = []
+        if self.title is not None:
+            lines.extend([self.title, ""])
+        for field_ in fields(self):
+            match field_.name:
+                case "id":
+                    # ID is always required
+                    if self.id is None:
+                        self.id = uuid.uuid4().hex
+                    lines.append(f"{field_.name}: {self.id}")
+                case "title":
+                    pass  # handled before
+                case "type_":
+                    self.item_type = ItemType.TAG
+                    lines.append(f"{field_.name}: {self.item_type}")
+                case "updated_time":
+                    # required, even if empty
+                    value_raw = getattr(self, field_.name)
+                    value = "" if value_raw is None else value_raw
+                    lines.append(f"{field_.name}: {value}")
+                case _:
+                    value_raw = getattr(self, field_.name)
+                    if value_raw is not None:
+                        lines.append(f"{field_.name}: {value_raw}")
+        return "\n".join(lines)
+
+
+@dataclass
+class NoteTagData(BaseData):
+    """Links a tag to a note."""
+
+    id: Optional[str] = None
+    note_id: Optional[str] = None
+    tag_id: Optional[str] = None
+    created_time: Optional[datetime] = None
+    updated_time: Optional[datetime] = None
+    user_created_time: Optional[datetime] = None
+    user_updated_time: Optional[datetime] = None
+    encryption_cipher_text: Optional[str] = None
+    encryption_applied: Optional[bool] = None
+    is_shared: Optional[bool] = None
+
+    def serialize(self) -> str:
+        lines = []
+        for field_ in fields(self):
+            match field_.name:
+                case "id":
+                    # ID is always required
+                    if self.id is None:
+                        self.id = uuid.uuid4().hex
+                    lines.append(f"{field_.name}: {self.id}")
+                case "type_":
+                    self.item_type = ItemType.NOTE_TAG
+                    lines.append(f"{field_.name}: {self.item_type}")
+                case "updated_time":
+                    # required, even if empty
+                    value_raw = getattr(self, field_.name)
+                    value = "" if value_raw is None else value_raw
+                    lines.append(f"{field_.name}: {value}")
+                case _:
+                    value_raw = getattr(self, field_.name)
+                    if value_raw is not None:
+                        lines.append(f"{field_.name}: {value_raw}")
+        return "\n".join(lines)
+
 
 @dataclass
 class EventData(BaseData):
@@ -304,6 +446,11 @@ def default_fields() -> Set[str]:
         return {"id", "item_type", "item_id", "type", "created_time"}
 
 
+AnyData = Union[
+    EventData, NoteData, NotebookData, NoteTagData, ResourceData, RevisionData, TagData
+]
+
+
 T = TypeVar(
     "T", EventData, NoteData, NotebookData, ResourceData, RevisionData, TagData, str
 )