"""Nextcloud class providing access to all API endpoints."""
import contextlib
import typing
from abc import ABC
from httpx import Headers
from ._exceptions import NextcloudExceptionNotFound
from ._misc import check_capabilities, require_capabilities
from ._preferences import AsyncPreferencesAPI, PreferencesAPI
from ._preferences_ex import (
AppConfigExAPI,
AsyncAppConfigExAPI,
AsyncPreferencesExAPI,
PreferencesExAPI,
)
from ._session import (
AppConfig,
AsyncNcSession,
AsyncNcSessionApp,
AsyncNcSessionBasic,
NcSession,
NcSessionApp,
NcSessionBasic,
ServerVersion,
)
from ._talk_api import _AsyncTalkAPI, _TalkAPI
from ._theming import ThemingInfo, get_parsed_theme
from .activity import _ActivityAPI, _AsyncActivityAPI
from .apps import _AppsAPI, _AsyncAppsAPI
from .calendar_api import _CalendarAPI
from .ex_app.defs import LogLvl
from .ex_app.events_listener import AsyncEventsListenerAPI, EventsListenerAPI
from .ex_app.occ_commands import AsyncOccCommandsAPI, OccCommandsAPI
from .ex_app.providers.providers import AsyncProvidersApi, ProvidersApi
from .ex_app.ui.ui import AsyncUiApi, UiApi
from .files.files import FilesAPI
from .files.files_async import AsyncFilesAPI
from .loginflow_v2 import _AsyncLoginFlowV2API, _LoginFlowV2API
from .notes import _AsyncNotesAPI, _NotesAPI
from .notifications import _AsyncNotificationsAPI, _NotificationsAPI
from .user_status import _AsyncUserStatusAPI, _UserStatusAPI
from .users import _AsyncUsersAPI, _UsersAPI
from .users_groups import _AsyncUsersGroupsAPI, _UsersGroupsAPI
from .weather_status import _AsyncWeatherStatusAPI, _WeatherStatusAPI
from .webhooks import _AsyncWebhooksAPI, _WebhooksAPI
class _NextcloudBasic(ABC): # pylint: disable=too-many-instance-attributes
apps: _AppsAPI
"""Nextcloud API for App management"""
activity: _ActivityAPI
"""Activity Application API"""
cal: _CalendarAPI
"""Nextcloud Calendar API"""
files: FilesAPI
"""Nextcloud API for File System and Files Sharing"""
preferences: PreferencesAPI
"""Nextcloud User Preferences API"""
notes: _NotesAPI
"""Nextcloud Notes API"""
notifications: _NotificationsAPI
"""Nextcloud API for managing user notifications"""
talk: _TalkAPI
"""Nextcloud Talk API"""
users: _UsersAPI
"""Nextcloud API for managing users."""
users_groups: _UsersGroupsAPI
"""Nextcloud API for managing user groups."""
user_status: _UserStatusAPI
"""Nextcloud API for managing users statuses"""
weather_status: _WeatherStatusAPI
"""Nextcloud API for managing user weather statuses"""
webhooks: _WebhooksAPI
"""Nextcloud API for managing webhooks"""
_session: NcSessionBasic
def __init__(self, session: NcSessionBasic):
self.apps = _AppsAPI(session)
self.activity = _ActivityAPI(session)
self.cal = _CalendarAPI(session)
self.files = FilesAPI(session)
self.preferences = PreferencesAPI(session)
self.notes = _NotesAPI(session)
self.notifications = _NotificationsAPI(session)
self.talk = _TalkAPI(session)
self.users = _UsersAPI(session)
self.users_groups = _UsersGroupsAPI(session)
self.user_status = _UserStatusAPI(session)
self.weather_status = _WeatherStatusAPI(session)
self.webhooks = _WebhooksAPI(session)
@property
def capabilities(self) -> dict:
"""Returns the capabilities of the Nextcloud instance."""
return self._session.capabilities
@property
def srv_version(self) -> ServerVersion:
"""Returns dictionary with the server version."""
return self._session.nc_version
def check_capabilities(self, capabilities: str | list[str]) -> list[str]:
"""Returns the list with missing capabilities if any."""
return check_capabilities(capabilities, self.capabilities)
def update_server_info(self) -> None:
"""Updates the capabilities and the Nextcloud version.
*In normal cases, it is called automatically and there is no need to call it manually.*
"""
self._session.update_server_info()
@property
def response_headers(self) -> Headers:
"""Returns the `HTTPX headers <https://www.python-httpx.org/api/#headers>`_ from the last response."""
return self._session.response_headers
@property
def theme(self) -> ThemingInfo | None:
"""Returns Theme information."""
return get_parsed_theme(self.capabilities["theming"]) if "theming" in self.capabilities else None
def perform_login(self) -> bool:
"""Performs login into Nextcloud if not already logged in; manual invocation of this method is unnecessary."""
try:
self.update_server_info()
except Exception: # noqa pylint: disable=broad-exception-caught
return False
return True
def ocs(
self,
method: str,
path: str,
*,
content: bytes | str | typing.Iterable[bytes] | typing.AsyncIterable[bytes] | None = None,
json: dict | list | None = None,
response_type: str | None = None,
params: dict | None = None,
**kwargs,
):
"""Performs OCS call and returns OCS response payload data."""
return self._session.ocs(
method, path, content=content, json=json, response_type=response_type, params=params, **kwargs
)
def download_log(self, fp) -> None:
"""Downloads Nextcloud log file. Requires Admin privileges."""
self._session.download2stream("/index.php/settings/admin/log/download", fp)
class _AsyncNextcloudBasic(ABC): # pylint: disable=too-many-instance-attributes
apps: _AsyncAppsAPI
"""Nextcloud API for App management"""
activity: _AsyncActivityAPI
"""Activity Application API"""
# cal: _CalendarAPI
# """Nextcloud Calendar API"""
files: AsyncFilesAPI
"""Nextcloud API for File System and Files Sharing"""
preferences: AsyncPreferencesAPI
"""Nextcloud User Preferences API"""
notes: _AsyncNotesAPI
"""Nextcloud Notes API"""
notifications: _AsyncNotificationsAPI
"""Nextcloud API for managing user notifications"""
talk: _AsyncTalkAPI
"""Nextcloud Talk API"""
users: _AsyncUsersAPI
"""Nextcloud API for managing users."""
users_groups: _AsyncUsersGroupsAPI
"""Nextcloud API for managing user groups."""
user_status: _AsyncUserStatusAPI
"""Nextcloud API for managing users statuses"""
weather_status: _AsyncWeatherStatusAPI
"""Nextcloud API for managing user weather statuses"""
webhooks: _AsyncWebhooksAPI
"""Nextcloud API for managing webhooks"""
_session: AsyncNcSessionBasic
def __init__(self, session: AsyncNcSessionBasic):
self.apps = _AsyncAppsAPI(session)
self.activity = _AsyncActivityAPI(session)
# self.cal = _CalendarAPI(session)
self.files = AsyncFilesAPI(session)
self.preferences = AsyncPreferencesAPI(session)
self.notes = _AsyncNotesAPI(session)
self.notifications = _AsyncNotificationsAPI(session)
self.talk = _AsyncTalkAPI(session)
self.users = _AsyncUsersAPI(session)
self.users_groups = _AsyncUsersGroupsAPI(session)
self.user_status = _AsyncUserStatusAPI(session)
self.weather_status = _AsyncWeatherStatusAPI(session)
self.webhooks = _AsyncWebhooksAPI(session)
@property
async def capabilities(self) -> dict:
"""Returns the capabilities of the Nextcloud instance."""
return await self._session.capabilities
@property
async def srv_version(self) -> ServerVersion:
"""Returns dictionary with the server version."""
return await self._session.nc_version
async def check_capabilities(self, capabilities: str | list[str]) -> list[str]:
"""Returns the list with missing capabilities if any."""
return check_capabilities(capabilities, await self.capabilities)
async def update_server_info(self) -> None:
"""Updates the capabilities and the Nextcloud version.
*In normal cases, it is called automatically and there is no need to call it manually.*
"""
await self._session.update_server_info()
@property
def response_headers(self) -> Headers:
"""Returns the `HTTPX headers <https://www.python-httpx.org/api/#headers>`_ from the last response."""
return self._session.response_headers
@property
async def theme(self) -> ThemingInfo | None:
"""Returns Theme information."""
return get_parsed_theme((await self.capabilities)["theming"]) if "theming" in await self.capabilities else None
async def perform_login(self) -> bool:
"""Performs login into Nextcloud if not already logged in; manual invocation of this method is unnecessary."""
try:
await self.update_server_info()
except Exception: # noqa pylint: disable=broad-exception-caught
return False
return True
async def ocs(
self,
method: str,
path: str,
*,
content: bytes | str | typing.Iterable[bytes] | typing.AsyncIterable[bytes] | None = None,
json: dict | list | None = None,
response_type: str | None = None,
params: dict | None = None,
**kwargs,
):
"""Performs OCS call and returns OCS response payload data."""
return await self._session.ocs(
method, path, content=content, json=json, response_type=response_type, params=params, **kwargs
)
async def download_log(self, fp) -> None:
"""Downloads Nextcloud log file. Requires Admin privileges."""
await self._session.download2stream("/index.php/settings/admin/log/download", fp)
[docs]
class Nextcloud(_NextcloudBasic):
"""Nextcloud client class.
Allows you to connect to Nextcloud and perform operations on files, shares, users, and everything else.
"""
_session: NcSession
loginflow_v2: _LoginFlowV2API
"""Nextcloud Login flow v2."""
[docs]
def __init__(self, **kwargs):
"""If the parameters are not specified, they will be taken from the environment.
:param nextcloud_url: url of the nextcloud instance.
:param nc_auth_user: login username. Optional.
:param nc_auth_pass: password or app-password for the username. Optional.
"""
self._session = NcSession(**kwargs)
self.loginflow_v2 = _LoginFlowV2API(self._session)
super().__init__(self._session)
@property
def user(self) -> str:
"""Returns current user ID."""
return self._session.user
class AsyncNextcloud(_AsyncNextcloudBasic):
"""Async Nextcloud client class.
Allows you to connect to Nextcloud and perform operations on files, shares, users, and everything else.
"""
_session: AsyncNcSession
loginflow_v2: _AsyncLoginFlowV2API
"""Nextcloud Login flow v2."""
def __init__(self, **kwargs):
"""If the parameters are not specified, they will be taken from the environment.
:param nextcloud_url: url of the nextcloud instance.
:param nc_auth_user: login username. Optional.
:param nc_auth_pass: password or app-password for the username. Optional.
"""
self._session = AsyncNcSession(**kwargs)
self.loginflow_v2 = _AsyncLoginFlowV2API(self._session)
super().__init__(self._session)
@property
async def user(self) -> str:
"""Returns current user ID."""
return await self._session.user
[docs]
class NextcloudApp(_NextcloudBasic):
"""Class for communication with Nextcloud in Nextcloud applications.
Provides additional API required for applications such as user impersonation,
endpoint registration, new authentication method, etc.
.. note:: Instance of this class should not be created directly in ``normal`` applications,
it will be provided for each app endpoint call.
"""
_session: NcSessionApp
appconfig_ex: AppConfigExAPI
"""Nextcloud App Preferences API for ExApps"""
preferences_ex: PreferencesExAPI
"""Nextcloud User Preferences API for ExApps"""
ui: UiApi
"""Nextcloud UI API for ExApps"""
providers: ProvidersApi
"""API for registering providers for Nextcloud"""
events_listener: EventsListenerAPI
"""API for registering Events listeners for ExApps"""
occ_commands: OccCommandsAPI
"""API for registering OCC command for ExApps"""
def __init__(self, **kwargs):
"""The parameters will be taken from the environment.
They can be overridden by specifying them in **kwargs**, but this behavior is highly discouraged.
"""
self._session = NcSessionApp(**kwargs)
super().__init__(self._session)
self.appconfig_ex = AppConfigExAPI(self._session)
self.preferences_ex = PreferencesExAPI(self._session)
self.ui = UiApi(self._session)
self.providers = ProvidersApi(self._session)
self.events_listener = EventsListenerAPI(self._session)
self.occ_commands = OccCommandsAPI(self._session)
@property
def enabled_state(self) -> bool:
"""Returns ``True`` if ExApp is enabled, ``False`` otherwise."""
with contextlib.suppress(Exception):
return bool(self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
return False
[docs]
def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
"""Writes log to the Nextcloud log file."""
int_log_lvl = int(log_lvl)
if int_log_lvl < 0 or int_log_lvl > 4:
raise ValueError("Invalid `log_lvl` value")
if not fast_send:
if self.check_capabilities("app_api"):
return
if int_log_lvl < self.capabilities["app_api"].get("loglevel", 0):
return
with contextlib.suppress(Exception):
self._session.ocs("POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content})
[docs]
def users_list(self) -> list[str]:
"""Returns list of users on the Nextcloud instance."""
return self._session.ocs("GET", f"{self._session.ae_url}/users")
@property
def user(self) -> str:
"""Property containing the current user ID.
**ExApps** can change user ID they impersonate with **set_user** method.
"""
return self._session.user
[docs]
def set_user(self, user_id: str):
"""Changes current User ID."""
if self._session.user != user_id:
self._session.set_user(user_id)
self.talk.config_sha = ""
self.talk.modified_since = 0
self.activity.last_given = 0
self.notes.last_etag = ""
self._session.update_server_info()
@property
def app_cfg(self) -> AppConfig:
"""Returns deploy config, with AppAPI version, Application version and name."""
return self._session.cfg
[docs]
def register_talk_bot(self, callback_url: str, display_name: str, description: str = "") -> tuple[str, str]:
"""Registers Talk BOT.
.. note:: AppAPI will add a record in a case of successful registration to the ``appconfig_ex`` table.
:param callback_url: URL suffix for fetching new messages. MUST be ``UNIQ`` for each bot the app provides.
:param display_name: The name under which the messages will be posted.
:param description: Optional description shown in the admin settings.
:return: Tuple with ID and the secret used for signing requests.
"""
require_capabilities("app_api", self._session.capabilities)
require_capabilities("spreed.features.bots-v1", self._session.capabilities)
params = {
"name": display_name,
"route": callback_url,
"description": description,
}
result = self._session.ocs("POST", f"{self._session.ae_url}/talk_bot", json=params)
return result["id"], result["secret"]
[docs]
def unregister_talk_bot(self, callback_url: str) -> bool:
"""Unregisters Talk BOT."""
require_capabilities("app_api", self._session.capabilities)
require_capabilities("spreed.features.bots-v1", self._session.capabilities)
params = {
"route": callback_url,
}
try:
self._session.ocs("DELETE", f"{self._session.ae_url}/talk_bot", json=params)
except NextcloudExceptionNotFound:
return False
return True
[docs]
def set_init_status(self, progress: int, error: str = "") -> None:
"""Sets state of the app initialization.
:param progress: a number from ``0`` to ``100`` indicating the percentage of application readiness for work.
After sending ``100`` AppAPI will enable the application.
:param error: if non-empty, signals to AppAPI that the application cannot be initialized successfully.
"""
self._session.ocs(
"PUT",
f"/ocs/v1.php/apps/app_api/apps/status/{self._session.cfg.app_name}",
json={
"progress": progress,
"error": error,
},
)
class AsyncNextcloudApp(_AsyncNextcloudBasic):
"""Class for communication with Nextcloud in Async Nextcloud applications.
Provides additional API required for applications such as user impersonation,
endpoint registration, new authentication method, etc.
.. note:: Instance of this class should not be created directly in ``normal`` applications,
it will be provided for each app endpoint call.
"""
_session: AsyncNcSessionApp
appconfig_ex: AsyncAppConfigExAPI
"""Nextcloud App Preferences API for ExApps"""
preferences_ex: AsyncPreferencesExAPI
"""Nextcloud User Preferences API for ExApps"""
ui: AsyncUiApi
"""Nextcloud UI API for ExApps"""
providers: AsyncProvidersApi
"""API for registering providers for Nextcloud"""
events_listener: AsyncEventsListenerAPI
"""API for registering Events listeners for ExApps"""
occ_commands: AsyncOccCommandsAPI
"""API for registering OCC command for ExApps"""
def __init__(self, **kwargs):
"""The parameters will be taken from the environment.
They can be overridden by specifying them in **kwargs**, but this behavior is highly discouraged.
"""
self._session = AsyncNcSessionApp(**kwargs)
super().__init__(self._session)
self.appconfig_ex = AsyncAppConfigExAPI(self._session)
self.preferences_ex = AsyncPreferencesExAPI(self._session)
self.ui = AsyncUiApi(self._session)
self.providers = AsyncProvidersApi(self._session)
self.events_listener = AsyncEventsListenerAPI(self._session)
self.occ_commands = AsyncOccCommandsAPI(self._session)
@property
async def enabled_state(self) -> bool:
"""Returns ``True`` if ExApp is enabled, ``False`` otherwise."""
with contextlib.suppress(Exception):
return bool(await self._session.ocs("GET", "/ocs/v1.php/apps/app_api/ex-app/state"))
return False
async def log(self, log_lvl: LogLvl, content: str, fast_send: bool = False) -> None:
"""Writes log to the Nextcloud log file."""
int_log_lvl = int(log_lvl)
if int_log_lvl < 0 or int_log_lvl > 4:
raise ValueError("Invalid `log_lvl` value")
if not fast_send:
if await self.check_capabilities("app_api"):
return
if int_log_lvl < (await self.capabilities)["app_api"].get("loglevel", 0):
return
with contextlib.suppress(Exception):
await self._session.ocs(
"POST", f"{self._session.ae_url}/log", json={"level": int_log_lvl, "message": content}
)
async def users_list(self) -> list[str]:
"""Returns list of users on the Nextcloud instance."""
return await self._session.ocs("GET", f"{self._session.ae_url}/users")
@property
async def user(self) -> str:
"""Property containing the current user ID.
**ExApps** can change user ID they impersonate with **set_user** method.
"""
return await self._session.user
async def set_user(self, user_id: str):
"""Changes current User ID."""
if await self._session.user != user_id:
self._session.set_user(user_id)
self.talk.config_sha = ""
self.talk.modified_since = 0
self.activity.last_given = 0
self.notes.last_etag = ""
await self._session.update_server_info()
@property
def app_cfg(self) -> AppConfig:
"""Returns deploy config, with AppAPI version, Application version and name."""
return self._session.cfg
async def register_talk_bot(self, callback_url: str, display_name: str, description: str = "") -> tuple[str, str]:
"""Registers Talk BOT.
.. note:: AppAPI will add a record in a case of successful registration to the ``appconfig_ex`` table.
:param callback_url: URL suffix for fetching new messages. MUST be ``UNIQ`` for each bot the app provides.
:param display_name: The name under which the messages will be posted.
:param description: Optional description shown in the admin settings.
:return: Tuple with ID and the secret used for signing requests.
"""
require_capabilities("app_api", await self._session.capabilities)
require_capabilities("spreed.features.bots-v1", await self._session.capabilities)
params = {
"name": display_name,
"route": callback_url,
"description": description,
}
result = await self._session.ocs("POST", f"{self._session.ae_url}/talk_bot", json=params)
return result["id"], result["secret"]
async def unregister_talk_bot(self, callback_url: str) -> bool:
"""Unregisters Talk BOT."""
require_capabilities("app_api", await self._session.capabilities)
require_capabilities("spreed.features.bots-v1", await self._session.capabilities)
params = {
"route": callback_url,
}
try:
await self._session.ocs("DELETE", f"{self._session.ae_url}/talk_bot", json=params)
except NextcloudExceptionNotFound:
return False
return True
async def set_init_status(self, progress: int, error: str = "") -> None:
"""Sets state of the app initialization.
:param progress: a number from ``0`` to ``100`` indicating the percentage of application readiness for work.
After sending ``100`` AppAPI will enable the application.
:param error: if non-empty, signals to AppAPI that the application cannot be initialized successfully.
"""
await self._session.ocs(
"PUT",
f"/ocs/v1.php/apps/app_api/apps/status/{self._session.cfg.app_name}",
json={
"progress": progress,
"error": error,
},
)