Source code for nc_py_api._preferences_ex

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


[docs] @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
[docs] class PreferencesExAPI(_BasicAppCfgPref): """User specific preferences API, avalaible as **nc.preferences_ex.<method>**.""" _url_suffix = "ex-app/preference"
[docs] 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)
[docs] class AppConfigExAPI(_BasicAppCfgPref): """Non-user(App) specific preferences API, avalaible as **nc.appconfig_ex.<method>**.""" _url_suffix = "ex-app/config"
[docs] 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)