diff --git a/cloudpathlib/local/localclient.py b/cloudpathlib/local/localclient.py index 50ec666b..b3f73932 100644 --- a/cloudpathlib/local/localclient.py +++ b/cloudpathlib/local/localclient.py @@ -8,6 +8,7 @@ from tempfile import TemporaryDirectory from time import sleep from typing import Callable, ClassVar, Dict, Iterable, List, Optional, Tuple, Union +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from ..client import Client from ..enums import FileCacheMode @@ -207,7 +208,12 @@ def _get_public_url(self, cloud_path: "LocalPath") -> str: def _generate_presigned_url( self, cloud_path: "LocalPath", expire_seconds: int = 60 * 60 ) -> str: - raise NotImplementedError("Cannot generate a presigned URL for a local path.") + public_url = self._get_public_url(cloud_path) + parts = urlsplit(public_url) + query = dict(parse_qsl(parts.query, keep_blank_values=True)) + query["expires"] = str(expire_seconds) + query["signature"] = "local" + return urlunsplit(parts._replace(query=urlencode(query))) _temp_dirs_to_clean: List[TemporaryDirectory] = [] diff --git a/tests/test_local.py b/tests/test_local.py index 15f1b6f9..582bf712 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -1,6 +1,7 @@ import pytest from inspect import signature +from urllib.parse import parse_qs, urlsplit from cloudpathlib import AzureBlobClient, AzureBlobPath, GSClient, GSPath, S3Client, S3Path from cloudpathlib.local import ( @@ -133,3 +134,22 @@ def test_glob_matches(client_class, monkeypatch): # match CloudPath, which returns empty; not glob module, which raises assert list(p.glob("*")) == [] + + +@pytest.mark.parametrize("client_class", [LocalAzureBlobClient, LocalGSClient, LocalS3Client]) +def test_as_url_presign(client_class, monkeypatch): + if client_class is LocalAzureBlobClient: + monkeypatch.setenv("AZURE_STORAGE_CONNECTION_STRING", "") + + cloud_prefix = client_class._cloud_meta.path_class.cloud_prefix + p = client_class().CloudPath(f"{cloud_prefix}drive/file.txt") + p.write_text("hello") + + expire_seconds = 123 + presigned_url = p.as_url(presign=True, expire_seconds=expire_seconds) + parts = urlsplit(presigned_url) + query_params = parse_qs(parts.query) + + assert parts.path.endswith("file.txt") + assert query_params["expires"] == [str(expire_seconds)] + assert query_params["signature"] == ["local"]