fix(filesystem): UNC paths are not supported

dlt/common/storages/configuration.py

@@ -1,6 +1,7 @@
 import os
-from typing import TYPE_CHECKING, Any, Literal, Optional, Type, get_args, ClassVar, Dict, Union
-from urllib.parse import urlparse
+import pathlib
+from typing import Any, Literal, Optional, Type, get_args, ClassVar, Dict, Union
+from urllib.parse import urlparse, unquote
 
 from dlt.common.configuration import configspec, resolve_type
 from dlt.common.configuration.exceptions import ConfigurationValueError
@@ -87,24 +88,22 @@ class FilesystemConfiguration(BaseConfiguration):
     @property
     def protocol(self) -> str:
         """`bucket_url` protocol"""
-        url = urlparse(self.bucket_url)
-        # this prevents windows absolute paths to be recognized as schemas
-        if not url.scheme or (os.path.isabs(self.bucket_url) and "\\" in self.bucket_url):
+        if self.is_local_path(self.bucket_url):
             return "file"
         else:
-            return url.scheme
+            return urlparse(self.bucket_url).scheme
 
     def on_resolved(self) -> None:
-        url = urlparse(self.bucket_url)
-        if not url.path and not url.netloc:
+        uri = urlparse(self.bucket_url)
+        if not uri.path and not uri.netloc:
             raise ConfigurationValueError(
-                "File path or netloc missing. Field bucket_url of FilesystemClientConfiguration"
-                " must contain valid url with a path or host:password component."
+                "File path and netloc are missing. Field bucket_url of"
+                " FilesystemClientConfiguration must contain valid uri with a path or host:password"
+                " component."
             )
         # this is just a path in a local file system
-        if url.path == self.bucket_url:
-            url = url._replace(scheme="file")
-            self.bucket_url = url.geturl()
+        if self.is_local_path(self.bucket_url):
+            self.bucket_url = self.make_file_uri(self.bucket_url)
 
     @resolve_type("credentials")
     def resolve_credentials_type(self) -> Type[CredentialsConfiguration]:
@@ -117,11 +116,57 @@ def fingerprint(self) -> str:
 
     def __str__(self) -> str:
         """Return displayable destination location"""
-        url = urlparse(self.bucket_url)
+        uri = urlparse(self.bucket_url)
         # do not show passwords
-        if url.password:
-            new_netloc = f"{url.username}:****@{url.hostname}"
-            if url.port:
-                new_netloc += f":{url.port}"
-            return url._replace(netloc=new_netloc).geturl()
+        if uri.password:
+            new_netloc = f"{uri.username}:****@{uri.hostname}"
+            if uri.port:
+                new_netloc += f":{uri.port}"
+            return uri._replace(netloc=new_netloc).geturl()
         return self.bucket_url
+
+    @staticmethod
+    def is_local_path(uri: str) -> bool:
+        """Checks if `uri` is a local path, without a schema"""
+        uri_parsed = urlparse(uri)
+        # this prevents windows absolute paths to be recognized as schemas
+        return not uri_parsed.scheme or os.path.isabs(uri)
+
+    @staticmethod
+    def make_local_path(file_uri: str) -> str:
+        """Gets a valid local filesystem path from file:// scheme.
+        Supports POSIX/Windows/UNC paths
+
+        Returns:
+            str: local filesystem path
+        """
+        uri = urlparse(file_uri)
+        if uri.scheme != "file":
+            raise ValueError(f"Must be file scheme but is {uri.scheme}")
+        if not uri.path and not uri.netloc:
+            raise ConfigurationValueError("File path and netloc are missing.")
+        local_path = unquote(uri.path)
+        if uri.netloc:
+            # or UNC file://localhost/path
+            local_path = "//" + unquote(uri.netloc) + local_path
+        else:
+            # if we are on windows, strip the POSIX root from path which is always absolute
+            if os.path.sep != local_path[0]:
+                # filesystem root
+                if local_path == "/":
+                    return str(pathlib.Path("/").resolve())
+                # this prevents /C:/ or ///share/ where both POSIX and Windows root are present
+                if os.path.isabs(local_path[1:]):
+                    local_path = local_path[1:]
+        return str(pathlib.Path(local_path))
+
+    @staticmethod
+    def make_file_uri(local_path: str) -> str:
+        """Creates a normalized file:// uri from a local path
+
+        netloc is never set. UNC paths are represented as file://host/path
+        """
+        p_ = pathlib.Path(local_path)
+        p_ = p_.expanduser().resolve()
+        # return "file:///" + p_.as_posix().lstrip("/")
+        return p_.as_uri()

dlt/common/storages/fsspec_filesystem.py

@@ -1,9 +1,11 @@
 import io
+import glob
 import gzip
 import mimetypes
 import pathlib
 import posixpath
 from io import BytesIO
+from gzip import GzipFile
 from typing import (
     Literal,
     cast,
@@ -41,6 +43,7 @@ class FileItem(TypedDict, total=False):
 
     file_url: str
     file_name: str
+    relative_path: str
     mime_type: str
     encoding: Optional[str]
     modification_date: pendulum.DateTime
@@ -181,12 +184,22 @@ def fsspec(self) -> AbstractFileSystem:
         else:
             return fsspec_filesystem(self["file_url"], self.credentials)[0]
 
+    @property
+    def local_file_path(self) -> str:
+        """Gets a valid local filesystem path from file:// scheme.
+        Supports POSIX/Windows/UNC paths
+
+        Returns:
+            str: local filesystem path
+        """
+        return FilesystemConfiguration.make_local_path(self["file_url"])
+
     def open(  # noqa: A003
         self,
         mode: str = "rb",
         compression: Literal["auto", "disable", "enable"] = "auto",
         **kwargs: Any,
-    ) -> IO[Any]:
+    ) -> Union[GzipFile, IO[Any]]:
         """Open the file as a fsspec file.
 
         This method opens the file represented by this dictionary as a file-like object using
@@ -213,7 +226,7 @@ def open(  # noqa: A003
             raise ValueError("""The argument `compression` must have one of the following values:
                 "auto", "enable", "disable".""")
 
-        opened_file: IO[Any]
+        opened_file: Union[IO[Any], GzipFile]
         # if the user has already extracted the content, we use it so there is no need to
         # download the file again.
         if "file_content" in self:
@@ -234,8 +247,13 @@ def open(  # noqa: A003
                 **text_kwargs,
             )
         else:
+            if "local" in self.fsspec.protocol:
+                # use native local file path to open file:// uris
+                file_url = self.local_file_path
+            else:
+                file_url = self["file_url"]
             opened_file = self.fsspec.open(
-                self["file_url"], mode=mode, compression=compression_arg, **kwargs
+                file_url, mode=mode, compression=compression_arg, **kwargs
             )
         return opened_file
 
@@ -245,11 +263,11 @@ def read_bytes(self) -> bytes:
         Returns:
             bytes: The file content.
         """
-        return (  # type: ignore
-            self["file_content"]
-            if "file_content" in self and self["file_content"] is not None
-            else self.fsspec.read_bytes(self["file_url"])
-        )
+        if "file_content" in self and self["file_content"] is not None:
+            return self["file_content"]  # type: ignore
+        else:
+            with self.open(mode="rb", compression="disable") as f:
+                return f.read()
 
 
 def guess_mime_type(file_name: str) -> Sequence[str]:
@@ -274,45 +292,49 @@ def glob_files(
     Returns:
         Iterable[FileItem]: The list of files.
     """
-    import os
-
-    bucket_url_parsed = urlparse(bucket_url)
-    # if this is a file path without a scheme
-    if not bucket_url_parsed.scheme or (os.path.isabs(bucket_url) and "\\" in bucket_url):
-        # this is a file so create a proper file url
-        bucket_url = pathlib.Path(bucket_url).absolute().as_uri()
+    is_local_fs = "local" in fs_client.protocol
+    if is_local_fs and FilesystemConfiguration.is_local_path(bucket_url):
+        bucket_url = FilesystemConfiguration.make_file_uri(bucket_url)
+        bucket_url_parsed = urlparse(bucket_url)
+    else:
         bucket_url_parsed = urlparse(bucket_url)
-    bucket_url_no_schema = bucket_url_parsed._replace(scheme="", query="").geturl()
-    bucket_url_no_schema = (
-        bucket_url_no_schema[2:] if bucket_url_no_schema.startswith("//") else bucket_url_no_schema
-    )
-    filter_url = posixpath.join(bucket_url_no_schema, file_glob)
 
-    glob_result = fs_client.glob(filter_url, detail=True)
-    if isinstance(glob_result, list):
-        raise NotImplementedError(
-            "Cannot request details when using fsspec.glob. For adlfs (Azure) please use version"
-            " 2023.9.0 or later"
-        )
+    if is_local_fs:
+        root_dir = FilesystemConfiguration.make_local_path(bucket_url)
+        # use a Python glob to get files
+        files = glob.glob(str(pathlib.Path(root_dir).joinpath(file_glob)), recursive=True)
+        glob_result = {file: fs_client.info(file) for file in files}
+    else:
+        root_dir = bucket_url_parsed._replace(scheme="", query="").geturl().lstrip("/")
+        filter_url = posixpath.join(root_dir, file_glob)
+        glob_result = fs_client.glob(filter_url, detail=True)
+        if isinstance(glob_result, list):
+            raise NotImplementedError(
+                "Cannot request details when using fsspec.glob. For adlfs (Azure) please use"
+                " version 2023.9.0 or later"
+            )
 
     for file, md in glob_result.items():
         if md["type"] != "file":
             continue
+        # relative paths are always POSIX
+        if is_local_fs:
+            rel_path = pathlib.Path(file).relative_to(root_dir).as_posix()
+            file_url = FilesystemConfiguration.make_file_uri(file)
+        else:
+            rel_path = posixpath.relpath(file, root_dir)
+            file_url = bucket_url_parsed._replace(
+                path=posixpath.join(bucket_url_parsed.path, rel_path)
+            ).geturl()
 
-        # make that absolute path on a file://
-        if bucket_url_parsed.scheme == "file" and not file.startswith("/"):
-            file = f"/{file}"
-        file_name = posixpath.relpath(file, bucket_url_no_schema)
-        file_url = bucket_url_parsed._replace(
-            path=posixpath.join(bucket_url_parsed.path, file_name)
-        ).geturl()
-
-        mime_type, encoding = guess_mime_type(file_name)
+        scheme = bucket_url_parsed.scheme
+        mime_type, encoding = guess_mime_type(rel_path)
         yield FileItem(
-            file_name=file_name,
+            file_name=posixpath.basename(rel_path),
+            relative_path=rel_path,
             file_url=file_url,
             mime_type=mime_type,
             encoding=encoding,
-            modification_date=MTIME_DISPATCH[bucket_url_parsed.scheme](md),
+            modification_date=MTIME_DISPATCH[scheme](md),
             size_in_bytes=int(md["size"]),
         )

dlt/sources/helpers/rest_client/paginators.py

@@ -171,6 +171,7 @@ class BaseReferencePaginator(BasePaginator):
       2. `update_request` method to update the request object with the next
         page reference.
     """
+
     def __init__(self) -> None:
         super().__init__()
         self.__next_reference: Optional[str] = None
@@ -208,6 +209,7 @@ class BaseNextUrlPaginator(BaseReferencePaginator):
 
     See `HeaderLinkPaginator` and `JSONResponsePaginator` for examples.
     """
+
     def update_request(self, request: Request) -> None:
         # Handle relative URLs
         if self._next_reference:

docs/tools/fix_grammar_gpt.py

@@ -18,7 +18,7 @@
 # constants
 BASE_DIR = "../website/docs"
 GPT_MODEL = "gpt-3.5-turbo-0125"
-MAX_CHUNK_SIZE = 14000 # make sure that this is below the context window size of the model to not have cut off files
+MAX_CHUNK_SIZE = 14000  # make sure that this is below the context window size of the model to not have cut off files
 
 SYSTEM_PROMPT = """\
 You are a grammar checker. Every message you get will be a document that is to be grammarchecked and returned as such.
@@ -104,7 +104,7 @@ def get_chunk_length(chunk: List[str]) -> int:
         chunks.append(current_chunk)
 
         # sanity test, make sure we still have the full doc
-        assert doc == functools.reduce(lambda a, b: a+b, chunks)
+        assert doc == functools.reduce(lambda a, b: a + b, chunks)
 
         fmt.note(f"Created {len(chunks)} chunks")
 

docs/website/docs/dlt-ecosystem/destinations/filesystem.md

@@ -174,7 +174,43 @@ If for any reason you want to have those files in a local folder, set up the `bu
 ```toml
 [destination.filesystem]
 bucket_url = "file:///absolute/path"  # three / for an absolute path
-# bucket_url = "file://relative/path" # two / for a relative path
+```
+
+`dlt` correctly handles the native local file paths. Indeed, using the `file://` schema may be not intuitive especially for Windows users.
+
+```toml
+[destination.filesystem]
+bucket_url = 'C:\a\b\c'
+```
+
+In the example above we specify `bucket_url` using **toml's literal strings** that do not require [escaping of backslashes](https://github.com/toml-lang/toml/blob/main/toml.md#string).
+
+```toml
+[destination.unc_destination]
+bucket_url = '\\localhost\c$\a\b\c'  # UNC equivalent of C:\a\b\c
+
+[destination.posix_destination]
+bucket_url = '/var/local/data'  # absolute POSIX style path
+
+[destination.relative_destination]
+bucket_url = '_storage/data'  # relative POSIX style path
+```
+
+In the examples above we define a few named filesystem destinations:
+* **unc_destination** demonstrates Windows UNC path in native form
+* **posix_destination** demonstrates native POSIX (Linux/Mac) absolute path
+* **relative_destination** demonstrates native POSIX (Linux/Mac) relative path. In this case  `filesystem` destination will store files in `$cwd/_storage/data` path
+where **$cwd** is your current working directory.
+
+`dlt` supports Windows [UNC paths with file:// scheme](https://en.wikipedia.org/wiki/File_URI_scheme). They can be specified using **host** or purely as **path**
+component.
+
+```toml
+[destination.unc_with_host]
+bucket_url="file://localhost/c$/a/b/c"
+
+[destination.unc_with_path]
+bucket_url="file:////localhost/c$/a/b/c"
 ```
 
 ## Write disposition
@@ -408,7 +444,4 @@ managed in the regular way by the final destination you have configured.
 You will also notice `init` files being present in the root folder and the special `dlt` folders. In the absence of the concepts of schemas and tables
 in blob storages and directories, `dlt` uses these special files to harmonize the behavior of the `filesystem` destination with the other implemented destinations.
 
-
-
-
 <!--@@@DLT_TUBA filesystem-->