1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
from __future__ import annotations
import sys
import warnings
from collections.abc import Iterator
from itertools import chain
from typing import TYPE_CHECKING
from typing import Any
from urllib.parse import urlsplit
from fsspec.asyn import sync
from upath import UnsupportedOperation
from upath._stat import UPathStatResult
from upath.core import UPath
from upath.types import JoinablePathLike
from upath.types import StatResultType
if TYPE_CHECKING:
from typing import Literal
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 upath._chain import FSSpecChainParser
from upath.types.storage_options import HTTPStorageOptions
__all__ = ["HTTPPath"]
class HTTPPath(UPath):
__slots__ = ()
if TYPE_CHECKING:
def __init__(
self,
*args: JoinablePathLike,
protocol: Literal["http", "https"] | None = ...,
chain_parser: FSSpecChainParser = ...,
**storage_options: Unpack[HTTPStorageOptions],
) -> None: ...
@classmethod
def _transform_init_args(
cls,
args: tuple[JoinablePathLike, ...],
protocol: str,
storage_options: dict[str, Any],
) -> tuple[tuple[JoinablePathLike, ...], str, dict[str, Any]]:
# allow initialization via a path argument and protocol keyword
if args and not str(args[0]).startswith(protocol):
args = (f"{protocol}://{str(args[0]).lstrip('/')}", *args[1:])
return args, protocol, storage_options
def __str__(self) -> str:
sr = urlsplit(super().__str__())
return sr._replace(path=sr.path or "/").geturl()
@property
def path(self) -> str:
sr = urlsplit(super().path)
return sr._replace(path=sr.path or "/").geturl()
def stat(self, follow_symlinks: bool = True) -> StatResultType:
if not follow_symlinks:
warnings.warn(
f"{type(self).__name__}.stat(follow_symlinks=False):"
" is currently ignored.",
UserWarning,
stacklevel=2,
)
info = self.fs.info(self.path)
if "url" in info:
info["type"] = "directory" if info["url"].endswith("/") else "file"
return UPathStatResult.from_info(info)
def iterdir(self) -> Iterator[Self]:
it = iter(super().iterdir())
try:
item0 = next(it)
except (StopIteration, NotADirectoryError):
raise NotADirectoryError(str(self))
except FileNotFoundError:
raise FileNotFoundError(str(self))
else:
yield from chain([item0], it)
def resolve(
self,
strict: bool = False,
follow_redirects: bool = True,
) -> Self:
"""Normalize the path and resolve redirects."""
# special handling of trailing slash behaviour
parts = list(self.parts)
if parts[-1:] == ["."]:
parts[-1:] = [""]
if parts[-2:] == ["", ".."]:
parts[-2:] = [""]
pth = self.with_segments(*parts)
resolved_path = super(HTTPPath, pth).resolve(strict=strict)
if follow_redirects:
cls = type(self)
# Get the fsspec fs
fs = self.fs
url = str(self)
# Ensure we have a session
session = sync(fs.loop, fs.set_session)
# Use HEAD requests if the server allows it, falling back to GETs
for method in (session.head, session.get):
r = sync(fs.loop, method, url, allow_redirects=True)
try:
r.raise_for_status()
except Exception as exc:
if method == session.get:
raise FileNotFoundError(self) from exc
else:
resolved_path = cls(str(r.url))
break
return resolved_path
def touch(self, mode: int = 0o666, exist_ok: bool = True) -> None:
raise UnsupportedOperation
def mkdir(
self,
mode: int = 0o777,
parents: bool = False,
exist_ok: bool = False,
) -> None:
raise UnsupportedOperation
def unlink(self, missing_ok: bool = False) -> None:
raise UnsupportedOperation
def write_bytes(self, data: bytes) -> int:
raise UnsupportedOperation("DataPath does not support writing")
def write_text(
self,
data: str,
encoding: str | None = None,
errors: str | None = None,
newline: str | None = None,
) -> int:
raise UnsupportedOperation("DataPath does not support writing")
|
