Drop sync API for Activity, Notes, User Status, Weather Status

README.md

@@ -17,7 +17,17 @@ Python library that provides a robust and well-documented API that allows develo
  * **Reliable**: Minimum number of incompatible changes.
  * **Robust**: All code is covered with tests as much as possible.
  * **Easy**: Designed to be easy to use.
- * **Sync + Async**: Provides both sync and async APIs.
+ * **Async-first**: Full async API with sync wrappers available for most modules.
+
+### Deprecation notice: sync API
+
+Starting with version **0.30.0**, we are gradually removing sync wrappers in favour of
+the async API. The following modules have already lost their sync counterparts:
+**Activity**, **Notes**, **User Status**, and **Weather Status**.
+
+All remaining sync methods will be phased out in future releases. If you are still
+using the sync `Nextcloud` / `NextcloudApp` classes, we recommend migrating to
+`AsyncNextcloud` / `AsyncNextcloudApp` as soon as possible.
 
 ### Differences between the Nextcloud and NextcloudApp classes
 

nc_py_api/_version.py

@@ -1,3 +1,3 @@
 """Version of nc_py_api."""
 
-__version__ = "0.24.0"
+__version__ = "0.30.0"

nc_py_api/activity.py

@@ -6,7 +6,7 @@
 
 from ._exceptions import NextcloudExceptionNotModified
 from ._misc import check_capabilities, nc_iso_time_to_datetime
-from ._session import AsyncNcSessionBasic, NcSessionBasic
+from ._session import AsyncNcSessionBasic
 
 
 @dataclasses.dataclass
@@ -136,58 +136,6 @@ def __repr__(self):
         )
 
 
-class _ActivityAPI:
-    """The class provides the Activity Application API."""
-
-    _ep_base: str = "/ocs/v1.php/apps/activity"
-    last_given: int
-    """Used by ``get_activities``, when **since** param is ``True``."""
-
-    def __init__(self, session: NcSessionBasic):
-        self._session = session
-        self.last_given = 0
-
-    @property
-    def available(self) -> bool:
-        """Returns True if the Nextcloud instance supports this feature, False otherwise."""
-        return not check_capabilities("activity.apiv2", self._session.capabilities)
-
-    def get_activities(
-        self,
-        filter_id: ActivityFilter | str = "",
-        since: int | bool = 0,
-        limit: int = 50,
-        object_type: str = "",
-        object_id: int = 0,
-        sort: str = "desc",
-    ) -> list[Activity]:
-        """Returns activities for the current user.
-
-        :param filter_id: Filter to apply, if needed.
-        :param since: Last activity ID you have seen. When specified, only activities after provided are returned.
-            Can be set to ``True`` to automatically use last ``last_given`` from previous calls. Default = **0**.
-        :param limit: Max number of activities to be returned.
-        :param object_type: Filter the activities to a given object.
-        :param object_id: Filter the activities to a given object.
-        :param sort: Sort activities ascending or descending. Default is ``desc``.
-
-        .. note:: ``object_type`` and ``object_id`` should only appear together with ``filter_id`` unset.
-        """
-        if since is True:
-            since = self.last_given
-        url, params = _get_activities(filter_id, since, limit, object_type, object_id, sort)
-        try:
-            result = self._session.ocs("GET", self._ep_base + url, params=params)
-        except NextcloudExceptionNotModified:
-            return []
-        self.last_given = int(self._session.response_headers["X-Activity-Last-Given"])
-        return [Activity(i) for i in result]
-
-    def get_filters(self) -> list[ActivityFilter]:
-        """Returns avalaible activity filters."""
-        return [ActivityFilter(i) for i in self._session.ocs("GET", self._ep_base + "/api/v2/activity/filters")]
-
-
 class _AsyncActivityAPI:
     """The class provides the async Activity Application API."""
 

nc_py_api/nextcloud.py

@@ -27,7 +27,7 @@
 )
 from ._talk_api import _AsyncTalkAPI, _TalkAPI
 from ._theming import ThemingInfo, get_parsed_theme
-from .activity import _ActivityAPI, _AsyncActivityAPI
+from .activity import _AsyncActivityAPI
 from .apps import _AppsAPI, _AsyncAppsAPI
 from .calendar_api import _CalendarAPI
 from .ex_app.defs import LogLvl
@@ -37,28 +37,24 @@
 from .files.files import FilesAPI
 from .files.files_async import AsyncFilesAPI
 from .loginflow_v2 import _AsyncLoginFlowV2API, _LoginFlowV2API
-from .notes import _AsyncNotesAPI, _NotesAPI
+from .notes import _AsyncNotesAPI
 from .notifications import _AsyncNotificationsAPI, _NotificationsAPI
-from .user_status import _AsyncUserStatusAPI, _UserStatusAPI
+from .user_status import _AsyncUserStatusAPI
 from .users import _AsyncUsersAPI, _UsersAPI
 from .users_groups import _AsyncUsersGroupsAPI, _UsersGroupsAPI
-from .weather_status import _AsyncWeatherStatusAPI, _WeatherStatusAPI
+from .weather_status import _AsyncWeatherStatusAPI
 from .webhooks import _AsyncWebhooksAPI, _WebhooksAPI
 
 
 class _NextcloudBasic(ABC):  # pylint: disable=too-many-instance-attributes
     apps: _AppsAPI
     """Nextcloud API for App management"""
-    activity: _ActivityAPI
-    """Activity Application API"""
     cal: _CalendarAPI
     """Nextcloud Calendar API"""
     files: FilesAPI
     """Nextcloud API for File System and Files Sharing"""
     preferences: PreferencesAPI
     """Nextcloud User Preferences API"""
-    notes: _NotesAPI
-    """Nextcloud Notes API"""
     notifications: _NotificationsAPI
     """Nextcloud API for managing user notifications"""
     talk: _TalkAPI
@@ -67,27 +63,19 @@ class _NextcloudBasic(ABC):  # pylint: disable=too-many-instance-attributes
     """Nextcloud API for managing users."""
     users_groups: _UsersGroupsAPI
     """Nextcloud API for managing user groups."""
-    user_status: _UserStatusAPI
-    """Nextcloud API for managing users statuses"""
-    weather_status: _WeatherStatusAPI
-    """Nextcloud API for managing user weather statuses"""
     webhooks: _WebhooksAPI
     """Nextcloud API for managing webhooks"""
     _session: NcSessionBasic
 
     def __init__(self, session: NcSessionBasic):
         self.apps = _AppsAPI(session)
-        self.activity = _ActivityAPI(session)
         self.cal = _CalendarAPI(session)
         self.files = FilesAPI(session)
         self.preferences = PreferencesAPI(session)
-        self.notes = _NotesAPI(session)
         self.notifications = _NotificationsAPI(session)
         self.talk = _TalkAPI(session)
         self.users = _UsersAPI(session)
         self.users_groups = _UsersGroupsAPI(session)
-        self.user_status = _UserStatusAPI(session)
-        self.weather_status = _WeatherStatusAPI(session)
         self.webhooks = _WebhooksAPI(session)
 
     @property
@@ -381,8 +369,6 @@ def set_user(self, user_id: str):
             self._session.set_user(user_id)
             self.talk.config_sha = ""
             self.talk.modified_since = 0
-            self.activity.last_given = 0
-            self.notes.last_etag = ""
             self._session.update_server_info()
 
     @property

nc_py_api/notes.py

@@ -9,7 +9,7 @@
 
 from ._exceptions import check_error
 from ._misc import check_capabilities, clear_from_params_empty, require_capabilities
-from ._session import AsyncNcSessionBasic, NcSessionBasic
+from ._session import AsyncNcSessionBasic
 
 
 @dataclasses.dataclass
@@ -91,144 +91,6 @@ class NotesSettings(typing.TypedDict):
     """Newly created note's files will have this file suffix. Default is **.txt**."""
 
 
-class _NotesAPI:
-    """Class implementing Nextcloud Notes API."""
-
-    _ep_base: str = "/index.php/apps/notes/api/v1"  # without `index.php` we will get 405 error.
-    last_etag: str
-    """Used by ``get_list``, when **etag** param is ``True``."""
-
-    def __init__(self, session: NcSessionBasic):
-        self._session = session
-        self.last_etag = ""
-
-    @property
-    def available(self) -> bool:
-        """Returns True if the Nextcloud instance supports this feature, False otherwise."""
-        return not check_capabilities("notes", self._session.capabilities)
-
-    def get_list(
-        self,
-        category: str | None = None,
-        modified_since: int | None = None,
-        limit: int | None = None,
-        cursor: str | None = None,
-        no_content: bool = False,
-        etag: bool = False,
-    ) -> list[Note]:
-        """Get information of all Notes.
-
-        :param category: Filter the result by category name. Notes with another category are not included in the result.
-        :param modified_since: When provided only results newer than given Unix timestamp are returned.
-        :param limit: Limit response to contain no more than the given number of notes.
-            If there are more notes, then the result is chunked and the HTTP response header
-            **X-Notes-Chunk-Cursor** is sent with a string value.
-
-            .. note:: Use :py:attr:`~nc_py_api.nextcloud.Nextcloud.response_headers` property to achieve that.
-        :param cursor: You should use the string value from the last request's HTTP response header
-            ``X-Notes-Chunk-Cursor`` in order to get the next chunk of notes.
-        :param no_content: Flag indicating should ``content`` field be excluded from response.
-        :param etag: Flag indicating should ``ETag`` from last call be used. Default = **False**.
-        """
-        require_capabilities("notes", self._session.capabilities)
-        params = {
-            "category": category,
-            "pruneBefore": modified_since,
-            "exclude": "content" if no_content else None,
-            "chunkSize": limit,
-            "chunkCursor": cursor,
-        }
-        clear_from_params_empty(list(params.keys()), params)
-        headers = {"If-None-Match": self.last_etag} if self.last_etag and etag else {}
-        r = _res_to_json(self._session.adapter.get(self._ep_base + "/notes", params=params, headers=headers))
-        self.last_etag = self._session.response_headers["ETag"]
-        return [Note(i) for i in r]
-
-    def by_id(self, note: Note) -> Note:
-        """Get updated information about :py:class:`~nc_py_api.notes.Note`."""
-        require_capabilities("notes", self._session.capabilities)
-        r = _res_to_json(
-            self._session.adapter.get(
-                self._ep_base + f"/notes/{note.note_id}", headers={"If-None-Match": f'"{note.etag}"'}
-            )
-        )
-        return Note(r) if r else note
-
-    def create(
-        self,
-        title: str,
-        content: str | None = None,
-        category: str | None = None,
-        favorite: bool | None = None,
-        last_modified: int | str | datetime.datetime | None = None,
-    ) -> Note:
-        """Create new Note."""
-        require_capabilities("notes", self._session.capabilities)
-        params = {
-            "title": title,
-            "content": content,
-            "category": category,
-            "favorite": favorite,
-            "modified": last_modified,
-        }
-        clear_from_params_empty(list(params.keys()), params)
-        return Note(_res_to_json(self._session.adapter.post(self._ep_base + "/notes", json=params)))
-
-    def update(
-        self,
-        note: Note,
-        title: str | None = None,
-        content: str | None = None,
-        category: str | None = None,
-        favorite: bool | None = None,
-        overwrite: bool = False,
-    ) -> Note:
-        """Updates Note.
-
-        ``overwrite`` specifies should be or not the Note updated even if it was changed on server(has different ETag).
-        """
-        require_capabilities("notes", self._session.capabilities)
-        headers = {"If-Match": f'"{note.etag}"'} if not overwrite else {}
-        params = {
-            "title": title,
-            "content": content,
-            "category": category,
-            "favorite": favorite,
-        }
-        clear_from_params_empty(list(params.keys()), params)
-        if not params:
-            raise ValueError("Nothing to update.")
-        return Note(
-            _res_to_json(
-                self._session.adapter.put(self._ep_base + f"/notes/{note.note_id}", json=params, headers=headers)
-            )
-        )
-
-    def delete(self, note: int | Note) -> None:
-        """Deletes a Note."""
-        require_capabilities("notes", self._session.capabilities)
-        note_id = note.note_id if isinstance(note, Note) else note
-        check_error(self._session.adapter.delete(self._ep_base + f"/notes/{note_id}"))
-
-    def get_settings(self) -> NotesSettings:
-        """Returns Notes App settings."""
-        require_capabilities("notes", self._session.capabilities)
-        r = _res_to_json(self._session.adapter.get(self._ep_base + "/settings"))
-        return {"notes_path": r["notesPath"], "file_suffix": r["fileSuffix"]}
-
-    def set_settings(self, notes_path: str | None = None, file_suffix: str | None = None) -> None:
-        """Change specified setting(s)."""
-        if notes_path is None and file_suffix is None:
-            raise ValueError("No setting to change.")
-        require_capabilities("notes", self._session.capabilities)
-        params = {
-            "notesPath": notes_path,
-            "fileSuffix": file_suffix,
-        }
-        clear_from_params_empty(list(params.keys()), params)
-        check_error(self._session.adapter.put(self._ep_base + "/settings", json=params))
-
-
 class _AsyncNotesAPI:
     """Class implements Async Nextcloud Notes API."""
 

nc_py_api/talk_bot.py

@@ -15,7 +15,7 @@
 from .nextcloud import AsyncNextcloudApp, NextcloudApp
 
 
-class ObjectContent(typing.TypedDict):
+class ObjectContent(typing.TypedDict, total=False):
     """Object content of :py:class:`~nc_py_api.talk_bot.TalkBotMessage`."""
 
     message: str
@@ -62,13 +62,22 @@ def object_name(self) -> str:
 
     @property
     def object_content(self) -> ObjectContent:
-        """Dictionary with a ``message`` and ``parameters`` keys."""
-        return json.loads(self._raw_data["object"]["content"])
+        """Dictionary with a ``message`` and ``parameters`` keys.
+
+        .. note:: May return an empty dict for system messages that have no content.
+        """
+        content = self._raw_data["object"].get("content")
+        if content is None:
+            return {}
+        return json.loads(content)
 
     @property
     def object_media_type(self) -> str:
-        """``text/markdown`` when the message should be interpreted as **Markdown**, otherwise ``text/plain``."""
-        return self._raw_data["object"]["mediaType"]
+        """``text/markdown`` when the message should be interpreted as **Markdown**, otherwise ``text/plain``.
+
+        .. note:: May return an empty string for system messages.
+        """
+        return self._raw_data["object"].get("mediaType", "")
 
     @property
     def conversation_token(self) -> str: