Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions HISTORY.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
- Implement sliced downloads in GSClient. (Issue [#387](https://github.com/drivendataorg/cloudpathlib/issues/387), PR [#389](https://github.com/drivendataorg/cloudpathlib/pull/389))
- Implement `as_url` with presigned parameter for all backends. (Issue [#235](https://github.com/drivendataorg/cloudpathlib/issues/235), PR [#236](https://github.com/drivendataorg/cloudpathlib/pull/236))
- Stream to and from Azure Blob Storage. (PR [#403](https://github.com/drivendataorg/cloudpathlib/pull/403))
- Implement `file:` URI scheme support for `AnyPath`. (Issue [#401](https://github.com/drivendataorg/cloudpathlib/issues/401), PR [#404](https://github.com/drivendataorg/cloudpathlib/pull/404))

## 0.17.0 (2023-12-21)

Expand Down
7 changes: 7 additions & 0 deletions cloudpathlib/anypath.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from .cloudpath import InvalidPrefixError, CloudPath
from .exceptions import AnyPathTypeError
from .url_utils import path_from_fileurl


class AnyPath(ABC):
Expand All @@ -22,6 +23,12 @@ def __new__(cls, *args, **kwargs) -> Union[CloudPath, Path]: # type: ignore
return CloudPath(*args, **kwargs) # type: ignore
except InvalidPrefixError as cloudpath_exception:
try:
if isinstance(args[0], str) and args[0].lower().startswith("file:"):
path = path_from_fileurl(args[0], **kwargs)
for part in args[1:]:
path /= part
return path

return Path(*args, **kwargs)
except TypeError as path_exception:
raise AnyPathTypeError(
Expand Down
31 changes: 31 additions & 0 deletions cloudpathlib/url_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from pathlib import PureWindowsPath, Path
from urllib.request import url2pathname
from urllib.parse import urlparse, unquote


def path_from_fileurl(urlstr, **kwargs):
"""
Take a file:// url and return a Path.

Adapted from:
https://github.com/AcademySoftwareFoundation/OpenTimelineIO/blob/4c17494dee2e515aedc8623741556fae3e4afe72/src/py-opentimelineio/opentimelineio/url_utils.py#L43-L72
"""
# explicitly unquote first in case drive colon is url encoded
unquoted = unquote(urlstr)

# Parse provided URL
parsed_result = urlparse(unquoted)

# Convert the parsed URL to a path
filepath = Path(url2pathname(parsed_result.path), **kwargs)

# If the network location is a window drive, reassemble the path
if PureWindowsPath(parsed_result.netloc).drive:
filepath = Path(parsed_result.netloc + parsed_result.path, **kwargs)

# Otherwise check if the specified index is a windows drive, then offset the path
elif len(filepath.parts) > 1 and PureWindowsPath(filepath.parts[1]).drive:
# Remove leading "/" if/when `request.url2pathname` yields "/S:/path/file.ext"
filepath = Path(*filepath.parts[1:], **kwargs)

return filepath
25 changes: 25 additions & 0 deletions docs/docs/anypath-polymorphism.md
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,31 @@ isinstance(cloud_path, AnyPath)
#> True
```

## `file:` URI Scheme

`AnyPath` also supports the [`file:` URI scheme](https://en.wikipedia.org/wiki/File_URI_scheme) _for paths that can be referenced with pathlib_ and returns a `Path` instance for those paths. If you need to roundtrip back to a `file:` URI, you can use the `Path.as_uri` method after any path manipulations that you do.

For example:

```python
from cloudpathlib import AnyPath

# hostname omitted variant
path = AnyPath("file:/root/mydir/myfile.txt")
path
#> PosixPath('/root/mydir/myfile.txt')

# explicit local path variant
path = AnyPath("file:///root/mydir/myfile.txt")
path
#> PosixPath('/root/mydir/myfile.txt')

# manipulate the path and return the file:// URI
parent_uri = path.parent.as_uri()
parent_uri
#> 'file:///root/mydir'
```

## How It Works

The constructor for `AnyPath` will first attempt to run the input through the `CloudPath` base class' constructor, which will validate the input against registered concrete `CloudPath` implementations. This will accept inputs that are already a cloud path class or a string with the appropriate URI scheme prefix (e.g., `s3://`). If no implementation validates successfully, it will then try to run the input through the `Path` constructor. If the `Path` constructor fails and raises a `TypeError`, then the `AnyPath` constructor will raise an `AnyPathTypeError` exception.
Expand Down
24 changes: 24 additions & 0 deletions tests/test_anypath.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import os
from pathlib import Path, PosixPath, WindowsPath

import pytest
Expand All @@ -20,6 +21,29 @@ def test_anypath_path():
assert issubclass(WindowsPath, AnyPath)
assert not issubclass(str, AnyPath)

# test `file:` scheme (only works with absolute paths; needs .absolute() on Windows)
assert AnyPath(path.absolute().resolve().as_uri()) == path.absolute().resolve()

# test file:// + multi arg
assert AnyPath(*path.absolute().resolve().as_uri().rsplit("/", 2)) == path.absolute().resolve()

# test no hostname
assert Path("/foo/bar") == AnyPath("file:/foo/bar")
assert Path("/foo/bar") == AnyPath("file:///foo/bar")

# windows tests
if os.name == "nt":
assert Path("c:\\hello\\test.txt") == AnyPath("file:/c:/hello/test.txt")
assert Path("c:\\hello\\test.txt") == AnyPath("file://c:/hello/test.txt")
assert Path("c:\\hello\\test.txt") == AnyPath("file:///c:/hello/test.txt")
assert Path("c:\\hello\\test.txt") == AnyPath("file://c%3A//hello/test.txt")
assert Path("c:\\hello\\test.txt") == AnyPath("file://localhost/c%3a/hello/test.txt")
assert Path("c:\\WINDOWS\\clock.avi") == AnyPath("file://localhost/c|/WINDOWS/clock.avi")
assert Path("c:\\WINDOWS\\clock.avi") == AnyPath("file:///c|/WINDOWS/clock.avi")
assert Path("c:\\hello\\test space.txt") == AnyPath(
"file://localhost/c%3a/hello/test%20space.txt"
)


def test_anypath_cloudpath(rig):
cloudpath = rig.create_cloud_path("a/b/c")
Expand Down