Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import asyncio
2import io
3import logging
4import random
5import re
6import weakref
7from copy import copy
8from urllib.parse import urlparse
10import aiohttp
11import yarl
13from fsspec.asyn import AbstractAsyncStreamedFile, AsyncFileSystem, sync, sync_wrapper
14from fsspec.callbacks import DEFAULT_CALLBACK
15from fsspec.exceptions import FSTimeoutError
16from fsspec.spec import AbstractBufferedFile
17from fsspec.utils import (
18 DEFAULT_BLOCK_SIZE,
19 glob_translate,
20 isfilelike,
21 nullcontext,
22 range_response_window,
23 response_is_requested_range,
24 tokenize,
25)
27from ..caching import AllBytes
29# https://stackoverflow.com/a/15926317/3821154
30# The attribute scan must stay linear: it stops at the next "<" or ">" and the
31# whitespace before href is a lookbehind rather than a second quantifier, so a
32# page cannot make it backtrack cubically (1 KB of whitespace after "<a"
33# used to stall ls() for seconds, 10 KB for hours).
34ex = re.compile(r"""<(a|A)\s[^<>]*?(?<=\s)(href|HREF)=["'](?P<url>[^"']+)""")
35ex2 = re.compile(r"""(?P<url>http[s]?://[-a-zA-Z0-9@:%_+.~#?&/=]+)""")
36logger = logging.getLogger("fsspec.http")
38_RETRYABLE_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504})
39# dropped or reset connections, truncated bodies, timeouts
40_RETRYABLE_EXCEPTIONS = (
41 aiohttp.ClientConnectionError,
42 aiohttp.ClientPayloadError,
43 asyncio.TimeoutError,
44)
45_RETRY_MAX_WAIT = 60.0
48def _is_retryable(exc, statuses=_RETRYABLE_STATUSES, exceptions=None):
49 """Whether a failed HTTP read is worth repeating.
51 A response error is retried when its status is in ``statuses``; any
52 other error is retried when it is an instance of ``exceptions``. The
53 default (None) is ``_RETRYABLE_EXCEPTIONS`` minus SSL failures, which
54 are deterministic. Anything else (including the ``FileNotFoundError``/
55 ``PermissionError`` a subclass may map 4xx codes to) propagates at once.
56 """
57 if isinstance(exc, aiohttp.ClientResponseError):
58 return exc.status in statuses
59 if exceptions is None:
60 return isinstance(exc, _RETRYABLE_EXCEPTIONS) and not isinstance(
61 exc, aiohttp.ClientSSLError
62 )
63 return isinstance(exc, exceptions)
66def _retry_after(exc):
67 """Delta-seconds ``Retry-After`` carried by a response error, else None."""
68 headers = getattr(exc, "headers", None)
69 if not headers:
70 return None
71 try:
72 return max(0.0, float(headers.get("Retry-After")))
73 except (TypeError, ValueError):
74 return None
77async def _with_retries(attempt, fs, label):
78 """Await ``attempt()``, repeating it on transient failures.
80 ``attempt`` is a zero-argument coroutine function covering the whole
81 request *and* body read, so a connection dropped mid-body counts as a
82 failure too. The policy comes from the ``retries``, ``retry_wait``,
83 ``retry_statuses`` and ``retry_exceptions`` attributes of ``fs``. A
84 ``Retry-After`` header is honoured when present; otherwise the wait
85 doubles from ``retry_wait`` on each retry, jittered. Either way the wait
86 is capped at ``_RETRY_MAX_WAIT`` seconds. Non-retryable errors and the
87 final failure propagate unchanged.
88 """
89 retries = fs.retries
90 for n in range(retries + 1):
91 try:
92 return await attempt()
93 except Exception as exc:
94 if n >= retries or not _is_retryable(
95 exc, fs.retry_statuses, fs.retry_exceptions
96 ):
97 raise
98 delay = _retry_after(exc)
99 if delay is None:
100 delay = fs.retry_wait * 2**n * (0.5 + random.random())
101 delay = min(_RETRY_MAX_WAIT, delay)
102 logger.warning(
103 "%s: attempt %d/%d failed (%r); retrying in %.1fs",
104 label,
105 n + 1,
106 retries + 1,
107 exc,
108 delay,
109 )
110 await asyncio.sleep(delay)
113async def get_client(**kwargs):
114 return aiohttp.ClientSession(**kwargs)
117class HTTPFileSystem(AsyncFileSystem):
118 """
119 Simple File-System for fetching data via HTTP(S)
121 ``ls()`` is implemented by loading the parent page and doing a regex
122 match on the result. If simple_link=True, anything of the form
123 "http(s)://server.com/stuff?thing=other"; otherwise only links within
124 HTML href tags will be used.
126 URLs are passed unfiltered to aiohttp, so all addresses are accessible. Where URLs are
127 supplied by a user, the calling application may wish to filter to prevent scanning.
128 """
130 protocol = ("http", "https")
131 sep = "/"
133 def __init__(
134 self,
135 simple_links=True,
136 block_size=None,
137 same_scheme=True,
138 size_policy=None,
139 cache_type="bytes",
140 cache_options=None,
141 asynchronous=False,
142 loop=None,
143 client_kwargs=None,
144 get_client=get_client,
145 encoded=False,
146 retries=3,
147 retry_wait=1.0,
148 retry_statuses=None,
149 retry_exceptions=None,
150 **storage_options,
151 ):
152 """
153 NB: if this is called async, you must await set_client
155 Parameters
156 ----------
157 block_size: int
158 Blocks to read bytes; if 0, will default to raw requests file-like
159 objects instead of HTTPFile instances
160 simple_links: bool
161 If True, will consider both HTML <a> tags and anything that looks
162 like a URL; if False, will consider only the former.
163 same_scheme: True
164 When doing ls/glob, if this is True, only consider paths that have
165 http/https matching the input URLs.
166 size_policy: this argument is deprecated
167 client_kwargs: dict
168 Passed to aiohttp.ClientSession, see
169 https://docs.aiohttp.org/en/stable/client_reference.html
170 For example, ``{'auth': aiohttp.BasicAuth('user', 'pass')}``
171 get_client: Callable[..., aiohttp.ClientSession]
172 A callable, which takes keyword arguments and constructs
173 an aiohttp.ClientSession. Its state will be managed by
174 the HTTPFileSystem class.
175 retries: int
176 How many times a transient failure of a read (see below)
177 is retried in ``cat_file``, ``HTTPFile`` block reads and the
178 ``info`` lookup behind ``open``; 0 disables retries.
179 retry_wait: float
180 Seconds to wait before the first retry; doubled on each further
181 retry, with jitter. A ``Retry-After`` header sent by the server
182 takes precedence. Either wait is capped at 60 s.
183 retry_statuses: container of int or None
184 HTTP status codes of a failed read that are retried; default
185 (None) is 408, 425, 429, 500, 502, 503 and 504.
186 retry_exceptions: exception class, tuple of classes or None
187 Errors other than a response status that are retried; default
188 (None) is dropped or reset connections (except SSL failures),
189 truncated bodies and timeouts.
190 storage_options: key-value
191 Any other parameters passed on to requests
192 cache_type, cache_options: defaults used in open()
193 """
194 super().__init__(self, asynchronous=asynchronous, loop=loop, **storage_options)
195 self.block_size = block_size if block_size is not None else DEFAULT_BLOCK_SIZE
196 self.simple_links = simple_links
197 self.same_schema = same_scheme
198 self.cache_type = cache_type
199 self.cache_options = cache_options
200 self.client_kwargs = client_kwargs or {}
201 self.get_client = get_client
202 self.encoded = encoded
203 if retries < 0:
204 raise ValueError(f"retries must be >= 0, got {retries!r}")
205 if retry_wait < 0:
206 raise ValueError(f"retry_wait must be >= 0, got {retry_wait!r}")
207 self.retries = retries
208 self.retry_wait = retry_wait
209 self.retry_statuses = (
210 retry_statuses if retry_statuses is not None else _RETRYABLE_STATUSES
211 )
212 self.retry_exceptions = retry_exceptions
213 self.kwargs = storage_options
214 self._session = None
216 # Clean caching-related parameters from `storage_options`
217 # before propagating them as `request_options` through `self.kwargs`.
218 # TODO: Maybe rename `self.kwargs` to `self.request_options` to make
219 # it clearer.
220 request_options = copy(storage_options)
221 self.use_listings_cache = request_options.pop("use_listings_cache", False)
222 request_options.pop("listings_expiry_time", None)
223 request_options.pop("max_paths", None)
224 request_options.pop("skip_instance_cache", None)
225 self.kwargs = request_options
227 @property
228 def fsid(self):
229 return "http"
231 def encode_url(self, url):
232 return yarl.URL(url, encoded=self.encoded)
234 @staticmethod
235 def close_session(loop, session):
236 if loop is not None and loop.is_running():
237 try:
238 sync(loop, session.close, timeout=0.1)
239 return
240 except (TimeoutError, FSTimeoutError, NotImplementedError, RuntimeError):
241 pass
242 connector = getattr(session, "_connector", None)
243 if connector is not None:
244 # close after loop is dead
245 connector._close()
247 async def set_session(self):
248 if self._session is None:
249 self._session = await self.get_client(loop=self.loop, **self.client_kwargs)
250 if not self.asynchronous:
251 weakref.finalize(self, self.close_session, self.loop, self._session)
252 return self._session
254 @classmethod
255 def _strip_protocol(cls, path):
256 """For HTTP, we always want to keep the full URL"""
257 return path
259 @classmethod
260 def _parent(cls, path):
261 # override, since _strip_protocol is different for URLs
262 par = super()._parent(path)
263 if len(par) > 7: # "http://..."
264 return par
265 return ""
267 async def _ls_real(self, url, detail=True, **kwargs):
268 # ignoring URL-encoded arguments
269 kw = self.kwargs.copy()
270 kw.update(kwargs)
271 logger.debug(url)
272 session = await self.set_session()
273 async with session.get(self.encode_url(url), **kw) as r:
274 self._raise_not_found_for_status(r, url)
276 if "Content-Type" in r.headers:
277 mimetype = r.headers["Content-Type"].partition(";")[0]
278 else:
279 mimetype = None
281 if mimetype in ("text/html", None):
282 try:
283 text = await r.text(errors="ignore")
284 if self.simple_links:
285 links = ex2.findall(text) + [u[2] for u in ex.findall(text)]
286 else:
287 links = [u[2] for u in ex.findall(text)]
288 except UnicodeDecodeError:
289 links = [] # binary, not HTML
290 else:
291 links = []
293 out = set()
294 parts = urlparse(url)
295 for l in links:
296 if isinstance(l, tuple):
297 l = l[1]
298 if l.startswith("/") and len(l) > 1:
299 # absolute URL on this server
300 l = f"{parts.scheme}://{parts.netloc}{l}"
301 if l.startswith("http"):
302 if self.same_schema and l.startswith(url.rstrip("/") + "/"):
303 out.add(l)
304 elif l.replace("https", "http").startswith(
305 url.replace("https", "http").rstrip("/") + "/"
306 ):
307 # allowed to cross http <-> https
308 out.add(l)
309 else:
310 if l not in ["..", "../"]:
311 # Ignore FTP-like "parent"
312 out.add("/".join([url.rstrip("/"), l.lstrip("/")]))
313 if not out and url.endswith("/"):
314 out = await self._ls_real(url.rstrip("/"), detail=False, **kwargs)
315 if detail:
316 return [
317 {
318 "name": u,
319 "size": None,
320 "type": "directory" if u.endswith("/") else "file",
321 }
322 for u in out
323 ]
324 else:
325 return sorted(out)
327 async def _ls(self, url, detail=True, **kwargs):
328 if self.use_listings_cache and url in self.dircache:
329 out = self.dircache[url]
330 else:
331 out = await self._ls_real(url, detail=True, **kwargs)
332 self.dircache[url] = out
333 if detail:
334 return out
335 return sorted(entry["name"] for entry in out)
337 ls = sync_wrapper(_ls)
339 def _raise_not_found_for_status(self, response, url):
340 """
341 Raises FileNotFoundError for 404s, otherwise uses raise_for_status.
342 """
343 if response.status == 404:
344 raise FileNotFoundError(url)
345 response.raise_for_status()
347 async def _cat_file(self, url, start=None, end=None, **kwargs):
348 kw = self.kwargs.copy()
349 kw.update(kwargs)
350 logger.debug(url)
352 ranged = start is not None or end is not None
353 if ranged:
354 if start == end or (start is None and end == 0):
355 return b""
356 headers = kw.pop("headers", {}).copy()
358 headers["Range"] = await self._process_limits(url, start, end)
359 kw["headers"] = headers
360 session = await self.set_session()
362 async def _once():
363 async with session.get(self.encode_url(url), **kw) as r:
364 out = await r.read()
365 self._raise_not_found_for_status(r, url)
366 # a server that ignores the Range header replies with the whole
367 # file, which is not the window the caller asked for
368 return (
369 range_response_window(out, r.status, r.headers, start, end)
370 if ranged
371 else out
372 )
374 return await _with_retries(_once, self, url)
376 async def _get_file(
377 self, rpath, lpath, chunk_size=5 * 2**20, callback=DEFAULT_CALLBACK, **kwargs
378 ):
379 kw = self.kwargs.copy()
380 kw.update(kwargs)
381 logger.debug(rpath)
382 session = await self.set_session()
383 async with session.get(self.encode_url(rpath), **kw) as r:
384 try:
385 size = int(r.headers["content-length"])
386 except (ValueError, KeyError):
387 size = None
389 callback.set_size(size)
390 self._raise_not_found_for_status(r, rpath)
391 if isfilelike(lpath):
392 outfile = lpath
393 else:
394 outfile = open(lpath, "wb") # noqa: ASYNC230
396 try:
397 chunk = True
398 while chunk:
399 chunk = await r.content.read(chunk_size)
400 outfile.write(chunk)
401 callback.relative_update(len(chunk))
402 finally:
403 if not isfilelike(lpath):
404 outfile.close()
406 async def _put_file(
407 self,
408 lpath,
409 rpath,
410 chunk_size=5 * 2**20,
411 callback=DEFAULT_CALLBACK,
412 method="post",
413 mode="overwrite",
414 **kwargs,
415 ):
416 if mode != "overwrite":
417 raise NotImplementedError("Exclusive write")
419 async def gen_chunks():
420 # Support passing arbitrary file-like objects
421 # and use them instead of streams.
422 if isinstance(lpath, io.IOBase):
423 context = nullcontext(lpath)
424 use_seek = False # might not support seeking
425 else:
426 context = open(lpath, "rb") # noqa: ASYNC230
427 use_seek = True
429 with context as f:
430 if use_seek:
431 callback.set_size(f.seek(0, 2))
432 f.seek(0)
433 else:
434 callback.set_size(getattr(f, "size", None))
436 chunk = f.read(chunk_size)
437 while chunk:
438 yield chunk
439 callback.relative_update(len(chunk))
440 chunk = f.read(chunk_size)
442 kw = self.kwargs.copy()
443 kw.update(kwargs)
444 session = await self.set_session()
446 method = method.lower()
447 if method not in ("post", "put"):
448 raise ValueError(
449 f"method has to be either 'post' or 'put', not: {method!r}"
450 )
452 meth = getattr(session, method)
453 async with meth(self.encode_url(rpath), data=gen_chunks(), **kw) as resp:
454 self._raise_not_found_for_status(resp, rpath)
456 async def _exists(self, path, strict=False, **kwargs):
457 kw = self.kwargs.copy()
458 kw.update(kwargs)
459 try:
460 logger.debug(path)
461 session = await self.set_session()
462 r = await session.get(self.encode_url(path), **kw)
463 async with r:
464 if strict:
465 self._raise_not_found_for_status(r, path)
466 return r.status < 400
467 except FileNotFoundError:
468 return False
469 except aiohttp.ClientError:
470 if strict:
471 raise
472 return False
474 async def _isfile(self, path, **kwargs):
475 return await self._exists(path, **kwargs)
477 def _open(
478 self,
479 path,
480 mode="rb",
481 block_size=None,
482 autocommit=None, # XXX: This differs from the base class.
483 cache_type=None,
484 cache_options=None,
485 size=None,
486 **kwargs,
487 ):
488 """Make a file-like object
490 Parameters
491 ----------
492 path: str
493 Full URL with protocol
494 mode: string
495 must be "rb"
496 block_size: int or None
497 Bytes to download in one request; use instance value if None. If
498 zero, will return a streaming Requests file-like instance.
499 kwargs: key-value
500 Any other parameters, passed to requests calls
501 """
502 if mode != "rb":
503 raise NotImplementedError
504 block_size = block_size if block_size is not None else self.block_size
505 kw = self.kwargs.copy()
506 kw["asynchronous"] = self.asynchronous
507 kw.update(kwargs)
508 info = {}
509 size = size or info.update(self.info(path, **kwargs)) or info["size"]
510 session = sync(self.loop, self.set_session)
511 if block_size and size and info.get("partial", True):
512 return HTTPFile(
513 self,
514 path,
515 session=session,
516 block_size=block_size,
517 mode=mode,
518 size=size,
519 cache_type=cache_type or self.cache_type,
520 cache_options=cache_options or self.cache_options,
521 loop=self.loop,
522 **kw,
523 )
524 else:
525 return HTTPStreamFile(
526 self,
527 path,
528 mode=mode,
529 loop=self.loop,
530 session=session,
531 **kw,
532 )
534 async def open_async(self, path, mode="rb", size=None, **kwargs):
535 session = await self.set_session()
536 if size is None:
537 try:
538 size = (await self._info(path, **kwargs))["size"]
539 except FileNotFoundError:
540 pass
541 return AsyncStreamFile(
542 self,
543 path,
544 loop=self.loop,
545 session=session,
546 size=size,
547 **kwargs,
548 )
550 def ukey(self, url):
551 """Unique identifier; assume HTTP files are static, unchanging"""
552 return tokenize(url, self.kwargs, self.protocol)
554 async def _info(self, url, **kwargs):
555 """Get info of URL
557 Tries to access location via HEAD, and then GET methods, but does
558 not fetch the data.
560 It is possible that the server does not supply any size information, in
561 which case size will be given as None (and certain operations on the
562 corresponding file will not work).
563 """
564 info = {}
565 session = await self.set_session()
567 async def _once():
568 # HEAD, then GET if that gave no size or failed for any reason
569 # (many servers mishandle HEAD); only a GET failure counts as a
570 # failed attempt.
571 for policy in ["head", "get"]:
572 try:
573 info.update(
574 await _file_info(
575 self.encode_url(url),
576 size_policy=policy,
577 session=session,
578 **self.kwargs,
579 **kwargs,
580 )
581 )
582 if info.get("size") is not None:
583 break
584 except Exception as exc:
585 if policy == "get":
586 raise
587 logger.debug("", exc_info=exc)
589 try:
590 await _with_retries(_once, self, url)
591 except Exception as exc:
592 # If get failed, then raise a FileNotFoundError
593 raise FileNotFoundError(url) from exc
595 return {"name": url, "size": None, **info, "type": "file"}
597 async def _glob(self, path, maxdepth=None, **kwargs):
598 """
599 Find files by glob-matching.
601 This implementation is idntical to the one in AbstractFileSystem,
602 but "?" is not considered as a character for globbing, because it is
603 so common in URLs, often identifying the "query" part.
604 """
605 if maxdepth is not None and maxdepth < 1:
606 raise ValueError("maxdepth must be at least 1")
607 import re
609 ends_with_slash = path.endswith("/") # _strip_protocol strips trailing slash
610 path = self._strip_protocol(path)
611 append_slash_to_dirname = ends_with_slash or path.endswith(("/**", "/*"))
612 idx_star = path.find("*") if path.find("*") >= 0 else len(path)
613 idx_brace = path.find("[") if path.find("[") >= 0 else len(path)
615 min_idx = min(idx_star, idx_brace)
617 detail = kwargs.pop("detail", False)
619 if not has_magic(path):
620 if await self._exists(path, **kwargs):
621 if not detail:
622 return [path]
623 else:
624 return {path: await self._info(path, **kwargs)}
625 else:
626 if not detail:
627 return [] # glob of non-existent returns empty
628 else:
629 return {}
630 elif "/" in path[:min_idx]:
631 min_idx = path[:min_idx].rindex("/")
632 root = path[: min_idx + 1]
633 depth = path[min_idx + 1 :].count("/") + 1
634 else:
635 root = ""
636 depth = path[min_idx + 1 :].count("/") + 1
638 if "**" in path:
639 if maxdepth is not None:
640 idx_double_stars = path.find("**")
641 depth_double_stars = path[idx_double_stars:].count("/") + 1
642 depth = depth - depth_double_stars + maxdepth
643 else:
644 depth = None
646 allpaths = await self._find(
647 root, maxdepth=depth, withdirs=True, detail=True, **kwargs
648 )
650 pattern = glob_translate(path + ("/" if ends_with_slash else ""))
651 pattern = re.compile(pattern)
653 out = {
654 (
655 p.rstrip("/")
656 if not append_slash_to_dirname
657 and info["type"] == "directory"
658 and p.endswith("/")
659 else p
660 ): info
661 for p, info in sorted(allpaths.items())
662 if pattern.match(p.rstrip("/"))
663 }
665 if detail:
666 return out
667 else:
668 return list(out)
670 async def _isdir(self, path):
671 # override, since all URLs are (also) files
672 try:
673 return bool(await self._ls(path))
674 except (FileNotFoundError, ValueError):
675 return False
677 async def _pipe_file(self, path, value, mode="overwrite", **kwargs):
678 """
679 Write bytes to a remote file over HTTP.
681 Parameters
682 ----------
683 path : str
684 Target URL where the data should be written
685 value : bytes
686 Data to be written
687 mode : str
688 Only 'overwrite' is accepted, allowing an unconditional PUT request.
689 The server determines PUT behavior, so other modes are rejected
690 defensively rather than assuming append or create semantics.
691 **kwargs : dict
692 Additional parameters to pass to the HTTP request
693 """
694 if mode != "overwrite":
695 raise ValueError(
696 "HTTP pipe_file only accepts mode='overwrite'; "
697 "PUT behavior is server-dependent, so other modes are "
698 "rejected defensively"
699 )
700 url = self._strip_protocol(path)
701 headers = kwargs.pop("headers", {})
702 headers["Content-Length"] = str(len(value))
704 session = await self.set_session()
706 async with session.put(
707 self.encode_url(url), data=value, headers=headers, **kwargs
708 ) as r:
709 r.raise_for_status()
712class HTTPFile(AbstractBufferedFile):
713 """
714 A file-like object pointing to a remote HTTP(S) resource
716 Supports only reading, with read-ahead of a predetermined block-size.
718 In the case that the server does not supply the filesize, only reading of
719 the complete file in one go is supported.
721 Parameters
722 ----------
723 url: str
724 Full URL of the remote resource, including the protocol
725 session: aiohttp.ClientSession or None
726 All calls will be made within this session, to avoid restarting
727 connections where the server allows this
728 block_size: int or None
729 The amount of read-ahead to do, in bytes. Default is 5MB, or the value
730 configured for the FileSystem creating this file
731 size: None or int
732 If given, this is the size of the file in bytes, and we don't attempt
733 to call the server to find the value.
734 kwargs: all other key-values are passed to requests calls.
735 """
737 def __init__(
738 self,
739 fs,
740 url,
741 session=None,
742 block_size=None,
743 mode="rb",
744 cache_type="bytes",
745 cache_options=None,
746 size=None,
747 loop=None,
748 asynchronous=False,
749 **kwargs,
750 ):
751 if mode != "rb":
752 raise NotImplementedError("File mode not supported")
753 self.asynchronous = asynchronous
754 self.loop = loop
755 self.url = url
756 self.session = session
757 self.details = {"name": url, "size": size, "type": "file"}
758 super().__init__(
759 fs=fs,
760 path=url,
761 mode=mode,
762 block_size=block_size,
763 cache_type=cache_type,
764 cache_options=cache_options,
765 **kwargs,
766 )
768 def read(self, length=-1):
769 """Read bytes from file
771 Parameters
772 ----------
773 length: int or None
774 Read up to this many bytes. If negative or None, read all content to end of
775 file. If the server has not supplied the filesize, attempting to
776 read only part of the data will raise a ValueError.
777 """
778 length = -1 if length is None else length
779 if (
780 (length < 0 and self.loc == 0) # explicit read all
781 # but not when the size is known and fits into a block anyways
782 and not (self.size is not None and self.size <= self.blocksize)
783 ):
784 self._fetch_all()
785 if self.size is None:
786 if length < 0:
787 self._fetch_all()
788 else:
789 length = min(self.size - self.loc, length)
790 return super().read(length)
792 async def async_fetch_all(self):
793 """Read whole file in one shot, without caching
795 This is only called when position is still at zero,
796 and read() is called without a byte-count.
797 """
798 logger.debug(f"Fetch all for {self}")
799 if not isinstance(self.cache, AllBytes):
800 r = await self.session.get(self.fs.encode_url(self.url), **self.kwargs)
801 async with r:
802 r.raise_for_status()
803 out = await r.read()
804 self.cache = AllBytes(
805 size=len(out), fetcher=None, blocksize=None, data=out
806 )
807 self.size = len(out)
809 _fetch_all = sync_wrapper(async_fetch_all)
811 async def async_fetch_range(self, start, end):
812 """Download a block of data
814 The expectation is that the server returns only the requested bytes,
815 with HTTP code 206. If this is not the case, we first check the headers,
816 and then stream the output - if the data size is bigger than we
817 requested, an exception is raised.
818 """
819 logger.debug(f"Fetch range for {self}: {start}-{end}")
820 kwargs = self.kwargs.copy()
821 headers = kwargs.pop("headers", {}).copy()
822 headers["Range"] = f"bytes={start}-{end - 1}"
823 if not any(k.lower() == "accept-encoding" for k in headers) and not any(
824 k.lower() == "accept-encoding" for k in self.session.headers
825 ):
826 # Match the unencoded size from _file_info unless explicitly overridden.
827 headers["Accept-Encoding"] = "identity"
828 logger.debug(f"{self.url} : {headers['Range']}")
830 async def _once():
831 r = await self.session.get(
832 self.fs.encode_url(self.url), headers=headers, **kwargs
833 )
834 async with r:
835 if r.status == 416:
836 if self.size is not None and start < self.size:
837 # CloudFront answers 416 (with an HTML error page as
838 # body) to a range inside the file when range requests
839 # follow each other too quickly; treating it as EOF
840 # would silently truncate the read, retrying with
841 # backoff gets the data. See
842 # https://github.com/fsspec/filesystem_spec/issues/1895
843 raise aiohttp.ClientPayloadError(
844 f"{headers['Range']} of {self.url} reported "
845 f"unsatisfiable, but the file has {self.size} bytes"
846 )
847 # range request outside file
848 return b""
849 r.raise_for_status()
851 # If the server has handled the range request, it should reply
852 # with status 206 (partial content). But we'll guess that a
853 # suitable Content-Range header or a Content-Length no more
854 # than the requested range also mean we have got the desired
855 # range.
856 response_is_range = response_is_requested_range(
857 r.status, r.headers, start, end
858 )
860 if response_is_range:
861 # partial content, as expected
862 out = await r.read()
863 if self.size is not None:
864 expected = min(end, self.size) - start
865 if len(out) < expected:
866 # short body with consistent headers: a truncated
867 # block would otherwise be served from the cache
868 raise aiohttp.ClientPayloadError(
869 f"{headers['Range']} of {self.url} returned "
870 f"{len(out)} bytes, expected {expected}"
871 )
872 elif start > 0:
873 raise ValueError(
874 "The HTTP server doesn't appear to support range "
875 "requests. Only reading this file from the beginning "
876 "is supported. Open with block_size=0 for a streaming "
877 "file interface."
878 )
879 else:
880 # Response is not a range, but we want the start of the
881 # file, so we can read the required amount anyway.
882 cl = 0
883 out = []
884 while True:
885 chunk = await r.content.read(2**20)
886 # data size unknown, let's read until we have enough
887 if chunk:
888 out.append(chunk)
889 cl += len(chunk)
890 if cl > end - start:
891 break
892 else:
893 break
894 out = b"".join(out)[: end - start]
895 return out
897 return await _with_retries(_once, self.fs, f"{self.url} ({headers['Range']})")
899 _fetch_range = sync_wrapper(async_fetch_range)
902magic_check = re.compile("([*[])")
905def has_magic(s):
906 match = magic_check.search(s)
907 return match is not None
910class HTTPStreamFile(AbstractBufferedFile):
911 def __init__(self, fs, url, mode="rb", loop=None, session=None, **kwargs):
912 self.asynchronous = kwargs.pop("asynchronous", False)
913 self.url = url
914 self.loop = loop
915 self.session = session
916 if mode != "rb":
917 raise ValueError
918 self.details = {"name": url, "size": None}
919 super().__init__(fs=fs, path=url, mode=mode, cache_type="none", **kwargs)
921 async def cor():
922 r = await self.session.get(self.fs.encode_url(url), **kwargs).__aenter__()
923 self.fs._raise_not_found_for_status(r, url)
924 return r
926 self.r = sync(self.loop, cor)
927 self.loop = fs.loop
928 self.leftover = b""
929 self._last_read = b""
931 def seek(self, loc, whence=0):
932 """Allow delimiter lookahead to rewind at most one block."""
933 if self.closed:
934 raise ValueError("I/O operation on closed file.")
935 if whence == 1:
936 loc += self.loc
937 elif whence != 0:
938 raise ValueError("Cannot seek streaming HTTP file")
939 if loc == self.loc:
940 return self.loc
941 rewind = self.loc - loc
942 if 0 < rewind <= len(self._last_read):
943 self.leftover = self._last_read[-rewind:] + self.leftover
944 self._last_read = self._last_read[:-rewind]
945 self.loc = loc
946 return self.loc
947 raise ValueError("Cannot seek streaming HTTP file")
949 async def _read(self, num=-1):
950 if self.closed:
951 raise ValueError("I/O operation on closed file.")
952 num = -1 if num is None else num
953 if num < 0:
954 out = self.leftover + await self.r.content.read(num)
955 self.leftover = b""
956 elif len(self.leftover) >= num:
957 out, self.leftover = self.leftover[:num], self.leftover[num:]
958 else:
959 out = self.leftover + await self.r.content.read(num - len(self.leftover))
960 self.leftover = b""
961 if num:
962 # Retain at most one block, even after a large bounded read.
963 self._last_read = out[-self.blocksize :] if num > 0 else b""
964 self.loc += len(out)
965 return out
967 read = sync_wrapper(_read)
969 def readuntil(self, char=b"\n", blocks=None):
970 # The unused delimiter lookahead must fit in the rewind buffer.
971 return super().readuntil(char, min(blocks or self.blocksize, self.blocksize))
973 async def _close(self):
974 self.r.close()
976 def close(self):
977 asyncio.run_coroutine_threadsafe(self._close(), self.loop)
978 self.leftover = self._last_read = b""
979 super().close()
982class AsyncStreamFile(AbstractAsyncStreamedFile):
983 def __init__(
984 self, fs, url, mode="rb", loop=None, session=None, size=None, **kwargs
985 ):
986 self.url = url
987 self.session = session
988 self.r = None
989 if mode != "rb":
990 raise ValueError
991 self.details = {"name": url, "size": None}
992 self.kwargs = kwargs
993 super().__init__(fs=fs, path=url, mode=mode, cache_type="none")
994 self.size = size
996 def seek(self, loc, whence=0):
997 if loc == 0 and whence == 1:
998 return self.loc
999 if loc == self.loc and whence == 0:
1000 return self.loc
1001 raise ValueError("Cannot seek streaming HTTP file")
1003 def seekable(self):
1004 return False
1006 async def read(self, num=-1):
1007 if self.closed:
1008 raise ValueError("I/O operation on closed file.")
1009 num = -1 if num is None else num
1010 if self.r is None:
1011 r = await self.session.get(
1012 self.fs.encode_url(self.url), **self.kwargs
1013 ).__aenter__()
1014 self.fs._raise_not_found_for_status(r, self.url)
1015 self.r = r
1016 out = await self.r.content.read(num)
1017 self.loc += len(out)
1018 return out
1020 async def close(self):
1021 if self.r is not None:
1022 self.r.close()
1023 self.r = None
1024 await super().close()
1027async def get_range(session, url, start, end, file=None, **kwargs):
1028 # explicit get a range when we know it must be safe
1029 kwargs = kwargs.copy()
1030 headers = kwargs.pop("headers", {}).copy()
1031 headers["Range"] = f"bytes={start}-{end - 1}"
1032 r = await session.get(url, headers=headers, **kwargs)
1033 r.raise_for_status()
1034 async with r:
1035 out = await r.read()
1036 if file:
1037 with open(file, "r+b") as f: # noqa: ASYNC230
1038 f.seek(start)
1039 f.write(out)
1040 else:
1041 return out
1044async def _file_info(url, session, size_policy="head", **kwargs):
1045 """Call HEAD on the server to get details about the file (size/checksum etc.)
1047 Default operation is to explicitly allow redirects and use encoding
1048 'identity' (no compression) to get the true size of the target.
1049 """
1050 logger.debug("Retrieve file size for %s", url)
1051 kwargs = kwargs.copy()
1052 ar = kwargs.pop("allow_redirects", True)
1053 head = kwargs.get("headers", {}).copy()
1054 head["Accept-Encoding"] = "identity"
1055 kwargs["headers"] = head
1057 info = {}
1058 if size_policy == "head":
1059 r = await session.head(url, allow_redirects=ar, **kwargs)
1060 elif size_policy == "get":
1061 r = await session.get(url, allow_redirects=ar, **kwargs)
1062 else:
1063 raise TypeError(f'size_policy must be "head" or "get", got {size_policy}')
1064 async with r:
1065 r.raise_for_status()
1067 if "Content-Length" in r.headers:
1068 # Some servers may choose to ignore Accept-Encoding and return
1069 # compressed content, in which case the returned size is unreliable.
1070 if "Content-Encoding" not in r.headers or r.headers["Content-Encoding"] in [
1071 "identity",
1072 "",
1073 ]:
1074 info["size"] = int(r.headers["Content-Length"])
1075 elif "Content-Range" in r.headers:
1076 info["size"] = int(r.headers["Content-Range"].split("/")[1])
1078 if "Content-Type" in r.headers:
1079 info["mimetype"] = r.headers["Content-Type"].partition(";")[0]
1081 if r.headers.get("Accept-Ranges") == "none":
1082 # Some servers may explicitly discourage partial content requests, but
1083 # the lack of "Accept-Ranges" does not always indicate they would fail
1084 info["partial"] = False
1086 info["url"] = str(r.url)
1088 for checksum_field in ["ETag", "Content-MD5", "Digest", "Last-Modified"]:
1089 if r.headers.get(checksum_field):
1090 info[checksum_field] = r.headers[checksum_field]
1092 return info
1095async def _file_size(url, session=None, *args, **kwargs):
1096 if session is None:
1097 session = await get_client()
1098 info = await _file_info(url, session=session, *args, **kwargs)
1099 return info.get("size")
1102file_size = sync_wrapper(_file_size)