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 <bigcat88@icloud.com>

Author: bigcat88

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