Source code for nc_py_api.user_status

"""Nextcloud API for working with user statuses."""

import dataclasses
import typing

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


[docs] @dataclasses.dataclass class ClearAt: """Determination when a user's predefined status will be cleared.""" clear_type: str """Possible values: ``period``, ``end-of``""" time: str | int """Depending of ``type`` it can be number of seconds relative to ``now`` or one of the next values: ``day``""" def __init__(self, raw_data: dict): self.clear_type = raw_data["type"] self.time = raw_data["time"]
[docs] @dataclasses.dataclass class PredefinedStatus: """Definition of the predefined status.""" status_id: str """ID of the predefined status""" icon: str """Icon in string(UTF) format""" message: str """The message defined for this status. It is translated, so it depends on the user's language setting.""" clear_at: ClearAt | None """When the default, if not override, the predefined status will be cleared.""" def __init__(self, raw_status: dict): self.status_id = raw_status["id"] self.icon = raw_status["icon"] self.message = raw_status["message"] clear_at_raw = raw_status.get("clearAt") if clear_at_raw: self.clear_at = ClearAt(clear_at_raw) else: self.clear_at = None
[docs] @dataclasses.dataclass class UserStatus: """Information about user status.""" user_id: str """The ID of the user this status is for""" def __init__(self, raw_data: dict): self._raw_data = raw_data self.user_id = raw_data["userId"] @property def status_message(self) -> str: """Message of the status.""" return self._raw_data.get("message", "") @property def status_icon(self) -> str: """The icon picked by the user (must be one emoji).""" return self._raw_data.get("icon", "") @property def status_clear_at(self) -> int | None: """Unix Timestamp representing the time to clear the status.""" return self._raw_data.get("clearAt", None) @property def status_type(self) -> str: """Status type, on of the: online, away, dnd, invisible, offline.""" return self._raw_data.get("status", "") def __repr__(self): return f"<{self.__class__.__name__} user_id={self.user_id}, status_type={self.status_type}>"
[docs] @dataclasses.dataclass(init=False) class CurrentUserStatus(UserStatus): """Information about current user status.""" @property def status_id(self) -> str | None: """ID of the predefined status.""" return self._raw_data["messageId"] @property def message_predefined(self) -> bool: """*True* if the status is predefined, *False* otherwise.""" return self._raw_data["messageIsPredefined"] @property def status_type_defined(self) -> bool: """*True* if :py:attr:`UserStatus.status_type` is set by user, *False* otherwise.""" return self._raw_data["statusIsUserDefined"] def __repr__(self): return ( f"<{self.__class__.__name__} user_id={self.user_id}, status_type={self.status_type}," f" status_id={self.status_id}>" )
[docs] class _UserStatusAPI: """Class providing the user status management API on the Nextcloud server.""" _ep_base: str = "/ocs/v1.php/apps/user_status/api/v1" def __init__(self, session: NcSessionBasic): self._session = session @property def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not check_capabilities("user_status.enabled", self._session.capabilities)
[docs] def get_list(self, limit: int | None = None, offset: int | None = None) -> list[UserStatus]: """Returns statuses for all users.""" require_capabilities("user_status.enabled", self._session.capabilities) data = kwargs_to_params(["limit", "offset"], limit=limit, offset=offset) result = self._session.ocs("GET", f"{self._ep_base}/statuses", params=data) return [UserStatus(i) for i in result]
[docs] def get_current(self) -> CurrentUserStatus: """Returns the current user status.""" require_capabilities("user_status.enabled", self._session.capabilities) return CurrentUserStatus(self._session.ocs("GET", f"{self._ep_base}/user_status"))
[docs] def get(self, user_id: str) -> UserStatus | None: """Returns the user status for the specified user.""" require_capabilities("user_status.enabled", self._session.capabilities) try: return UserStatus(self._session.ocs("GET", f"{self._ep_base}/statuses/{user_id}")) except NextcloudExceptionNotFound: return None
[docs] 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.enabled", self._session.capabilities) result = self._session.ocs("GET", f"{self._ep_base}/predefined_statuses") return [PredefinedStatus(i) for i in result]
[docs] def set_predefined(self, status_id: str, clear_at: int = 0) -> None: """Set predefined status for the current user. :param status_id: ``predefined`` status ID. :param clear_at: *optional* time in seconds before the status is cleared. """ if self._session.nc_version["major"] < 27: return require_capabilities("user_status.enabled", self._session.capabilities) params: dict[str, int | str] = {"messageId": status_id} if clear_at: params["clearAt"] = clear_at self._session.ocs("PUT", f"{self._ep_base}/user_status/message/predefined", params=params)
[docs] def set_status_type(self, value: typing.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("PUT", f"{self._ep_base}/user_status/status", params={"statusType": value})
[docs] def set_status(self, message: str | None = None, clear_at: int = 0, status_icon: str = "") -> None: """Sets current user status. :param message: Message text to set in the status. :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.enabled", self._session.capabilities) if message is None: self._session.ocs("DELETE", f"{self._ep_base}/user_status/message") return if status_icon: require_capabilities("user_status.supports_emoji", self._session.capabilities) params: dict[str, int | str] = {"message": message} if clear_at: params["clearAt"] = clear_at if status_icon: params["statusIcon"] = status_icon self._session.ocs("PUT", f"{self._ep_base}/user_status/message/custom", params=params)
[docs] def get_backup_status(self, user_id: str = "") -> UserStatus | None: """Get the backup status of the user if any.""" 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.") return self.get(f"_{user_id}")
[docs] def restore_backup_status(self, status_id: str) -> CurrentUserStatus | None: """Restores the backup state as current for the current user.""" require_capabilities("user_status.enabled", self._session.capabilities) require_capabilities("user_status.restore", self._session.capabilities) result = self._session.ocs("DELETE", f"{self._ep_base}/user_status/revert/{status_id}") return result if result else None
class _AsyncUserStatusAPI: """Class provides async user status management API on the Nextcloud server.""" _ep_base: str = "/ocs/v1.php/apps/user_status/api/v1" def __init__(self, session: AsyncNcSessionBasic): self._session = session @property async def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not check_capabilities("user_status.enabled", await self._session.capabilities) async def get_list(self, limit: int | None = None, offset: int | None = None) -> list[UserStatus]: """Returns statuses for all users.""" require_capabilities("user_status.enabled", await self._session.capabilities) data = kwargs_to_params(["limit", "offset"], limit=limit, offset=offset) result = await self._session.ocs("GET", f"{self._ep_base}/statuses", params=data) return [UserStatus(i) for i in result] async def get_current(self) -> CurrentUserStatus: """Returns the current user status.""" require_capabilities("user_status.enabled", await self._session.capabilities) return CurrentUserStatus(await self._session.ocs("GET", f"{self._ep_base}/user_status")) async def get(self, user_id: str) -> UserStatus | None: """Returns the user status for the specified user.""" require_capabilities("user_status.enabled", await self._session.capabilities) try: return UserStatus(await self._session.ocs("GET", f"{self._ep_base}/statuses/{user_id}")) except NextcloudExceptionNotFound: return None async def get_predefined(self) -> list[PredefinedStatus]: """Returns a list of predefined statuses available for installation on this Nextcloud instance.""" if (await self._session.nc_version)["major"] < 27: return [] require_capabilities("user_status.enabled", await self._session.capabilities) result = await self._session.ocs("GET", f"{self._ep_base}/predefined_statuses") return [PredefinedStatus(i) for i in result] async def set_predefined(self, status_id: str, clear_at: int = 0) -> None: """Set predefined status for the current user. :param status_id: ``predefined`` status ID. :param clear_at: *optional* time in seconds before the status is cleared. """ if (await self._session.nc_version)["major"] < 27: return require_capabilities("user_status.enabled", await self._session.capabilities) params: dict[str, int | str] = {"messageId": status_id} if clear_at: params["clearAt"] = clear_at await self._session.ocs("PUT", f"{self._ep_base}/user_status/message/predefined", params=params) async def set_status_type(self, value: typing.Literal["online", "away", "dnd", "invisible", "offline"]) -> None: """Sets the status type for the current user.""" require_capabilities("user_status.enabled", await self._session.capabilities) await self._session.ocs("PUT", f"{self._ep_base}/user_status/status", params={"statusType": value}) async def set_status(self, message: str | None = None, clear_at: int = 0, status_icon: str = "") -> None: """Sets current user status. :param message: Message text to set in the status. :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.enabled", await self._session.capabilities) if message is None: await self._session.ocs("DELETE", f"{self._ep_base}/user_status/message") return if status_icon: require_capabilities("user_status.supports_emoji", await self._session.capabilities) params: dict[str, int | str] = {"message": message} if clear_at: params["clearAt"] = clear_at if status_icon: params["statusIcon"] = status_icon await self._session.ocs("PUT", f"{self._ep_base}/user_status/message/custom", params=params) async def get_backup_status(self, user_id: str = "") -> UserStatus | None: """Get the backup status of the user if any.""" require_capabilities("user_status.enabled", await self._session.capabilities) user_id = user_id if user_id else await self._session.user if not user_id: raise ValueError("user_id can not be empty.") return await self.get(f"_{user_id}") async def restore_backup_status(self, status_id: str) -> CurrentUserStatus | None: """Restores the backup state as current for the current user.""" require_capabilities("user_status.enabled", await self._session.capabilities) require_capabilities("user_status.restore", await self._session.capabilities) result = await self._session.ocs("DELETE", f"{self._ep_base}/user_status/revert/{status_id}") return result if result else None