Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix UPath.rename for absolute paths #225

Merged
merged 9 commits into from
Aug 23, 2024
42 changes: 31 additions & 11 deletions upath/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,17 @@
from typing import Mapping
from typing import Sequence
from typing import TextIO
from typing import TypedDict
from typing import TypeVar
from typing import overload
from urllib.parse import urlsplit

if sys.version_info >= (3, 11):
from typing import Self
from typing import Unpack
else:
from typing_extensions import Self
from typing_extensions import Unpack

from fsspec.registry import get_filesystem_class
from fsspec.spec import AbstractFileSystem
Expand Down Expand Up @@ -91,6 +94,11 @@ def _make_instance(cls, args, kwargs):
return cls(*args, **kwargs)


class _UPathRenameParams(TypedDict, total=False):
recursive: bool
maxdepth: int | None


# accessors are deprecated
_FSSpecAccessor = FSSpecAccessorShim

Expand Down Expand Up @@ -1005,21 +1013,33 @@ def rmdir(self, recursive: bool = True) -> None: # fixme: non-standard
def rename(
self,
target: str | os.PathLike[str] | UPath,
*,
recursive: bool = False,
maxdepth: int | None = None,
**kwargs: Any,
) -> UPath: # fixme: non-standard
target_: UPath
if not isinstance(target, UPath):
target_ = self.parent.joinpath(target).resolve()
**kwargs: Unpack[_UPathRenameParams], # note: non-standard compared to pathlib
) -> Self:
if isinstance(target, str) and self.storage_options:
target = UPath(target, **self.storage_options)
target_protocol = get_upath_protocol(target)
if target_protocol:
if target_protocol != self.protocol:
raise ValueError(
f"expected protocol {self.protocol!r}, got: {target_protocol!r}"
)
if not isinstance(target, UPath):
target_ = UPath(target, **self.storage_options)
else:
target_ = target
# avoid calling .resolve for subclasses of UPath
if ".." in target_.parts or "." in target_.parts:
target_ = target_.resolve()
else:
target_ = target
parent = self.parent
# avoid calling .resolve for subclasses of UPath
if ".." in parent.parts or "." in parent.parts:
parent = parent.resolve()
target_ = parent.joinpath(os.path.normpath(target))
assert isinstance(target_, type(self)), "identical protocols enforced above"
self.fs.mv(
self.path,
target_.path,
recursive=recursive,
maxdepth=maxdepth,
**kwargs,
)
return target_
Expand Down
31 changes: 20 additions & 11 deletions upath/implementations/smb.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,20 @@
from __future__ import annotations

import os
import sys
import warnings

if sys.version_info >= (3, 11):
from typing import Self
from typing import Unpack
else:
from typing_extensions import Self
from typing_extensions import Unpack

import smbprotocol.exceptions

from upath import UPath
from upath.core import _UPathRenameParams


class SMBPath(UPath):
Expand All @@ -29,24 +41,21 @@ def iterdir(self):
else:
return super().iterdir()

def rename(self, target, **kwargs):
if "recursive" in kwargs:
def rename(
self,
target: str | os.PathLike[str] | UPath,
**kwargs: Unpack[_UPathRenameParams], # note: non-standard compared to pathlib
) -> Self:
if kwargs.pop("recursive", None) is not None:
warnings.warn(
"SMBPath.rename(): recursive is currently ignored.",
UserWarning,
stacklevel=2,
)
if "maxdepth" in kwargs:
if kwargs.pop("maxdepth", None) is not None:
warnings.warn(
"SMBPath.rename(): maxdepth is currently ignored.",
UserWarning,
stacklevel=2,
)
if not isinstance(target, UPath):
target = self.parent.joinpath(target).resolve()
self.fs.mv(
self.path,
target.path,
**kwargs,
)
return target
return super().rename(target, **kwargs)
10 changes: 10 additions & 0 deletions upath/tests/cases.py
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,11 @@ def test_rename(self):
assert target == moved
assert not upath.exists()
assert moved.exists()
# reverse with an absolute path as str
back = moved.rename(str(upath))
assert back == upath
assert not moved.exists()
assert back.exists()

def test_rename2(self):
upath = self.path.joinpath("folder1/file2.txt")
Expand All @@ -291,6 +296,11 @@ def test_rename2(self):
assert target_path == moved
assert not upath.exists()
assert moved.exists()
# reverse with a relative path as UPath
back = moved.rename(UPath("file2.txt"))
assert back == upath
assert not moved.exists()
assert back.exists()

def test_replace(self):
pass
Expand Down
Loading