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)
|