fetch.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418
  1. """
  2. Support for streaming http requests in emscripten.
  3. A few caveats -
  4. Firstly, you can't do streaming http in the main UI thread, because atomics.wait isn't allowed.
  5. Streaming only works if you're running pyodide in a web worker.
  6. Secondly, this uses an extra web worker and SharedArrayBuffer to do the asynchronous fetch
  7. operation, so it requires that you have crossOriginIsolation enabled, by serving over https
  8. (or from localhost) with the two headers below set:
  9. Cross-Origin-Opener-Policy: same-origin
  10. Cross-Origin-Embedder-Policy: require-corp
  11. You can tell if cross origin isolation is successfully enabled by looking at the global crossOriginIsolated variable in
  12. javascript console. If it isn't, streaming requests will fallback to XMLHttpRequest, i.e. getting the whole
  13. request into a buffer and then returning it. it shows a warning in the javascript console in this case.
  14. Finally, the webworker which does the streaming fetch is created on initial import, but will only be started once
  15. control is returned to javascript. Call `await wait_for_streaming_ready()` to wait for streaming fetch.
  16. NB: in this code, there are a lot of javascript objects. They are named js_*
  17. to make it clear what type of object they are.
  18. """
  19. from __future__ import annotations
  20. import io
  21. import json
  22. from email.parser import Parser
  23. from importlib.resources import files
  24. from typing import TYPE_CHECKING, Any
  25. import js # type: ignore[import-not-found]
  26. from pyodide.ffi import ( # type: ignore[import-not-found]
  27. JsArray,
  28. JsException,
  29. JsProxy,
  30. to_js,
  31. )
  32. if TYPE_CHECKING:
  33. from typing_extensions import Buffer
  34. from .request import EmscriptenRequest
  35. from .response import EmscriptenResponse
  36. """
  37. There are some headers that trigger unintended CORS preflight requests.
  38. See also https://github.com/koenvo/pyodide-http/issues/22
  39. """
  40. HEADERS_TO_IGNORE = ("user-agent",)
  41. SUCCESS_HEADER = -1
  42. SUCCESS_EOF = -2
  43. ERROR_TIMEOUT = -3
  44. ERROR_EXCEPTION = -4
  45. _STREAMING_WORKER_CODE = (
  46. files(__package__)
  47. .joinpath("emscripten_fetch_worker.js")
  48. .read_text(encoding="utf-8")
  49. )
  50. class _RequestError(Exception):
  51. def __init__(
  52. self,
  53. message: str | None = None,
  54. *,
  55. request: EmscriptenRequest | None = None,
  56. response: EmscriptenResponse | None = None,
  57. ):
  58. self.request = request
  59. self.response = response
  60. self.message = message
  61. super().__init__(self.message)
  62. class _StreamingError(_RequestError):
  63. pass
  64. class _TimeoutError(_RequestError):
  65. pass
  66. def _obj_from_dict(dict_val: dict[str, Any]) -> JsProxy:
  67. return to_js(dict_val, dict_converter=js.Object.fromEntries)
  68. class _ReadStream(io.RawIOBase):
  69. def __init__(
  70. self,
  71. int_buffer: JsArray,
  72. byte_buffer: JsArray,
  73. timeout: float,
  74. worker: JsProxy,
  75. connection_id: int,
  76. request: EmscriptenRequest,
  77. ):
  78. self.int_buffer = int_buffer
  79. self.byte_buffer = byte_buffer
  80. self.read_pos = 0
  81. self.read_len = 0
  82. self.connection_id = connection_id
  83. self.worker = worker
  84. self.timeout = int(1000 * timeout) if timeout > 0 else None
  85. self.is_live = True
  86. self._is_closed = False
  87. self.request: EmscriptenRequest | None = request
  88. def __del__(self) -> None:
  89. self.close()
  90. # this is compatible with _base_connection
  91. def is_closed(self) -> bool:
  92. return self._is_closed
  93. # for compatibility with RawIOBase
  94. @property
  95. def closed(self) -> bool:
  96. return self.is_closed()
  97. def close(self) -> None:
  98. if not self.is_closed():
  99. self.read_len = 0
  100. self.read_pos = 0
  101. self.int_buffer = None
  102. self.byte_buffer = None
  103. self._is_closed = True
  104. self.request = None
  105. if self.is_live:
  106. self.worker.postMessage(_obj_from_dict({"close": self.connection_id}))
  107. self.is_live = False
  108. super().close()
  109. def readable(self) -> bool:
  110. return True
  111. def writable(self) -> bool:
  112. return False
  113. def seekable(self) -> bool:
  114. return False
  115. def readinto(self, byte_obj: Buffer) -> int:
  116. if not self.int_buffer:
  117. raise _StreamingError(
  118. "No buffer for stream in _ReadStream.readinto",
  119. request=self.request,
  120. response=None,
  121. )
  122. if self.read_len == 0:
  123. # wait for the worker to send something
  124. js.Atomics.store(self.int_buffer, 0, ERROR_TIMEOUT)
  125. self.worker.postMessage(_obj_from_dict({"getMore": self.connection_id}))
  126. if (
  127. js.Atomics.wait(self.int_buffer, 0, ERROR_TIMEOUT, self.timeout)
  128. == "timed-out"
  129. ):
  130. raise _TimeoutError
  131. data_len = self.int_buffer[0]
  132. if data_len > 0:
  133. self.read_len = data_len
  134. self.read_pos = 0
  135. elif data_len == ERROR_EXCEPTION:
  136. string_len = self.int_buffer[1]
  137. # decode the error string
  138. js_decoder = js.TextDecoder.new()
  139. json_str = js_decoder.decode(self.byte_buffer.slice(0, string_len))
  140. raise _StreamingError(
  141. f"Exception thrown in fetch: {json_str}",
  142. request=self.request,
  143. response=None,
  144. )
  145. else:
  146. # EOF, free the buffers and return zero
  147. # and free the request
  148. self.is_live = False
  149. self.close()
  150. return 0
  151. # copy from int32array to python bytes
  152. ret_length = min(self.read_len, len(memoryview(byte_obj)))
  153. subarray = self.byte_buffer.subarray(
  154. self.read_pos, self.read_pos + ret_length
  155. ).to_py()
  156. memoryview(byte_obj)[0:ret_length] = subarray
  157. self.read_len -= ret_length
  158. self.read_pos += ret_length
  159. return ret_length
  160. class _StreamingFetcher:
  161. def __init__(self) -> None:
  162. # make web-worker and data buffer on startup
  163. self.streaming_ready = False
  164. js_data_blob = js.Blob.new(
  165. [_STREAMING_WORKER_CODE], _obj_from_dict({"type": "application/javascript"})
  166. )
  167. def promise_resolver(js_resolve_fn: JsProxy, js_reject_fn: JsProxy) -> None:
  168. def onMsg(e: JsProxy) -> None:
  169. self.streaming_ready = True
  170. js_resolve_fn(e)
  171. def onErr(e: JsProxy) -> None:
  172. js_reject_fn(e) # Defensive: never happens in ci
  173. self.js_worker.onmessage = onMsg
  174. self.js_worker.onerror = onErr
  175. js_data_url = js.URL.createObjectURL(js_data_blob)
  176. self.js_worker = js.globalThis.Worker.new(js_data_url)
  177. self.js_worker_ready_promise = js.globalThis.Promise.new(promise_resolver)
  178. def send(self, request: EmscriptenRequest) -> EmscriptenResponse:
  179. headers = {
  180. k: v for k, v in request.headers.items() if k not in HEADERS_TO_IGNORE
  181. }
  182. body = request.body
  183. fetch_data = {"headers": headers, "body": to_js(body), "method": request.method}
  184. # start the request off in the worker
  185. timeout = int(1000 * request.timeout) if request.timeout > 0 else None
  186. js_shared_buffer = js.SharedArrayBuffer.new(1048576)
  187. js_int_buffer = js.Int32Array.new(js_shared_buffer)
  188. js_byte_buffer = js.Uint8Array.new(js_shared_buffer, 8)
  189. js.Atomics.store(js_int_buffer, 0, ERROR_TIMEOUT)
  190. js.Atomics.notify(js_int_buffer, 0)
  191. js_absolute_url = js.URL.new(request.url, js.location).href
  192. self.js_worker.postMessage(
  193. _obj_from_dict(
  194. {
  195. "buffer": js_shared_buffer,
  196. "url": js_absolute_url,
  197. "fetchParams": fetch_data,
  198. }
  199. )
  200. )
  201. # wait for the worker to send something
  202. js.Atomics.wait(js_int_buffer, 0, ERROR_TIMEOUT, timeout)
  203. if js_int_buffer[0] == ERROR_TIMEOUT:
  204. raise _TimeoutError(
  205. "Timeout connecting to streaming request",
  206. request=request,
  207. response=None,
  208. )
  209. elif js_int_buffer[0] == SUCCESS_HEADER:
  210. # got response
  211. # header length is in second int of intBuffer
  212. string_len = js_int_buffer[1]
  213. # decode the rest to a JSON string
  214. js_decoder = js.TextDecoder.new()
  215. # this does a copy (the slice) because decode can't work on shared array
  216. # for some silly reason
  217. json_str = js_decoder.decode(js_byte_buffer.slice(0, string_len))
  218. # get it as an object
  219. response_obj = json.loads(json_str)
  220. return EmscriptenResponse(
  221. request=request,
  222. status_code=response_obj["status"],
  223. headers=response_obj["headers"],
  224. body=_ReadStream(
  225. js_int_buffer,
  226. js_byte_buffer,
  227. request.timeout,
  228. self.js_worker,
  229. response_obj["connectionID"],
  230. request,
  231. ),
  232. )
  233. elif js_int_buffer[0] == ERROR_EXCEPTION:
  234. string_len = js_int_buffer[1]
  235. # decode the error string
  236. js_decoder = js.TextDecoder.new()
  237. json_str = js_decoder.decode(js_byte_buffer.slice(0, string_len))
  238. raise _StreamingError(
  239. f"Exception thrown in fetch: {json_str}", request=request, response=None
  240. )
  241. else:
  242. raise _StreamingError(
  243. f"Unknown status from worker in fetch: {js_int_buffer[0]}",
  244. request=request,
  245. response=None,
  246. )
  247. # check if we are in a worker or not
  248. def is_in_browser_main_thread() -> bool:
  249. return hasattr(js, "window") and hasattr(js, "self") and js.self == js.window
  250. def is_cross_origin_isolated() -> bool:
  251. return hasattr(js, "crossOriginIsolated") and js.crossOriginIsolated
  252. def is_in_node() -> bool:
  253. return (
  254. hasattr(js, "process")
  255. and hasattr(js.process, "release")
  256. and hasattr(js.process.release, "name")
  257. and js.process.release.name == "node"
  258. )
  259. def is_worker_available() -> bool:
  260. return hasattr(js, "Worker") and hasattr(js, "Blob")
  261. _fetcher: _StreamingFetcher | None = None
  262. if is_worker_available() and (
  263. (is_cross_origin_isolated() and not is_in_browser_main_thread())
  264. and (not is_in_node())
  265. ):
  266. _fetcher = _StreamingFetcher()
  267. else:
  268. _fetcher = None
  269. def send_streaming_request(request: EmscriptenRequest) -> EmscriptenResponse | None:
  270. if _fetcher and streaming_ready():
  271. return _fetcher.send(request)
  272. else:
  273. _show_streaming_warning()
  274. return None
  275. _SHOWN_TIMEOUT_WARNING = False
  276. def _show_timeout_warning() -> None:
  277. global _SHOWN_TIMEOUT_WARNING
  278. if not _SHOWN_TIMEOUT_WARNING:
  279. _SHOWN_TIMEOUT_WARNING = True
  280. message = "Warning: Timeout is not available on main browser thread"
  281. js.console.warn(message)
  282. _SHOWN_STREAMING_WARNING = False
  283. def _show_streaming_warning() -> None:
  284. global _SHOWN_STREAMING_WARNING
  285. if not _SHOWN_STREAMING_WARNING:
  286. _SHOWN_STREAMING_WARNING = True
  287. message = "Can't stream HTTP requests because: \n"
  288. if not is_cross_origin_isolated():
  289. message += " Page is not cross-origin isolated\n"
  290. if is_in_browser_main_thread():
  291. message += " Python is running in main browser thread\n"
  292. if not is_worker_available():
  293. message += " Worker or Blob classes are not available in this environment." # Defensive: this is always False in browsers that we test in
  294. if streaming_ready() is False:
  295. message += """ Streaming fetch worker isn't ready. If you want to be sure that streaming fetch
  296. is working, you need to call: 'await urllib3.contrib.emscripten.fetch.wait_for_streaming_ready()`"""
  297. from js import console
  298. console.warn(message)
  299. def send_request(request: EmscriptenRequest) -> EmscriptenResponse:
  300. try:
  301. js_xhr = js.XMLHttpRequest.new()
  302. if not is_in_browser_main_thread():
  303. js_xhr.responseType = "arraybuffer"
  304. if request.timeout:
  305. js_xhr.timeout = int(request.timeout * 1000)
  306. else:
  307. js_xhr.overrideMimeType("text/plain; charset=ISO-8859-15")
  308. if request.timeout:
  309. # timeout isn't available on the main thread - show a warning in console
  310. # if it is set
  311. _show_timeout_warning()
  312. js_xhr.open(request.method, request.url, False)
  313. for name, value in request.headers.items():
  314. if name.lower() not in HEADERS_TO_IGNORE:
  315. js_xhr.setRequestHeader(name, value)
  316. js_xhr.send(to_js(request.body))
  317. headers = dict(Parser().parsestr(js_xhr.getAllResponseHeaders()))
  318. if not is_in_browser_main_thread():
  319. body = js_xhr.response.to_py().tobytes()
  320. else:
  321. body = js_xhr.response.encode("ISO-8859-15")
  322. return EmscriptenResponse(
  323. status_code=js_xhr.status, headers=headers, body=body, request=request
  324. )
  325. except JsException as err:
  326. if err.name == "TimeoutError":
  327. raise _TimeoutError(err.message, request=request)
  328. elif err.name == "NetworkError":
  329. raise _RequestError(err.message, request=request)
  330. else:
  331. # general http error
  332. raise _RequestError(err.message, request=request)
  333. def streaming_ready() -> bool | None:
  334. if _fetcher:
  335. return _fetcher.streaming_ready
  336. else:
  337. return None # no fetcher, return None to signify that
  338. async def wait_for_streaming_ready() -> bool:
  339. if _fetcher:
  340. await _fetcher.js_worker_ready_promise
  341. return True
  342. else:
  343. return False