| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103 |
- from __future__ import annotations
- import os
- from typing import Any
- from upath._compat import FSSpecAccessorShim as _FSSpecAccessorShim
- from upath._flavour import upath_strip_protocol
- from upath.core import UPath
- __all__ = [
- "CloudPath",
- "GCSPath",
- "S3Path",
- "AzurePath",
- ]
- # accessors are deprecated
- _CloudAccessor = _FSSpecAccessorShim
- class CloudPath(UPath):
- __slots__ = ()
- @classmethod
- def _transform_init_args(
- cls,
- args: tuple[str | os.PathLike, ...],
- protocol: str,
- storage_options: dict[str, Any],
- ) -> tuple[tuple[str | os.PathLike, ...], str, dict[str, Any]]:
- for key in ["bucket", "netloc"]:
- bucket = storage_options.pop(key, None)
- if bucket:
- if str(args[0]).startswith("/"):
- args = (f"{protocol}://{bucket}{args[0]}", *args[1:])
- else:
- args0 = upath_strip_protocol(args[0])
- args = (f"{protocol}://{bucket}/", args0, *args[1:])
- break
- return super()._transform_init_args(args, protocol, storage_options)
- def mkdir(
- self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False
- ) -> None:
- if not parents and not exist_ok and self.exists():
- raise FileExistsError(self.path)
- super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
- def iterdir(self):
- if self.is_file():
- raise NotADirectoryError(str(self))
- if self.parts[-1:] == ("",):
- yield from self.parent.iterdir()
- else:
- yield from super().iterdir()
- def relative_to(self, other, /, *_deprecated, walk_up=False):
- # use the parent implementation for the ValueError logic
- super().relative_to(other, *_deprecated, walk_up=False)
- return self
- class GCSPath(CloudPath):
- __slots__ = ()
- def __init__(
- self, *args, protocol: str | None = None, **storage_options: Any
- ) -> None:
- super().__init__(*args, protocol=protocol, **storage_options)
- if not self.drive and len(self.parts) > 1:
- raise ValueError("non key-like path provided (bucket/container missing)")
- def mkdir(
- self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False
- ) -> None:
- try:
- super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
- except TypeError as err:
- if "unexpected keyword argument 'create_parents'" in str(err):
- self.fs.mkdir(self.path)
- class S3Path(CloudPath):
- __slots__ = ()
- def __init__(
- self, *args, protocol: str | None = None, **storage_options: Any
- ) -> None:
- super().__init__(*args, protocol=protocol, **storage_options)
- if not self.drive and len(self.parts) > 1:
- raise ValueError("non key-like path provided (bucket/container missing)")
- class AzurePath(CloudPath):
- __slots__ = ()
- def __init__(
- self, *args, protocol: str | None = None, **storage_options: Any
- ) -> None:
- super().__init__(*args, protocol=protocol, **storage_options)
- if not self.drive and len(self.parts) > 1:
- raise ValueError("non key-like path provided (bucket/container missing)")
|
