extend check_capabilities (#92)

Allows this:

```python3

check_capabilities("user_status.enabled")
```

Needed for **Talk**, as it provides a very big amount of `capabilities`,
need a clean way to check for them.

---------

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

Author: bigcat88

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.0.32 - 2023-08-2x]
+
+### Changed
+
+- `require_capabilities`/`check_capabilities` can accept value with `dot`: like `files_sharing.api_enabled` and check for sub-values.
+
 ## [0.0.31 - 2023-08-17]
 
 ### Added

docs/FirstSteps.rst

@@ -11,6 +11,9 @@ Full support is only available from version ``27.1`` of Nextcloud.
 .. note:: In many cases, even if you want to develop an application,
     it's a good idea to first debug and develop part of it as a client.
 
+Basics
+^^^^^^
+
 Creating Nextcloud client class
 """""""""""""""""""""""""""""""
 
@@ -36,6 +39,40 @@ To test if this works, let's print the capabilities of the Nextcloud instance:
     pretty_capabilities = dumps(nc.capabilities, indent=4, sort_keys=True)
     print(pretty_capabilities)
 
+Checking Nextcloud capabilities
+"""""""""""""""""""""""""""""""
+
+In most cases, APIs perform capability checks before invoking them and raise a :class:`~nc_py_api._exceptions.NextcloudMissingCapabilities`
+exception if the Nextcloud instance lacks the requisite capability.
+However, there are situations where this approach might not be the most convenient,
+and you may wish to earlier whether a certain capability is available and active.
+
+To address this need, the ``check_capabilities`` method is provided.
+This method offers a straightforward way to proactively check for the existence and status of a particular capability.
+
+Using this method is quite simple:
+
+.. code-block:: python
+
+    import nc_py_api
+
+
+    nc = Nextcloud(nextcloud_url="http://nextcloud.local", nc_auth_user="admin", nc_auth_pass="admin")
+    if nc.check_capabilities("files_sharing"):  # check one capability
+        print("Sharing API is not present.")
+
+    # check child values in the same call
+    if nc.check_capabilities("files_sharing.api_enabled"):
+        print("Sharing API is present, but is not enabled.")
+
+    # check multiply capabilities at one
+    missing_cap = nc.check_capabilities(["files_sharing.api_enabled", "user_status.enabled"])
+    if missing_cap:
+        print(f"Missing capabilities: {missing_cap}")
+
+Files
+^^^^^
+
 Getting list of files of User
 """""""""""""""""""""""""""""
 
@@ -92,7 +129,7 @@ Example of using ``file.find()`` to search for file objects.
 .. literalinclude:: ../examples/as_client/files/find.py
 
 Conclusion
-""""""""""
+^^^^^^^^^^
 
 Once you have a good understanding of working with files, you can move on to more APIs.
 

docs/reference/Exceptions.rst

@@ -10,3 +10,6 @@ Avalaible as `nc_py_api.{exception_name}`
 
 .. autoclass:: NextcloudExceptionNotFound
     :members:
+
+.. autoclass:: NextcloudMissingCapabilities
+    :members:

nc_py_api/__init__.py

@@ -1,7 +1,11 @@
 """All possible stuff for Nextcloud & NextcloudApp that can be used."""
 
 from . import ex_app, options
-from ._exceptions import NextcloudException, NextcloudExceptionNotFound
+from ._exceptions import (
+    NextcloudException,
+    NextcloudExceptionNotFound,
+    NextcloudMissingCapabilities,
+)
 from ._version import __version__
 from .files import FilePermissions, FsNode
 from .files.sharing import ShareType

nc_py_api/_exceptions.py

@@ -27,6 +27,13 @@ def __init__(self, reason="Not found", info: str = ""):
         super().__init__(404, reason=reason, info=info)
 
 
+class NextcloudMissingCapabilities(NextcloudException):
+    """The exception that is thrown when required capability for API is missing."""
+
+    def __init__(self, reason="Missing capability", info: str = ""):
+        super().__init__(412, reason=reason, info=info)
+
+
 def check_error(code: int, info: str = ""):
     """Checks HTTP code from Nextcloud, and raises exception in case of error.
 

nc_py_api/_misc.py

@@ -7,7 +7,7 @@
 from string import ascii_lowercase, ascii_uppercase, digits
 from typing import Union
 
-from ._exceptions import NextcloudException
+from ._exceptions import NextcloudMissingCapabilities
 
 
 def kwargs_to_dict(keys: list[str], **kwargs) -> dict:
@@ -19,14 +19,32 @@ def require_capabilities(capabilities: Union[str, list[str]], srv_capabilities:
     """Checks for capabilities and raises an exception if any of them are missing."""
     result = check_capabilities(capabilities, srv_capabilities)
     if result:
-        raise NextcloudException(404, f"{result} is not available")
+        raise NextcloudMissingCapabilities(info=str(result))
+
+
+def __check_sub_capability(split_capabilities: list[str], srv_capabilities: dict) -> bool:
+    """Returns ``True`` if such capability is present and **enabled**."""
+    n_split_capabilities = len(split_capabilities)
+    capabilities_nesting = srv_capabilities
+    for i, v in enumerate(split_capabilities):
+        if i != 0 and i == n_split_capabilities - 1:
+            return bool(capabilities_nesting.get(v, False))
+        if v not in capabilities_nesting:
+            return False
+        capabilities_nesting = capabilities_nesting[v]
+    return True
 
 
 def check_capabilities(capabilities: Union[str, list[str]], srv_capabilities: dict) -> list[str]:
     """Checks for capabilities and returns a list of missing ones."""
     if isinstance(capabilities, str):
         capabilities = [capabilities]
-    return [i for i in capabilities if i not in srv_capabilities]
+    missing_capabilities = []
+    for capability in capabilities:
+        split_capabilities = capability.split(".")
+        if not __check_sub_capability(split_capabilities, srv_capabilities):
+            missing_capabilities.append(capability)
+    return missing_capabilities
 
 
 def random_string(size: int) -> str:

nc_py_api/files/sharing.py

@@ -131,7 +131,7 @@ def __init__(self, session: _session.NcSessionBasic):
     @property
     def available(self) -> bool:
         """Returns True if the Nextcloud instance supports this feature, False otherwise."""
-        return not _misc.check_capabilities("files_sharing", self._session.capabilities)
+        return not _misc.check_capabilities("files_sharing.api_enabled", self._session.capabilities)
 
     def get_list(
         self, shared_with_me=False, reshares=False, subfiles=False, path: typing.Union[str, FsNode] = ""
@@ -143,7 +143,7 @@ def get_list(
         :param subfiles: Only get all sub shares in a folder.
         :param path: Get shares for a specific path.
         """
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         path = path.user_path if isinstance(path, FsNode) else path
         params = {
             "shared_with_me": "true" if shared_with_me else "false",
@@ -157,13 +157,13 @@ def get_list(
 
     def get_by_id(self, share_id: int) -> Share:
         """Get Share by share ID."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         result = self._session.ocs(method="GET", path=f"{self._ep_base}/shares/{share_id}")
         return Share(result[0] if isinstance(result, list) else result)
 
     def get_inherited(self, path: str) -> list[Share]:
         """Get all shares relative to a file, e.g., parent folders shares."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         result = self._session.ocs(method="GET", path=f"{self._ep_base}/shares/inherited", params={"path": path})
         return [Share(i) for i in result]
 
@@ -195,7 +195,7 @@ def create(
             * ``note`` - string with note, if any. default = ``""``
             * ``label`` - string with label, if any. default = ``""``
         """
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         path = path.user_path if isinstance(path, FsNode) else path
         params = {
             "path": path,
@@ -226,7 +226,7 @@ def update(self, share_id: typing.Union[int, Share], **kwargs) -> Share:
         :param kwargs: Available for update: ``permissions``, ``password``, ``send_password_by_talk``,
           ``public_upload``, ``expire_date``, ``note``, ``label``.
         """
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         share_id = share_id.share_id if isinstance(share_id, Share) else share_id
         params: dict = {}
         if "permissions" in kwargs:
@@ -250,7 +250,7 @@ def delete(self, share_id: typing.Union[int, Share]) -> None:
 
         :param share_id: The Share object or an ID of the share.
         """
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         share_id = share_id.share_id if isinstance(share_id, Share) else share_id
         self._session.ocs(method="DELETE", path=f"{self._ep_base}/shares/{share_id}")
 
@@ -260,23 +260,23 @@ def get_pending(self) -> list[Share]:
 
     def accept_share(self, share_id: typing.Union[int, Share]) -> None:
         """Accept pending share."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         share_id = share_id.share_id if isinstance(share_id, Share) else share_id
         self._session.ocs(method="POST", path=f"{self._ep_base}/pending/{share_id}")
 
     def decline_share(self, share_id: typing.Union[int, Share]) -> None:
         """Decline pending share."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         share_id = share_id.share_id if isinstance(share_id, Share) else share_id
         self._session.ocs(method="DELETE", path=f"{self._ep_base}/pending/{share_id}")
 
     def get_deleted(self) -> list[Share]:
         """Get a list of deleted shares."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         return [Share(i) for i in self._session.ocs(method="GET", path=f"{self._ep_base}/deletedshares")]
 
     def undelete(self, share_id: typing.Union[int, Share]) -> None:
         """Undelete a deleted share."""
-        _misc.require_capabilities("files_sharing", self._session.capabilities)
+        _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities)
         share_id = share_id.share_id if isinstance(share_id, Share) else share_id
         self._session.ocs(method="POST", path=f"{self._ep_base}/deletedshares/{share_id}")

nc_py_api/users/status.py

@@ -98,30 +98,30 @@ def __init__(self, session: NcSessionBasic):
     @property
     def available(self) -> bool:
         """Returns True if the Nextcloud instance supports this feature, False otherwise."""
-        return not check_capabilities("user_status", self._session.capabilities)
+        return not check_capabilities("user_status.enabled", self._session.capabilities)
 
     def get_list(self, limit: Optional[int] = None, offset: Optional[int] = None) -> list[UserStatus]:
         """Returns statuses for all users.
 
         :param limit: limits the number of results.
         :param offset: offset of results.
         """
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         data = kwargs_to_dict(["limit", "offset"], limit=limit, offset=offset)
         result = self._session.ocs(method="GET", path=f"{self._ep_base}/statuses", params=data)
         return [UserStatus(i) for i in result]
 
     def get_current(self) -> CurrentUserStatus:
         """Returns the current user status."""
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         return CurrentUserStatus(self._session.ocs(method="GET", path=f"{self._ep_base}/user_status"))
 
     def get(self, user_id: str) -> Optional[UserStatus]:
         """Returns the user status for the specified user.
 
         :param user_id: User ID for getting status.
         """
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         try:
             return UserStatus(self._session.ocs(method="GET", path=f"{self._ep_base}/statuses/{user_id}"))
         except NextcloudExceptionNotFound:
@@ -131,7 +131,7 @@ def get_predefined(self) -> list[PredefinedStatus]:
         """Returns a list of predefined statuses available for installation on this Nextcloud instance."""
         if self._session.nc_version["major"] < 27:
             return []
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         result = self._session.ocs(method="GET", path=f"{self._ep_base}/predefined_statuses")
         return [PredefinedStatus(i) for i in result]
 
@@ -143,14 +143,15 @@ def set_predefined(self, status_id: str, clear_at: int = 0) -> None:
         """
         if self._session.nc_version["major"] < 27:
             return
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         params: dict[str, Union[int, str]] = {"messageId": status_id}
         if clear_at:
             params["clearAt"] = clear_at
         self._session.ocs(method="PUT", path=f"{self._ep_base}/user_status/message/predefined", params=params)
 
     def set_status_type(self, value: Literal["online", "away", "dnd", "invisible", "offline"]) -> None:
         """Sets the status type for the current user."""
+        require_capabilities("user_status.enabled", self._session.capabilities)
         self._session.ocs(method="PUT", path=f"{self._ep_base}/user_status/status", params={"statusType": value})
 
     def set_status(self, message: Optional[str] = None, clear_at: int = 0, status_icon: str = "") -> None:
@@ -160,12 +161,12 @@ def set_status(self, message: Optional[str] = None, clear_at: int = 0, status_ic
         :param clear_at: Unix Timestamp, representing the time to clear the status.
         :param status_icon: The icon picked by the user (must be one emoji)
         """
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         if message is None:
             self._session.ocs(method="DELETE", path=f"{self._ep_base}/user_status/message")
             return
         if status_icon:
-            require_capabilities("supports_emoji", self._session.capabilities["user_status"])
+            require_capabilities("user_status.supports_emoji", self._session.capabilities)
         params: dict[str, Union[int, str]] = {"message": message}
         if clear_at:
             params["clearAt"] = clear_at
@@ -178,7 +179,7 @@ def get_backup_status(self, user_id: str = "") -> Optional[UserStatus]:
 
         :param user_id: User ID for getting status.
         """
-        require_capabilities("user_status", self._session.capabilities)
+        require_capabilities("user_status.enabled", self._session.capabilities)
         user_id = user_id if user_id else self._session.user
         if not user_id:
             raise ValueError("user_id can not be empty.")
@@ -189,7 +190,7 @@ def restore_backup_status(self, status_id: str) -> Optional[CurrentUserStatus]:
 
         :param status_id: backup status ID.
         """
-        require_capabilities("user_status", self._session.capabilities)
-        require_capabilities("restore", self._session.capabilities["user_status"])
+        require_capabilities("user_status.enabled", self._session.capabilities)
+        require_capabilities("user_status.restore", self._session.capabilities)
         result = self._session.ocs(method="DELETE", path=f"{self._ep_base}/user_status/revert/{status_id}")
         return result if result else None

nc_py_api/users/weather.py

@@ -52,11 +52,11 @@ def __init__(self, session: NcSessionBasic):
     @property
     def available(self) -> bool:
         """Returns True if the Nextcloud instance supports this feature, False otherwise."""
-        return not check_capabilities("weather_status", self._session.capabilities)
+        return not check_capabilities("weather_status.enabled", self._session.capabilities)
 
     def get_location(self) -> WeatherLocation:
         """Returns the current location set on the Nextcloud server for the user."""
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         return WeatherLocation(self._session.ocs(method="GET", path=f"{self._ep_base}/location"))
 
     def set_location(
@@ -68,7 +68,7 @@ def set_location(
         :param longitude: east–west position of a point on the surface of the Earth.
         :param address: city, index(*optional*) and country, e.g. "Paris, 75007, France"
         """
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         params: dict[str, Union[str, float]] = {}
         if latitude is not None and longitude is not None:
             params.update({"lat": latitude, "lon": longitude})
@@ -81,24 +81,24 @@ def set_location(
 
     def get_forecast(self) -> list[dict]:
         """Get forecast for the current location."""
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         return self._session.ocs(method="GET", path=f"{self._ep_base}/forecast")
 
     def get_favorites(self) -> list[str]:
         """Returns favorites addresses list."""
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         return self._session.ocs(method="GET", path=f"{self._ep_base}/favorites")
 
     def set_favorites(self, favorites: list[str]) -> bool:
         """Sets favorites addresses list."""
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         result = self._session.ocs(method="PUT", path=f"{self._ep_base}/favorites", json={"favorites": favorites})
         return result.get("success", False)
 
     def set_mode(self, mode: WeatherLocationMode) -> bool:
         """Change the weather status mode."""
         if int(mode) == WeatherLocationMode.UNKNOWN.value:
             raise ValueError("This mode can not be set")
-        require_capabilities("weather_status", self._session.capabilities)
+        require_capabilities("weather_status.enabled", self._session.capabilities)
         result = self._session.ocs(method="PUT", path=f"{self._ep_base}/mode", params={"mode": int(mode)})
         return result.get("success", False)

tests/misc_test.py

@@ -36,6 +36,8 @@ def test_require_capabilities():
         require_capabilities(["non_exist_capability", "app_ecosystem_v2"], NC_APP.capabilities)
     with pytest.raises(NextcloudException):
         require_capabilities(["non_exist_capability", "non_exist_capability2", "app_ecosystem_v2"], NC_APP.capabilities)
+    with pytest.raises(NextcloudException):
+        require_capabilities("app_ecosystem_v2.non_exist_capability", NC_APP.capabilities)
 
 
 def test_config_get_value():