http.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897
  1. import asyncio
  2. import io
  3. import logging
  4. import re
  5. import weakref
  6. from copy import copy
  7. from urllib.parse import urlparse
  8. import aiohttp
  9. import yarl
  10. from fsspec.asyn import AbstractAsyncStreamedFile, AsyncFileSystem, sync, sync_wrapper
  11. from fsspec.callbacks import DEFAULT_CALLBACK
  12. from fsspec.exceptions import FSTimeoutError
  13. from fsspec.spec import AbstractBufferedFile
  14. from fsspec.utils import (
  15. DEFAULT_BLOCK_SIZE,
  16. glob_translate,
  17. isfilelike,
  18. nullcontext,
  19. tokenize,
  20. )
  21. from ..caching import AllBytes
  22. # https://stackoverflow.com/a/15926317/3821154
  23. ex = re.compile(r"""<(a|A)\s+(?:[^>]*?\s+)?(href|HREF)=["'](?P<url>[^"']+)""")
  24. ex2 = re.compile(r"""(?P<url>http[s]?://[-a-zA-Z0-9@:%_+.~#?&/=]+)""")
  25. logger = logging.getLogger("fsspec.http")
  26. async def get_client(**kwargs):
  27. return aiohttp.ClientSession(**kwargs)
  28. class HTTPFileSystem(AsyncFileSystem):
  29. """
  30. Simple File-System for fetching data via HTTP(S)
  31. ``ls()`` is implemented by loading the parent page and doing a regex
  32. match on the result. If simple_link=True, anything of the form
  33. "http(s)://server.com/stuff?thing=other"; otherwise only links within
  34. HTML href tags will be used.
  35. """
  36. protocol = ("http", "https")
  37. sep = "/"
  38. def __init__(
  39. self,
  40. simple_links=True,
  41. block_size=None,
  42. same_scheme=True,
  43. size_policy=None,
  44. cache_type="bytes",
  45. cache_options=None,
  46. asynchronous=False,
  47. loop=None,
  48. client_kwargs=None,
  49. get_client=get_client,
  50. encoded=False,
  51. **storage_options,
  52. ):
  53. """
  54. NB: if this is called async, you must await set_client
  55. Parameters
  56. ----------
  57. block_size: int
  58. Blocks to read bytes; if 0, will default to raw requests file-like
  59. objects instead of HTTPFile instances
  60. simple_links: bool
  61. If True, will consider both HTML <a> tags and anything that looks
  62. like a URL; if False, will consider only the former.
  63. same_scheme: True
  64. When doing ls/glob, if this is True, only consider paths that have
  65. http/https matching the input URLs.
  66. size_policy: this argument is deprecated
  67. client_kwargs: dict
  68. Passed to aiohttp.ClientSession, see
  69. https://docs.aiohttp.org/en/stable/client_reference.html
  70. For example, ``{'auth': aiohttp.BasicAuth('user', 'pass')}``
  71. get_client: Callable[..., aiohttp.ClientSession]
  72. A callable, which takes keyword arguments and constructs
  73. an aiohttp.ClientSession. Its state will be managed by
  74. the HTTPFileSystem class.
  75. storage_options: key-value
  76. Any other parameters passed on to requests
  77. cache_type, cache_options: defaults used in open()
  78. """
  79. super().__init__(self, asynchronous=asynchronous, loop=loop, **storage_options)
  80. self.block_size = block_size if block_size is not None else DEFAULT_BLOCK_SIZE
  81. self.simple_links = simple_links
  82. self.same_schema = same_scheme
  83. self.cache_type = cache_type
  84. self.cache_options = cache_options
  85. self.client_kwargs = client_kwargs or {}
  86. self.get_client = get_client
  87. self.encoded = encoded
  88. self.kwargs = storage_options
  89. self._session = None
  90. # Clean caching-related parameters from `storage_options`
  91. # before propagating them as `request_options` through `self.kwargs`.
  92. # TODO: Maybe rename `self.kwargs` to `self.request_options` to make
  93. # it clearer.
  94. request_options = copy(storage_options)
  95. self.use_listings_cache = request_options.pop("use_listings_cache", False)
  96. request_options.pop("listings_expiry_time", None)
  97. request_options.pop("max_paths", None)
  98. request_options.pop("skip_instance_cache", None)
  99. self.kwargs = request_options
  100. @property
  101. def fsid(self):
  102. return "http"
  103. def encode_url(self, url):
  104. return yarl.URL(url, encoded=self.encoded)
  105. @staticmethod
  106. def close_session(loop, session):
  107. if loop is not None and loop.is_running():
  108. try:
  109. sync(loop, session.close, timeout=0.1)
  110. return
  111. except (TimeoutError, FSTimeoutError, NotImplementedError):
  112. pass
  113. connector = getattr(session, "_connector", None)
  114. if connector is not None:
  115. # close after loop is dead
  116. connector._close()
  117. async def set_session(self):
  118. if self._session is None:
  119. self._session = await self.get_client(loop=self.loop, **self.client_kwargs)
  120. if not self.asynchronous:
  121. weakref.finalize(self, self.close_session, self.loop, self._session)
  122. return self._session
  123. @classmethod
  124. def _strip_protocol(cls, path):
  125. """For HTTP, we always want to keep the full URL"""
  126. return path
  127. @classmethod
  128. def _parent(cls, path):
  129. # override, since _strip_protocol is different for URLs
  130. par = super()._parent(path)
  131. if len(par) > 7: # "http://..."
  132. return par
  133. return ""
  134. async def _ls_real(self, url, detail=True, **kwargs):
  135. # ignoring URL-encoded arguments
  136. kw = self.kwargs.copy()
  137. kw.update(kwargs)
  138. logger.debug(url)
  139. session = await self.set_session()
  140. async with session.get(self.encode_url(url), **self.kwargs) as r:
  141. self._raise_not_found_for_status(r, url)
  142. if "Content-Type" in r.headers:
  143. mimetype = r.headers["Content-Type"].partition(";")[0]
  144. else:
  145. mimetype = None
  146. if mimetype in ("text/html", None):
  147. try:
  148. text = await r.text(errors="ignore")
  149. if self.simple_links:
  150. links = ex2.findall(text) + [u[2] for u in ex.findall(text)]
  151. else:
  152. links = [u[2] for u in ex.findall(text)]
  153. except UnicodeDecodeError:
  154. links = [] # binary, not HTML
  155. else:
  156. links = []
  157. out = set()
  158. parts = urlparse(url)
  159. for l in links:
  160. if isinstance(l, tuple):
  161. l = l[1]
  162. if l.startswith("/") and len(l) > 1:
  163. # absolute URL on this server
  164. l = f"{parts.scheme}://{parts.netloc}{l}"
  165. if l.startswith("http"):
  166. if self.same_schema and l.startswith(url.rstrip("/") + "/"):
  167. out.add(l)
  168. elif l.replace("https", "http").startswith(
  169. url.replace("https", "http").rstrip("/") + "/"
  170. ):
  171. # allowed to cross http <-> https
  172. out.add(l)
  173. else:
  174. if l not in ["..", "../"]:
  175. # Ignore FTP-like "parent"
  176. out.add("/".join([url.rstrip("/"), l.lstrip("/")]))
  177. if not out and url.endswith("/"):
  178. out = await self._ls_real(url.rstrip("/"), detail=False)
  179. if detail:
  180. return [
  181. {
  182. "name": u,
  183. "size": None,
  184. "type": "directory" if u.endswith("/") else "file",
  185. }
  186. for u in out
  187. ]
  188. else:
  189. return sorted(out)
  190. async def _ls(self, url, detail=True, **kwargs):
  191. if self.use_listings_cache and url in self.dircache:
  192. out = self.dircache[url]
  193. else:
  194. out = await self._ls_real(url, detail=detail, **kwargs)
  195. self.dircache[url] = out
  196. return out
  197. ls = sync_wrapper(_ls)
  198. def _raise_not_found_for_status(self, response, url):
  199. """
  200. Raises FileNotFoundError for 404s, otherwise uses raise_for_status.
  201. """
  202. if response.status == 404:
  203. raise FileNotFoundError(url)
  204. response.raise_for_status()
  205. async def _cat_file(self, url, start=None, end=None, **kwargs):
  206. kw = self.kwargs.copy()
  207. kw.update(kwargs)
  208. logger.debug(url)
  209. if start is not None or end is not None:
  210. if start == end:
  211. return b""
  212. headers = kw.pop("headers", {}).copy()
  213. headers["Range"] = await self._process_limits(url, start, end)
  214. kw["headers"] = headers
  215. session = await self.set_session()
  216. async with session.get(self.encode_url(url), **kw) as r:
  217. out = await r.read()
  218. self._raise_not_found_for_status(r, url)
  219. return out
  220. async def _get_file(
  221. self, rpath, lpath, chunk_size=5 * 2**20, callback=DEFAULT_CALLBACK, **kwargs
  222. ):
  223. kw = self.kwargs.copy()
  224. kw.update(kwargs)
  225. logger.debug(rpath)
  226. session = await self.set_session()
  227. async with session.get(self.encode_url(rpath), **kw) as r:
  228. try:
  229. size = int(r.headers["content-length"])
  230. except (ValueError, KeyError):
  231. size = None
  232. callback.set_size(size)
  233. self._raise_not_found_for_status(r, rpath)
  234. if isfilelike(lpath):
  235. outfile = lpath
  236. else:
  237. outfile = open(lpath, "wb") # noqa: ASYNC230
  238. try:
  239. chunk = True
  240. while chunk:
  241. chunk = await r.content.read(chunk_size)
  242. outfile.write(chunk)
  243. callback.relative_update(len(chunk))
  244. finally:
  245. if not isfilelike(lpath):
  246. outfile.close()
  247. async def _put_file(
  248. self,
  249. lpath,
  250. rpath,
  251. chunk_size=5 * 2**20,
  252. callback=DEFAULT_CALLBACK,
  253. method="post",
  254. mode="overwrite",
  255. **kwargs,
  256. ):
  257. if mode != "overwrite":
  258. raise NotImplementedError("Exclusive write")
  259. async def gen_chunks():
  260. # Support passing arbitrary file-like objects
  261. # and use them instead of streams.
  262. if isinstance(lpath, io.IOBase):
  263. context = nullcontext(lpath)
  264. use_seek = False # might not support seeking
  265. else:
  266. context = open(lpath, "rb") # noqa: ASYNC230
  267. use_seek = True
  268. with context as f:
  269. if use_seek:
  270. callback.set_size(f.seek(0, 2))
  271. f.seek(0)
  272. else:
  273. callback.set_size(getattr(f, "size", None))
  274. chunk = f.read(chunk_size)
  275. while chunk:
  276. yield chunk
  277. callback.relative_update(len(chunk))
  278. chunk = f.read(chunk_size)
  279. kw = self.kwargs.copy()
  280. kw.update(kwargs)
  281. session = await self.set_session()
  282. method = method.lower()
  283. if method not in ("post", "put"):
  284. raise ValueError(
  285. f"method has to be either 'post' or 'put', not: {method!r}"
  286. )
  287. meth = getattr(session, method)
  288. async with meth(self.encode_url(rpath), data=gen_chunks(), **kw) as resp:
  289. self._raise_not_found_for_status(resp, rpath)
  290. async def _exists(self, path, strict=False, **kwargs):
  291. kw = self.kwargs.copy()
  292. kw.update(kwargs)
  293. try:
  294. logger.debug(path)
  295. session = await self.set_session()
  296. r = await session.get(self.encode_url(path), **kw)
  297. async with r:
  298. if strict:
  299. self._raise_not_found_for_status(r, path)
  300. return r.status < 400
  301. except FileNotFoundError:
  302. return False
  303. except aiohttp.ClientError:
  304. if strict:
  305. raise
  306. return False
  307. async def _isfile(self, path, **kwargs):
  308. return await self._exists(path, **kwargs)
  309. def _open(
  310. self,
  311. path,
  312. mode="rb",
  313. block_size=None,
  314. autocommit=None, # XXX: This differs from the base class.
  315. cache_type=None,
  316. cache_options=None,
  317. size=None,
  318. **kwargs,
  319. ):
  320. """Make a file-like object
  321. Parameters
  322. ----------
  323. path: str
  324. Full URL with protocol
  325. mode: string
  326. must be "rb"
  327. block_size: int or None
  328. Bytes to download in one request; use instance value if None. If
  329. zero, will return a streaming Requests file-like instance.
  330. kwargs: key-value
  331. Any other parameters, passed to requests calls
  332. """
  333. if mode != "rb":
  334. raise NotImplementedError
  335. block_size = block_size if block_size is not None else self.block_size
  336. kw = self.kwargs.copy()
  337. kw["asynchronous"] = self.asynchronous
  338. kw.update(kwargs)
  339. info = {}
  340. size = size or info.update(self.info(path, **kwargs)) or info["size"]
  341. session = sync(self.loop, self.set_session)
  342. if block_size and size and info.get("partial", True):
  343. return HTTPFile(
  344. self,
  345. path,
  346. session=session,
  347. block_size=block_size,
  348. mode=mode,
  349. size=size,
  350. cache_type=cache_type or self.cache_type,
  351. cache_options=cache_options or self.cache_options,
  352. loop=self.loop,
  353. **kw,
  354. )
  355. else:
  356. return HTTPStreamFile(
  357. self,
  358. path,
  359. mode=mode,
  360. loop=self.loop,
  361. session=session,
  362. **kw,
  363. )
  364. async def open_async(self, path, mode="rb", size=None, **kwargs):
  365. session = await self.set_session()
  366. if size is None:
  367. try:
  368. size = (await self._info(path, **kwargs))["size"]
  369. except FileNotFoundError:
  370. pass
  371. return AsyncStreamFile(
  372. self,
  373. path,
  374. loop=self.loop,
  375. session=session,
  376. size=size,
  377. **kwargs,
  378. )
  379. def ukey(self, url):
  380. """Unique identifier; assume HTTP files are static, unchanging"""
  381. return tokenize(url, self.kwargs, self.protocol)
  382. async def _info(self, url, **kwargs):
  383. """Get info of URL
  384. Tries to access location via HEAD, and then GET methods, but does
  385. not fetch the data.
  386. It is possible that the server does not supply any size information, in
  387. which case size will be given as None (and certain operations on the
  388. corresponding file will not work).
  389. """
  390. info = {}
  391. session = await self.set_session()
  392. for policy in ["head", "get"]:
  393. try:
  394. info.update(
  395. await _file_info(
  396. self.encode_url(url),
  397. size_policy=policy,
  398. session=session,
  399. **self.kwargs,
  400. **kwargs,
  401. )
  402. )
  403. if info.get("size") is not None:
  404. break
  405. except Exception as exc:
  406. if policy == "get":
  407. # If get failed, then raise a FileNotFoundError
  408. raise FileNotFoundError(url) from exc
  409. logger.debug("", exc_info=exc)
  410. return {"name": url, "size": None, **info, "type": "file"}
  411. async def _glob(self, path, maxdepth=None, **kwargs):
  412. """
  413. Find files by glob-matching.
  414. This implementation is idntical to the one in AbstractFileSystem,
  415. but "?" is not considered as a character for globbing, because it is
  416. so common in URLs, often identifying the "query" part.
  417. """
  418. if maxdepth is not None and maxdepth < 1:
  419. raise ValueError("maxdepth must be at least 1")
  420. import re
  421. ends_with_slash = path.endswith("/") # _strip_protocol strips trailing slash
  422. path = self._strip_protocol(path)
  423. append_slash_to_dirname = ends_with_slash or path.endswith(("/**", "/*"))
  424. idx_star = path.find("*") if path.find("*") >= 0 else len(path)
  425. idx_brace = path.find("[") if path.find("[") >= 0 else len(path)
  426. min_idx = min(idx_star, idx_brace)
  427. detail = kwargs.pop("detail", False)
  428. if not has_magic(path):
  429. if await self._exists(path, **kwargs):
  430. if not detail:
  431. return [path]
  432. else:
  433. return {path: await self._info(path, **kwargs)}
  434. else:
  435. if not detail:
  436. return [] # glob of non-existent returns empty
  437. else:
  438. return {}
  439. elif "/" in path[:min_idx]:
  440. min_idx = path[:min_idx].rindex("/")
  441. root = path[: min_idx + 1]
  442. depth = path[min_idx + 1 :].count("/") + 1
  443. else:
  444. root = ""
  445. depth = path[min_idx + 1 :].count("/") + 1
  446. if "**" in path:
  447. if maxdepth is not None:
  448. idx_double_stars = path.find("**")
  449. depth_double_stars = path[idx_double_stars:].count("/") + 1
  450. depth = depth - depth_double_stars + maxdepth
  451. else:
  452. depth = None
  453. allpaths = await self._find(
  454. root, maxdepth=depth, withdirs=True, detail=True, **kwargs
  455. )
  456. pattern = glob_translate(path + ("/" if ends_with_slash else ""))
  457. pattern = re.compile(pattern)
  458. out = {
  459. (
  460. p.rstrip("/")
  461. if not append_slash_to_dirname
  462. and info["type"] == "directory"
  463. and p.endswith("/")
  464. else p
  465. ): info
  466. for p, info in sorted(allpaths.items())
  467. if pattern.match(p.rstrip("/"))
  468. }
  469. if detail:
  470. return out
  471. else:
  472. return list(out)
  473. async def _isdir(self, path):
  474. # override, since all URLs are (also) files
  475. try:
  476. return bool(await self._ls(path))
  477. except (FileNotFoundError, ValueError):
  478. return False
  479. async def _pipe_file(self, path, value, mode="overwrite", **kwargs):
  480. """
  481. Write bytes to a remote file over HTTP.
  482. Parameters
  483. ----------
  484. path : str
  485. Target URL where the data should be written
  486. value : bytes
  487. Data to be written
  488. mode : str
  489. How to write to the file - 'overwrite' or 'append'
  490. **kwargs : dict
  491. Additional parameters to pass to the HTTP request
  492. """
  493. url = self._strip_protocol(path)
  494. headers = kwargs.pop("headers", {})
  495. headers["Content-Length"] = str(len(value))
  496. session = await self.set_session()
  497. async with session.put(url, data=value, headers=headers, **kwargs) as r:
  498. r.raise_for_status()
  499. class HTTPFile(AbstractBufferedFile):
  500. """
  501. A file-like object pointing to a remote HTTP(S) resource
  502. Supports only reading, with read-ahead of a predetermined block-size.
  503. In the case that the server does not supply the filesize, only reading of
  504. the complete file in one go is supported.
  505. Parameters
  506. ----------
  507. url: str
  508. Full URL of the remote resource, including the protocol
  509. session: aiohttp.ClientSession or None
  510. All calls will be made within this session, to avoid restarting
  511. connections where the server allows this
  512. block_size: int or None
  513. The amount of read-ahead to do, in bytes. Default is 5MB, or the value
  514. configured for the FileSystem creating this file
  515. size: None or int
  516. If given, this is the size of the file in bytes, and we don't attempt
  517. to call the server to find the value.
  518. kwargs: all other key-values are passed to requests calls.
  519. """
  520. def __init__(
  521. self,
  522. fs,
  523. url,
  524. session=None,
  525. block_size=None,
  526. mode="rb",
  527. cache_type="bytes",
  528. cache_options=None,
  529. size=None,
  530. loop=None,
  531. asynchronous=False,
  532. **kwargs,
  533. ):
  534. if mode != "rb":
  535. raise NotImplementedError("File mode not supported")
  536. self.asynchronous = asynchronous
  537. self.loop = loop
  538. self.url = url
  539. self.session = session
  540. self.details = {"name": url, "size": size, "type": "file"}
  541. super().__init__(
  542. fs=fs,
  543. path=url,
  544. mode=mode,
  545. block_size=block_size,
  546. cache_type=cache_type,
  547. cache_options=cache_options,
  548. **kwargs,
  549. )
  550. def read(self, length=-1):
  551. """Read bytes from file
  552. Parameters
  553. ----------
  554. length: int
  555. Read up to this many bytes. If negative, read all content to end of
  556. file. If the server has not supplied the filesize, attempting to
  557. read only part of the data will raise a ValueError.
  558. """
  559. if (
  560. (length < 0 and self.loc == 0) # explicit read all
  561. # but not when the size is known and fits into a block anyways
  562. and not (self.size is not None and self.size <= self.blocksize)
  563. ):
  564. self._fetch_all()
  565. if self.size is None:
  566. if length < 0:
  567. self._fetch_all()
  568. else:
  569. length = min(self.size - self.loc, length)
  570. return super().read(length)
  571. async def async_fetch_all(self):
  572. """Read whole file in one shot, without caching
  573. This is only called when position is still at zero,
  574. and read() is called without a byte-count.
  575. """
  576. logger.debug(f"Fetch all for {self}")
  577. if not isinstance(self.cache, AllBytes):
  578. r = await self.session.get(self.fs.encode_url(self.url), **self.kwargs)
  579. async with r:
  580. r.raise_for_status()
  581. out = await r.read()
  582. self.cache = AllBytes(
  583. size=len(out), fetcher=None, blocksize=None, data=out
  584. )
  585. self.size = len(out)
  586. _fetch_all = sync_wrapper(async_fetch_all)
  587. def _parse_content_range(self, headers):
  588. """Parse the Content-Range header"""
  589. s = headers.get("Content-Range", "")
  590. m = re.match(r"bytes (\d+-\d+|\*)/(\d+|\*)", s)
  591. if not m:
  592. return None, None, None
  593. if m[1] == "*":
  594. start = end = None
  595. else:
  596. start, end = [int(x) for x in m[1].split("-")]
  597. total = None if m[2] == "*" else int(m[2])
  598. return start, end, total
  599. async def async_fetch_range(self, start, end):
  600. """Download a block of data
  601. The expectation is that the server returns only the requested bytes,
  602. with HTTP code 206. If this is not the case, we first check the headers,
  603. and then stream the output - if the data size is bigger than we
  604. requested, an exception is raised.
  605. """
  606. logger.debug(f"Fetch range for {self}: {start}-{end}")
  607. kwargs = self.kwargs.copy()
  608. headers = kwargs.pop("headers", {}).copy()
  609. headers["Range"] = f"bytes={start}-{end - 1}"
  610. logger.debug(f"{self.url} : {headers['Range']}")
  611. r = await self.session.get(
  612. self.fs.encode_url(self.url), headers=headers, **kwargs
  613. )
  614. async with r:
  615. if r.status == 416:
  616. # range request outside file
  617. return b""
  618. r.raise_for_status()
  619. # If the server has handled the range request, it should reply
  620. # with status 206 (partial content). But we'll guess that a suitable
  621. # Content-Range header or a Content-Length no more than the
  622. # requested range also mean we have got the desired range.
  623. response_is_range = (
  624. r.status == 206
  625. or self._parse_content_range(r.headers)[0] == start
  626. or int(r.headers.get("Content-Length", end + 1)) <= end - start
  627. )
  628. if response_is_range:
  629. # partial content, as expected
  630. out = await r.read()
  631. elif start > 0:
  632. raise ValueError(
  633. "The HTTP server doesn't appear to support range requests. "
  634. "Only reading this file from the beginning is supported. "
  635. "Open with block_size=0 for a streaming file interface."
  636. )
  637. else:
  638. # Response is not a range, but we want the start of the file,
  639. # so we can read the required amount anyway.
  640. cl = 0
  641. out = []
  642. while True:
  643. chunk = await r.content.read(2**20)
  644. # data size unknown, let's read until we have enough
  645. if chunk:
  646. out.append(chunk)
  647. cl += len(chunk)
  648. if cl > end - start:
  649. break
  650. else:
  651. break
  652. out = b"".join(out)[: end - start]
  653. return out
  654. _fetch_range = sync_wrapper(async_fetch_range)
  655. magic_check = re.compile("([*[])")
  656. def has_magic(s):
  657. match = magic_check.search(s)
  658. return match is not None
  659. class HTTPStreamFile(AbstractBufferedFile):
  660. def __init__(self, fs, url, mode="rb", loop=None, session=None, **kwargs):
  661. self.asynchronous = kwargs.pop("asynchronous", False)
  662. self.url = url
  663. self.loop = loop
  664. self.session = session
  665. if mode != "rb":
  666. raise ValueError
  667. self.details = {"name": url, "size": None}
  668. super().__init__(fs=fs, path=url, mode=mode, cache_type="none", **kwargs)
  669. async def cor():
  670. r = await self.session.get(self.fs.encode_url(url), **kwargs).__aenter__()
  671. self.fs._raise_not_found_for_status(r, url)
  672. return r
  673. self.r = sync(self.loop, cor)
  674. self.loop = fs.loop
  675. def seek(self, loc, whence=0):
  676. if loc == 0 and whence == 1:
  677. return
  678. if loc == self.loc and whence == 0:
  679. return
  680. raise ValueError("Cannot seek streaming HTTP file")
  681. async def _read(self, num=-1):
  682. out = await self.r.content.read(num)
  683. self.loc += len(out)
  684. return out
  685. read = sync_wrapper(_read)
  686. async def _close(self):
  687. self.r.close()
  688. def close(self):
  689. asyncio.run_coroutine_threadsafe(self._close(), self.loop)
  690. super().close()
  691. class AsyncStreamFile(AbstractAsyncStreamedFile):
  692. def __init__(
  693. self, fs, url, mode="rb", loop=None, session=None, size=None, **kwargs
  694. ):
  695. self.url = url
  696. self.session = session
  697. self.r = None
  698. if mode != "rb":
  699. raise ValueError
  700. self.details = {"name": url, "size": None}
  701. self.kwargs = kwargs
  702. super().__init__(fs=fs, path=url, mode=mode, cache_type="none")
  703. self.size = size
  704. async def read(self, num=-1):
  705. if self.r is None:
  706. r = await self.session.get(
  707. self.fs.encode_url(self.url), **self.kwargs
  708. ).__aenter__()
  709. self.fs._raise_not_found_for_status(r, self.url)
  710. self.r = r
  711. out = await self.r.content.read(num)
  712. self.loc += len(out)
  713. return out
  714. async def close(self):
  715. if self.r is not None:
  716. self.r.close()
  717. self.r = None
  718. await super().close()
  719. async def get_range(session, url, start, end, file=None, **kwargs):
  720. # explicit get a range when we know it must be safe
  721. kwargs = kwargs.copy()
  722. headers = kwargs.pop("headers", {}).copy()
  723. headers["Range"] = f"bytes={start}-{end - 1}"
  724. r = await session.get(url, headers=headers, **kwargs)
  725. r.raise_for_status()
  726. async with r:
  727. out = await r.read()
  728. if file:
  729. with open(file, "r+b") as f: # noqa: ASYNC230
  730. f.seek(start)
  731. f.write(out)
  732. else:
  733. return out
  734. async def _file_info(url, session, size_policy="head", **kwargs):
  735. """Call HEAD on the server to get details about the file (size/checksum etc.)
  736. Default operation is to explicitly allow redirects and use encoding
  737. 'identity' (no compression) to get the true size of the target.
  738. """
  739. logger.debug("Retrieve file size for %s", url)
  740. kwargs = kwargs.copy()
  741. ar = kwargs.pop("allow_redirects", True)
  742. head = kwargs.get("headers", {}).copy()
  743. head["Accept-Encoding"] = "identity"
  744. kwargs["headers"] = head
  745. info = {}
  746. if size_policy == "head":
  747. r = await session.head(url, allow_redirects=ar, **kwargs)
  748. elif size_policy == "get":
  749. r = await session.get(url, allow_redirects=ar, **kwargs)
  750. else:
  751. raise TypeError(f'size_policy must be "head" or "get", got {size_policy}')
  752. async with r:
  753. r.raise_for_status()
  754. if "Content-Length" in r.headers:
  755. # Some servers may choose to ignore Accept-Encoding and return
  756. # compressed content, in which case the returned size is unreliable.
  757. if "Content-Encoding" not in r.headers or r.headers["Content-Encoding"] in [
  758. "identity",
  759. "",
  760. ]:
  761. info["size"] = int(r.headers["Content-Length"])
  762. elif "Content-Range" in r.headers:
  763. info["size"] = int(r.headers["Content-Range"].split("/")[1])
  764. if "Content-Type" in r.headers:
  765. info["mimetype"] = r.headers["Content-Type"].partition(";")[0]
  766. if r.headers.get("Accept-Ranges") == "none":
  767. # Some servers may explicitly discourage partial content requests, but
  768. # the lack of "Accept-Ranges" does not always indicate they would fail
  769. info["partial"] = False
  770. info["url"] = str(r.url)
  771. for checksum_field in ["ETag", "Content-MD5", "Digest", "Last-Modified"]:
  772. if r.headers.get(checksum_field):
  773. info[checksum_field] = r.headers[checksum_field]
  774. return info
  775. async def _file_size(url, session=None, *args, **kwargs):
  776. if session is None:
  777. session = await get_client()
  778. info = await _file_info(url, session=session, *args, **kwargs)
  779. return info.get("size")
  780. file_size = sync_wrapper(_file_size)