File: _preferences_ex.py

package info (click to toggle)
nc-py-api 0.19.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,320 kB
  • sloc: python: 12,415; makefile: 238; xml: 100; javascript: 56; sh: 14
file content (175 lines) | stat: -rw-r--r-- 7,179 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
"""Nextcloud API for working with apps V2's storage w/wo user context(table oc_appconfig_ex/oc_preferences_ex)."""

import dataclasses

from ._exceptions import NextcloudExceptionNotFound
from ._misc import require_capabilities
from ._session import AsyncNcSessionBasic, NcSessionBasic


@dataclasses.dataclass
class CfgRecord:
    """A representation of a single key-value pair returned from the **get_values** method."""

    key: str
    value: str

    def __init__(self, raw_data: dict):
        self.key = raw_data["configkey"]
        self.value = raw_data["configvalue"]


class _BasicAppCfgPref:
    _url_suffix: str

    def __init__(self, session: NcSessionBasic):
        self._session = session

    def get_value(self, key: str, default=None) -> str | None:
        """Returns the value of the key, if found, or the specified default value."""
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", self._session.capabilities)
        r = self.get_values([key])
        if r:
            return r[0].value
        return default

    def get_values(self, keys: list[str]) -> list[CfgRecord]:
        """Returns the :py:class:`CfgRecord` for each founded key."""
        if not keys:
            return []
        if not all(keys):
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", self._session.capabilities)
        data = {"configKeys": keys}
        results = self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}/get-values", json=data)
        return [CfgRecord(i) for i in results]

    def delete(self, keys: str | list[str], not_fail=True) -> None:
        """Deletes config/preference entries by the provided keys."""
        if isinstance(keys, str):
            keys = [keys]
        if not keys:
            return
        if not all(keys):
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", self._session.capabilities)
        try:
            self._session.ocs("DELETE", f"{self._session.ae_url}/{self._url_suffix}", json={"configKeys": keys})
        except NextcloudExceptionNotFound as e:
            if not not_fail:
                raise e from None


class _AsyncBasicAppCfgPref:
    _url_suffix: str

    def __init__(self, session: AsyncNcSessionBasic):
        self._session = session

    async def get_value(self, key: str, default=None) -> str | None:
        """Returns the value of the key, if found, or the specified default value."""
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", await self._session.capabilities)
        r = await self.get_values([key])
        if r:
            return r[0].value
        return default

    async def get_values(self, keys: list[str]) -> list[CfgRecord]:
        """Returns the :py:class:`CfgRecord` for each founded key."""
        if not keys:
            return []
        if not all(keys):
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", await self._session.capabilities)
        data = {"configKeys": keys}
        results = await self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}/get-values", json=data)
        return [CfgRecord(i) for i in results]

    async def delete(self, keys: str | list[str], not_fail=True) -> None:
        """Deletes config/preference entries by the provided keys."""
        if isinstance(keys, str):
            keys = [keys]
        if not keys:
            return
        if not all(keys):
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", await self._session.capabilities)
        try:
            await self._session.ocs("DELETE", f"{self._session.ae_url}/{self._url_suffix}", json={"configKeys": keys})
        except NextcloudExceptionNotFound as e:
            if not not_fail:
                raise e from None


class PreferencesExAPI(_BasicAppCfgPref):
    """User specific preferences API, avalaible as **nc.preferences_ex.<method>**."""

    _url_suffix = "ex-app/preference"

    def set_value(self, key: str, value: str) -> None:
        """Sets a value for a key."""
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", self._session.capabilities)
        params = {"configKey": key, "configValue": value}
        self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}", json=params)


class AsyncPreferencesExAPI(_AsyncBasicAppCfgPref):
    """User specific preferences API."""

    _url_suffix = "ex-app/preference"

    async def set_value(self, key: str, value: str) -> None:
        """Sets a value for a key."""
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", await self._session.capabilities)
        params = {"configKey": key, "configValue": value}
        await self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}", json=params)


class AppConfigExAPI(_BasicAppCfgPref):
    """Non-user(App) specific preferences API, avalaible as **nc.appconfig_ex.<method>**."""

    _url_suffix = "ex-app/config"

    def set_value(self, key: str, value: str, sensitive: bool | None = None) -> None:
        """Sets a value and if specified the sensitive flag for a key.

        .. note:: A sensitive flag ensures key values are truncated in Nextcloud logs.
            Default for new records is ``False`` when sensitive is *unspecified*, if changes existing record and
            sensitive is *unspecified* it will not change the existing `sensitive` flag.
        """
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", self._session.capabilities)
        params: dict = {"configKey": key, "configValue": value}
        if sensitive is not None:
            params["sensitive"] = sensitive
        self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}", json=params)


class AsyncAppConfigExAPI(_AsyncBasicAppCfgPref):
    """Non-user(App) specific preferences API."""

    _url_suffix = "ex-app/config"

    async def set_value(self, key: str, value: str, sensitive: bool | None = None) -> None:
        """Sets a value and if specified the sensitive flag for a key.

        .. note:: A sensitive flag ensures key values are truncated in Nextcloud logs.
            Default for new records is ``False`` when sensitive is *unspecified*, if changes existing record and
            sensitive is *unspecified* it will not change the existing `sensitive` flag.
        """
        if not key:
            raise ValueError("`key` parameter can not be empty")
        require_capabilities("app_api", await self._session.capabilities)
        params: dict = {"configKey": key, "configValue": value}
        if sensitive is not None:
            params["sensitive"] = sensitive
        await self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}", json=params)