New download_log method (#199)

Ref:


https://docs.nextcloud.com/server/latest/developer_manual/_static/openapi.html#/operations/settings-log_settings-download

Signed-off-by: Alexander Piskun <[email protected]>

CHANGELOG.md

@@ -6,8 +6,9 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
+- `download_log` method to download `nextcloud.log`. #199
 - NextcloudApp: API for registering Speech to Text providers(*avalaible from Nextcloud 29*). #196
-- NextcloudApp: API for registering Text Processing providers(*avalaible from Nextcloud 29*). #197
+- NextcloudApp: API for registering Text Processing providers(*avalaible from Nextcloud 29*). #198
 
 ### Fixed
 

nc_py_api/_session.py

@@ -1,5 +1,7 @@
 """Session represents one connection to Nextcloud. All related stuff for these live here."""
 
+import builtins
+import pathlib
 import re
 import typing
 from abc import ABC, abstractmethod
@@ -250,6 +252,15 @@ def user(self) -> str:
  def set_user(self, user_id: str) -> None:
  self._user = user_id
 
+ def download2stream(self, url_path: str, fp, dav: bool = False, **kwargs):
+ if isinstance(fp, str | pathlib.Path):
+ with builtins.open(fp, "wb") as f:
+ self._download2fp(url_path, f, dav, **kwargs)
+ elif hasattr(fp, "write"):
+ self._download2fp(url_path, fp, dav, **kwargs)
+ else:
+ raise TypeError("`fp` must be a path to file or an object with `write` method.")
+
  def _get_adapter_kwargs(self, dav: bool) -> dict[str, typing.Any]:
  if dav:
  return {
@@ -276,6 +287,13 @@ def _response_event(self, response: Response) -> None:
  return
  self.response_headers = response.headers
 
+ def _download2fp(self, url_path: str, fp, dav: bool, **kwargs):
+ adapter = self.adapter_dav if dav else self.adapter
+ with adapter.stream("GET", url_path) as response:
+ check_error(response)
+ for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
+ fp.write(data_chunk)
+
 
 class AsyncNcSessionBasic(NcSessionBase, ABC):
  adapter: AsyncClient
@@ -354,6 +372,15 @@ async def user(self) -> str:
  def set_user(self, user: str) -> None:
  self._user = user
 
+ async def download2stream(self, url_path: str, fp, dav: bool = False, **kwargs):
+ if isinstance(fp, str | pathlib.Path):
+ with builtins.open(fp, "wb") as f:
+ await self._download2fp(url_path, f, dav, **kwargs)
+ elif hasattr(fp, "write"):
+ await self._download2fp(url_path, fp, dav, **kwargs)
+ else:
+ raise TypeError("`fp` must be a path to file or an object with `write` method.")
+
  def _get_adapter_kwargs(self, dav: bool) -> dict[str, typing.Any]:
  if dav:
  return {
@@ -380,6 +407,13 @@ async def _response_event(self, response: Response) -> None:
  return
  self.response_headers = response.headers
 
+ async def _download2fp(self, url_path: str, fp, dav: bool, **kwargs):
+ adapter = self.adapter_dav if dav else self.adapter
+ async with adapter.stream("GET", url_path) as response:
+ check_error(response)
+ async for data_chunk in response.aiter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
+ fp.write(data_chunk)
+
 
 class NcSession(NcSessionBasic):
  cfg: Config

nc_py_api/files/files.py

@@ -98,14 +98,8 @@ def download2stream(self, path: str | FsNode, fp, **kwargs) -> None:
  The object must implement the ``file.write`` method and be able to write binary data.
  :param kwargs: **chunk_size** an int value specifying chunk size to write. Default = **5Mb**
  """
- path = path.user_path if isinstance(path, FsNode) else path
- if isinstance(fp, str | Path):
- with builtins.open(fp, "wb") as f:
- self.__download2stream(path, f, **kwargs)
- elif hasattr(fp, "write"):
- self.__download2stream(path, fp, **kwargs)
- else:
- raise TypeError("`fp` must be a path to file or an object with `write` method.")
+ path = quote(dav_get_obj_path(self._session.user, path.user_path if isinstance(path, FsNode) else path))
+ self._session.download2stream(path, fp, dav=True, **kwargs)
 
  def download_directory_as_zip(self, path: str | FsNode, local_path: str | Path | None = None, **kwargs) -> Path:
  """Downloads a remote directory as zip archive.
@@ -439,12 +433,6 @@ def _listdir(
  self._session.cfg.dav_url_suffix, webdav_response, user, path, properties, exclude_self, prop_type
  )
 
- def __download2stream(self, path: str, fp, **kwargs) -> None:
- with self._session.adapter_dav.stream("GET", quote(dav_get_obj_path(self._session.user, path))) as response:
- check_error(response, f"download_stream: user={self._session.user}, path={path}")
- for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
- fp.write(data_chunk)
-
  def __upload_stream(self, path: str, fp, chunk_size: int) -> FsNode:
  _tmp_path = "nc-py-api-" + random_string(56)
  _dav_path = quote(dav_get_obj_path(self._session.user, _tmp_path, root_path="/uploads"))
@@ -561,14 +549,8 @@ async def download2stream(self, path: str | FsNode, fp, **kwargs) -> None:
  The object must implement the ``file.write`` method and be able to write binary data.
  :param kwargs: **chunk_size** an int value specifying chunk size to write. Default = **5Mb**
  """
- path = path.user_path if isinstance(path, FsNode) else path
- if isinstance(fp, str | Path):
- with builtins.open(fp, "wb") as f:
- await self.__download2stream(path, f, **kwargs)
- elif hasattr(fp, "write"):
- await self.__download2stream(path, fp, **kwargs)
- else:
- raise TypeError("`fp` must be a path to file or an object with `write` method.")
+ path = quote(dav_get_obj_path(await self._session.user, path.user_path if isinstance(path, FsNode) else path))
+ await self._session.download2stream(path, fp, dav=True, **kwargs)
 
  async def download_directory_as_zip(
  self, path: str | FsNode, local_path: str | Path | None = None, **kwargs
@@ -909,14 +891,6 @@ async def _listdir(
  self._session.cfg.dav_url_suffix, webdav_response, user, path, properties, exclude_self, prop_type
  )
 
- async def __download2stream(self, path: str, fp, **kwargs) -> None:
- async with self._session.adapter_dav.stream(
- "GET", quote(dav_get_obj_path(await self._session.user, path))
- ) as response:
- check_error(response, f"download_stream: user={await self._session.user}, path={path}")
- async for data_chunk in response.aiter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
- fp.write(data_chunk)
-
  async def __upload_stream(self, path: str, fp, chunk_size: int) -> FsNode:
  _tmp_path = "nc-py-api-" + random_string(56)
  _dav_path = quote(dav_get_obj_path(await self._session.user, _tmp_path, root_path="/uploads"))

nc_py_api/nextcloud.py

@@ -135,6 +135,10 @@ def ocs(
  """Performs OCS call and returns OCS response payload data."""
  return self._session.ocs(method, path, content=content, json=json, 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
@@ -229,6 +233,10 @@ async def ocs(
  """Performs OCS call and returns OCS response payload data."""
  return await self._session.ocs(method, path, content=content, json=json, 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)
+
 
 class Nextcloud(_NextcloudBasic):
  """Nextcloud client class.

tests/actual_tests/misc_test.py

@@ -1,4 +1,5 @@
 import datetime
+import io
 import os
 
 import pytest
@@ -206,3 +207,16 @@ async def test_perform_login_async(anc_any):
  assert not new_nc._session._capabilities
  await new_nc.perform_login()
  assert new_nc._session._capabilities
+
+
+def test_download_log(nc_any):
+ buf = io.BytesIO()
+ nc_any.download_log(buf)
+ assert buf.tell() > 0
+
+
+@pytest.mark.asyncio(scope="session")
+async def test_download_log_async(anc_any):
+ buf = io.BytesIO()
+ await anc_any.download_log(buf)
+ assert buf.tell() > 0