Add local presigned URL implementation for local clients

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] = []

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"]