Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/upath/implementations/cloud.py: 43%

1from __future__ import annotations 

2 

3import os 

4from typing import Any 

5 

6from upath._compat import FSSpecAccessorShim as _FSSpecAccessorShim 

7from upath._flavour import upath_strip_protocol 

8from upath.core import UPath 

9 

10__all__ = [ 

11 "CloudPath", 

12 "GCSPath", 

13 "S3Path", 

14 "AzurePath", 

15] 

16 

17 

18# accessors are deprecated 

19_CloudAccessor = _FSSpecAccessorShim 

20 

21 

22class CloudPath(UPath): 

23 __slots__ = () 

24 

25 @classmethod 

26 def _transform_init_args( 

27 cls, 

28 args: tuple[str | os.PathLike, ...], 

29 protocol: str, 

30 storage_options: dict[str, Any], 

31 ) -> tuple[tuple[str | os.PathLike, ...], str, dict[str, Any]]: 

32 for key in ["bucket", "netloc"]: 

33 bucket = storage_options.pop(key, None) 

34 if bucket: 

35 if str(args[0]).startswith("/"): 

36 args = (f"{protocol}://{bucket}{args[0]}", *args[1:]) 

37 else: 

38 args0 = upath_strip_protocol(args[0]) 

39 args = (f"{protocol}://{bucket}/", args0, *args[1:]) 

40 break 

41 return super()._transform_init_args(args, protocol, storage_options) 

42 

43 def mkdir( 

44 self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False 

45 ) -> None: 

46 if not parents and not exist_ok and self.exists(): 

47 raise FileExistsError(self.path) 

48 super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok) 

49 

50 def iterdir(self): 

51 if self.is_file(): 

52 raise NotADirectoryError(str(self)) 

53 if self.parts[-1:] == ("",): 

54 yield from self.parent.iterdir() 

55 else: 

56 yield from super().iterdir() 

57 

58 def relative_to(self, other, /, *_deprecated, walk_up=False): 

59 # use the parent implementation for the ValueError logic 

60 super().relative_to(other, *_deprecated, walk_up=False) 

61 return self 

62 

63 

64class GCSPath(CloudPath): 

65 __slots__ = () 

66 

67 def __init__( 

68 self, *args, protocol: str | None = None, **storage_options: Any 

69 ) -> None: 

70 super().__init__(*args, protocol=protocol, **storage_options) 

71 if not self.drive and len(self.parts) > 1: 

72 raise ValueError("non key-like path provided (bucket/container missing)") 

73 

74 def mkdir( 

75 self, mode: int = 0o777, parents: bool = False, exist_ok: bool = False 

76 ) -> None: 

77 try: 

78 super().mkdir(mode=mode, parents=parents, exist_ok=exist_ok) 

79 except TypeError as err: 

80 if "unexpected keyword argument 'create_parents'" in str(err): 

81 self.fs.mkdir(self.path) 

82 

83 

84class S3Path(CloudPath): 

85 __slots__ = () 

86 

87 def __init__( 

88 self, *args, protocol: str | None = None, **storage_options: Any 

89 ) -> None: 

90 super().__init__(*args, protocol=protocol, **storage_options) 

91 if not self.drive and len(self.parts) > 1: 

92 raise ValueError("non key-like path provided (bucket/container missing)") 

93 

94 

95class AzurePath(CloudPath): 

96 __slots__ = () 

97 

98 def __init__( 

99 self, *args, protocol: str | None = None, **storage_options: Any 

100 ) -> None: 

101 super().__init__(*args, protocol=protocol, **storage_options) 

102 if not self.drive and len(self.parts) > 1: 

103 raise ValueError("non key-like path provided (bucket/container missing)")