added Nextcloud.response_headers property (#81)

This will allow to implement Talk API, as we need
`X-Nextcloud-Talk-Hash` and `X-Nextcloud-Talk-Modified-Before` headers.

---------

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.0.30 - 2023-08-17]
+
+### Added
+
+- `Nextcloud.response_headers` property, to get headers from last response.
+
 ## [0.0.29 - 2023-08-13]
 
 ### Added

docs/conf.py

@@ -20,7 +20,8 @@
     "sphinx_copybutton",
     "sphinx_inline_tabs",
     "sphinx_issues",
-    "sphinx_rtd_theme"]
+    "sphinx_rtd_theme",
+]
 
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
@@ -57,7 +58,10 @@
 # Default is False. You can activate this mode temporarily using the -n command-line
 # switch.
 nitpicky = True
-nitpick_ignore_regex = [(r"py:class", r"starlette\.requests\.Request")]
+nitpick_ignore_regex = [
+    (r"py:class", r"starlette\.requests\.Request"),
+    (r"py:.*", r"httpx.*"),
+]
 
 autodoc_member_order = "bysource"
 

nc_py_api/_session.py

@@ -14,7 +14,9 @@
 from urllib.parse import quote, urlencode
 
 from fastapi import Request
-from httpx import Client, Limits, ReadTimeout, Response
+from httpx import Client
+from httpx import Headers as HttpxHeaders
+from httpx import Limits, ReadTimeout, Response
 
 try:
     from xxhash import xxh64
@@ -137,6 +139,7 @@ class NcSessionBasic(ABC):
     user: str
     custom_headers: dict
     _capabilities: dict
+    response_headers: HttpxHeaders
 
     @abstractmethod
     def __init__(self, **kwargs):
@@ -145,6 +148,7 @@ def __init__(self, **kwargs):
         self.custom_headers = kwargs.get("headers", {})
         self.limits = Limits(max_keepalive_connections=20, max_connections=20, keepalive_expiry=60.0)
         self.init_adapter()
+        self.response_headers = HttpxHeaders()
 
     def __del__(self):
         if hasattr(self, "adapter") and self.adapter:
@@ -200,6 +204,7 @@ def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[byte
         except ReadTimeout:
             raise NextcloudException(408, info=info) from None
 
+        self.response_headers = response.headers
         check_error(response.status_code, info)
         response_data = loads(response.text)
         ocs_meta = response_data["ocs"]["meta"]
@@ -235,11 +240,12 @@ def dav_stream(
 
     def _dav(self, method: str, path: str, headers: dict, data: Optional[bytes], **kwargs) -> Response:
         self.init_adapter()
-        # self.cfg.
         timeout = kwargs.pop("timeout", self.cfg.options.timeout_dav)
-        return self.adapter.request(
+        result = self.adapter.request(
             method, self.cfg.endpoint + path, headers=headers, content=data, timeout=timeout, **kwargs
         )
+        self.response_headers = result.headers
+        return result
 
     def _dav_stream(self, method: str, path: str, headers: dict, data: Optional[bytes], **kwargs) -> Iterator[Response]:
         self.init_adapter()

nc_py_api/files/files.py

@@ -162,6 +162,7 @@ def download_directory_as_zip(
         with self._session.get_stream(
             "/index.php/apps/files/ajax/download.php", params={"dir": path}
         ) as response:  # type: ignore
+            self._session.response_headers = response.headers
             check_error(response.status_code, f"download_directory_as_zip: user={self._session.user}, path={path}")
             result_path = local_path if local_path else os.path.basename(path)
             with open(
@@ -446,6 +447,7 @@ def __download2stream(self, path: str, fp, **kwargs) -> None:
         with self._session.dav_stream(
             "GET", self._dav_get_obj_path(self._session.user, path)
         ) as response:  # type: ignore
+            self._session.response_headers = response.headers
             check_error(response.status_code, f"download_stream: user={self._session.user}, path={path}")
             for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", 4 * 1024 * 1024)):
                 fp.write(data_chunk)

nc_py_api/nextcloud.py

@@ -3,6 +3,7 @@
 from typing import Optional, Union
 
 from fastapi import Request
+from httpx import Headers as HttpxHeaders
 
 from ._misc import check_capabilities
 from ._session import AppConfig, NcSession, NcSessionApp, NcSessionBasic, ServerVersion
@@ -57,6 +58,11 @@ def update_server_info(self) -> None:
         """
         self._session.update_server_info()
 
+    @property
+    def response_headers(self) -> HttpxHeaders:
+        """Returns the `HTTPX headers <https://www.python-httpx.org/api/#headers>`_ from the last response."""
+        return self._session.response_headers
+
     @property
     def theme(self) -> Optional[ThemingInfo]:
         """Returns Theme information."""

pyproject.toml

@@ -105,7 +105,7 @@ preview = true
 line-length = 120
 target-version = "py39"
 select = ["A", "B", "C", "D", "E", "F", "G", "I", "UP", "SIM", "Q", "W"]
-extend-ignore = ["D107", "D105", "D203", "D213", "D401"]
+extend-ignore = ["D107", "D105", "D203", "D213", "D401", "I001"]
 
 [tool.ruff.per-file-ignores]
 "nc_py_api/__init__.py" = ["F401"]

tests/files_test.py

@@ -115,10 +115,12 @@ def test_file_download2stream(nc, data_type, chunk_size):
     srv_admin_manual_buf = MyBytesIO()
     content = "".join(choice(ascii_lowercase) for _ in range(64)) if data_type == "str" else randbytes(64)
     nc.files.upload("test_file.txt", content=content)
+    old_headers = nc.response_headers
     if chunk_size is not None:
         nc.files.download2stream("/test_file.txt", srv_admin_manual_buf, chunk_size=chunk_size)
     else:
         nc.files.download2stream("/test_file.txt", srv_admin_manual_buf)
+    assert nc.response_headers != old_headers
     assert nc.files.download("test_file.txt") == srv_admin_manual_buf.getbuffer()
     if chunk_size is None:
         assert srv_admin_manual_buf.n_write_calls == 1
@@ -542,7 +544,9 @@ def test_download_as_zip(nc):
         nc.files.upload("test_root_folder/0.txt", content="")
         nc.files.upload("test_root_folder/1.txt", content="123")
         nc.files.upload("test_root_folder/test_subfolder/0.txt", content="")
+        old_headers = nc.response_headers
         result = nc.files.download_directory_as_zip("test_root_folder")
+        assert nc.response_headers != old_headers
         try:
             with zipfile.ZipFile(result, "r") as zip_ref:
                 assert zip_ref.filelist[0].filename == "test_root_folder/"
@@ -560,7 +564,9 @@ def test_download_as_zip(nc):
                 assert len(zip_ref.filelist) == 6
         finally:
             os.remove(result)
+        old_headers = nc.response_headers
         result = nc.files.download_directory_as_zip("test_root_folder/test_subfolder", "2.zip")
+        assert nc.response_headers != old_headers
         try:
             assert str(result) == "2.zip"
             with zipfile.ZipFile(result, "r") as zip_ref:

tests/misc_test.py

@@ -1,5 +1,5 @@
 import pytest
-from gfixture import NC_APP
+from gfixture import NC_APP, NC_TO_TEST
 
 from nc_py_api import NextcloudException
 from nc_py_api._deffered_error import DeferredError  # noqa
@@ -53,3 +53,10 @@ def test_deffered_error():
 
     with pytest.raises(ModuleNotFoundError):
         unknown_non_exist_module.some_class_or_func()
+
+
+@pytest.mark.parametrize("nc", NC_TO_TEST)
+def test_ocs_response_headers(nc):
+    old_headers = nc.response_headers
+    nc.users.get_details()
+    assert old_headers != nc.response_headers