From 6e9c581b8bf335d898042cb704f370acc0dd9a16 Mon Sep 17 00:00:00 2001 From: Review Date: Tue, 24 Mar 2026 08:33:33 +0000 Subject: [PATCH 1/2] feat: expose S3 presigned download URL on FsNode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add download_url and download_url_expiration properties to FsNodeInfo. The oc:downloadURL property was already requested via PROPFIND but never parsed — now it is. Also request nc:download-url-expiration property. When the Nextcloud storage backend is S3 with use_presigned_url enabled, these provide a direct download URL bypassing the Nextcloud proxy. Returns empty string / zero when not available (non-S3 backends). Closes #418 --- nc_py_api/files/__init__.py | 16 ++++++++++++++++ nc_py_api/files/_files.py | 7 ++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/nc_py_api/files/__init__.py b/nc_py_api/files/__init__.py index 1968b713..98854f8e 100644 --- a/nc_py_api/files/__init__.py +++ b/nc_py_api/files/__init__.py @@ -107,6 +107,8 @@ def __init__(self, **kwargs): self.creation_date = kwargs.get("creation_date", datetime.datetime(1970, 1, 1)) except (ValueError, TypeError): self.creation_date = datetime.datetime(1970, 1, 1) + self._download_url: str = kwargs.get("download_url", "") + self._download_url_expiration: int = kwargs.get("download_url_expiration", 0) self._trashbin: dict[str, str | int] = {} for i in ("trashbin_filename", "trashbin_original_location", "trashbin_deletion_time"): if i in kwargs: @@ -132,6 +134,20 @@ def permissions(self) -> str: """Permissions for the object.""" return self._raw_data["permissions"] + @property + def download_url(self) -> str: + """S3 presigned URL for direct download, bypassing Nextcloud. + + Only available when the storage backend is S3 with ``use_presigned_url`` enabled. + Empty string when not available. + """ + return self._download_url + + @property + def download_url_expiration(self) -> int: + """Expiration timestamp for :py:attr:`download_url`. Zero when not available.""" + return self._download_url_expiration + @property def last_modified(self) -> datetime.datetime: """Time when the object was last modified. diff --git a/nc_py_api/files/_files.py b/nc_py_api/files/_files.py index e7c3cff5..fced7a8b 100644 --- a/nc_py_api/files/_files.py +++ b/nc_py_api/files/_files.py @@ -26,6 +26,7 @@ "oc:id", "oc:fileid", "oc:downloadURL", + "nc:download-url-expiration", "oc:dDC", "oc:permissions", "oc:checksums", @@ -306,6 +307,10 @@ def _parse_record(full_path: str, prop_stats: list[dict]) -> FsNode: # noqa pyl fs_node_args["mimetype"] = prop["d:getcontenttype"] if "oc:permissions" in prop_keys: fs_node_args["permissions"] = prop["oc:permissions"] + if "oc:downloadURL" in prop_keys and prop["oc:downloadURL"]: + fs_node_args["download_url"] = prop["oc:downloadURL"] + if "nc:download-url-expiration" in prop_keys and prop["nc:download-url-expiration"]: + fs_node_args["download_url_expiration"] = int(prop["nc:download-url-expiration"]) if "oc:favorite" in prop_keys: fs_node_args["favorite"] = bool(int(prop["oc:favorite"])) if "nc:trashbin-filename" in prop_keys: @@ -367,7 +372,7 @@ def _webdav_response_to_records(webdav_res: Response, info: str) -> list[dict]: if "d:error" in response_data: err = response_data["d:error"] raise NextcloudException( - reason=f'{err["s:exception"]}: {err["s:message"]}'.replace("\n", ""), info=info, response=webdav_res + reason=f"{err['s:exception']}: {err['s:message']}".replace("\n", ""), info=info, response=webdav_res ) response = response_data["d:multistatus"].get("d:response", []) return [response] if isinstance(response, dict) else response From e0cf706613a09614aa0f8ee5d18fdb6a81f8f00d Mon Sep 17 00:00:00 2001 From: Review Date: Tue, 24 Mar 2026 09:44:31 +0000 Subject: [PATCH 2/2] test: add unit tests for download URL properties and harden parsing Add 7 unit tests covering FsNodeInfo defaults, value pass-through, _parse_record with/without presigned URL data, false values from non-S3 backends, and malformed expiration values. Guard int() conversion of download-url-expiration with contextlib.suppress to handle malformed server responses gracefully. --- nc_py_api/files/_files.py | 4 +- tests_unit/test_download_url.py | 92 +++++++++++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 1 deletion(-) create mode 100644 tests_unit/test_download_url.py diff --git a/nc_py_api/files/_files.py b/nc_py_api/files/_files.py index fced7a8b..38dbc6dc 100644 --- a/nc_py_api/files/_files.py +++ b/nc_py_api/files/_files.py @@ -1,5 +1,6 @@ """Helper functions for **FilesAPI** and **AsyncFilesAPI** classes.""" +import contextlib import enum from datetime import datetime, timezone from io import BytesIO @@ -310,7 +311,8 @@ def _parse_record(full_path: str, prop_stats: list[dict]) -> FsNode: # noqa pyl if "oc:downloadURL" in prop_keys and prop["oc:downloadURL"]: fs_node_args["download_url"] = prop["oc:downloadURL"] if "nc:download-url-expiration" in prop_keys and prop["nc:download-url-expiration"]: - fs_node_args["download_url_expiration"] = int(prop["nc:download-url-expiration"]) + with contextlib.suppress(TypeError, ValueError): + fs_node_args["download_url_expiration"] = int(prop["nc:download-url-expiration"]) if "oc:favorite" in prop_keys: fs_node_args["favorite"] = bool(int(prop["oc:favorite"])) if "nc:trashbin-filename" in prop_keys: diff --git a/tests_unit/test_download_url.py b/tests_unit/test_download_url.py new file mode 100644 index 00000000..f8c7d6df --- /dev/null +++ b/tests_unit/test_download_url.py @@ -0,0 +1,92 @@ +"""Tests for S3 presigned download URL properties on FsNode/FsNodeInfo.""" + +from nc_py_api.files import FsNode, FsNodeInfo +from nc_py_api.files._files import _parse_record + + +def test_fsnode_info_download_url_defaults(): + info = FsNodeInfo() + assert info.download_url == "" + assert info.download_url_expiration == 0 + + +def test_fsnode_info_download_url_with_values(): + info = FsNodeInfo(download_url="https://s3.example.com/bucket/obj?sig=abc", download_url_expiration=1700000000) + assert info.download_url == "https://s3.example.com/bucket/obj?sig=abc" + assert info.download_url_expiration == 1700000000 + + +def test_fsnode_passes_download_url_to_info(): + node = FsNode( + "files/admin/test.txt", file_id="00000123", download_url="https://s3.test/f", download_url_expiration=9999 + ) + assert node.info.download_url == "https://s3.test/f" + assert node.info.download_url_expiration == 9999 + + +def test_parse_record_with_download_url(): + prop_stat = { + "d:status": "HTTP/1.1 200 OK", + "d:prop": { + "oc:id": "00000123", + "oc:fileid": "123", + "oc:permissions": "RGDNVW", + "d:getetag": '"abc123"', + "oc:downloadURL": "https://s3.example.com/bucket/urn:oid:123?X-Amz-Signature=abc", + "nc:download-url-expiration": "1700000000", + }, + } + node = _parse_record("files/admin/test.txt", [prop_stat]) + assert node.info.download_url == "https://s3.example.com/bucket/urn:oid:123?X-Amz-Signature=abc" + assert node.info.download_url_expiration == 1700000000 + + +def test_parse_record_without_download_url(): + prop_stat = { + "d:status": "HTTP/1.1 200 OK", + "d:prop": { + "oc:id": "00000123", + "oc:fileid": "123", + "oc:permissions": "RGDNVW", + "d:getetag": '"abc123"', + }, + } + node = _parse_record("files/admin/test.txt", [prop_stat]) + assert node.info.download_url == "" + assert node.info.download_url_expiration == 0 + + +def test_parse_record_with_false_download_url(): + """When storage doesn't support presigned URLs, server returns 'false'.""" + prop_stat = { + "d:status": "HTTP/1.1 200 OK", + "d:prop": { + "oc:id": "00000123", + "oc:fileid": "123", + "oc:permissions": "RGDNVW", + "d:getetag": '"abc123"', + "oc:downloadURL": False, + "nc:download-url-expiration": False, + }, + } + node = _parse_record("files/admin/test.txt", [prop_stat]) + assert node.info.download_url == "" + assert node.info.download_url_expiration == 0 + + +def test_parse_record_with_malformed_expiration(): + """Malformed expiration should not crash parsing.""" + prop_stat = { + "d:status": "HTTP/1.1 200 OK", + "d:prop": { + "oc:id": "00000123", + "oc:fileid": "123", + "oc:permissions": "RGDNVW", + "d:getetag": '"abc123"', + "oc:downloadURL": "https://s3.test/f", + "nc:download-url-expiration": "not-a-number", + }, + } + node = _parse_record("files/admin/test.txt", [prop_stat]) + assert node.info.download_url == "https://s3.test/f" + assert node.info.download_url_expiration == 0