1
"""
2
Monkeypatch aiohttp to introduce Fetch API support.
3
4
Based on https://github.com/pyodide/pyodide/issues/3711#issuecomment-1773523301
5
with some modifications.
6
"""
7
 
8
# ruff: noqa: PLR0913, TRY301, TRY300
9
 
10
from collections.abc import Iterable
11
from contextlib import suppress
12
from typing import Any
13
 
14
from aiohttp import ClientSession, ClientTimeout, CookieJar, InvalidURL, hdrs, payload
15
from aiohttp.client_reqrep import _merge_ssl_params
16
from aiohttp.helpers import TimeoutHandle, get_env_proxy_for_url, strip_auth_from_url
17
from multidict import CIMultiDict, istr
18
from yarl import URL
19
 
20
 
21
class Content:
22
    __slots__ = ("_exception", "_jsresp")
23
 
24
    def __init__(self, _jsresp):
25
        self._jsresp = _jsresp
26
        self._exception = None
27
 
28
    async def read(self):
29
        if self._exception:
30
            raise self._exception
31
        buf = await self._jsresp.arrayBuffer()
32
        self._jsresp = None
33
        return buf.to_bytes()
34
 
35
    def exception(self):
36
        return self._exception
37
 
38
    def set_exception(self, exc: BaseException) -> None:
39
        self._exception = exc
40
 
41
 
42
async def _request(
43
    self,
44
    method: str,
45
    str_or_url,
46
    *,
47
    params=None,
48
    data: Any = None,
49
    json: Any = None,
50
    cookies=None,
51
    headers=None,
52
    skip_auto_headers: Iterable[str] | None = None,
53
    auth=None,
54
    allow_redirects: bool = True,
55
    max_redirects: int = 10,
56
    compress: str | None = None,
57
    chunked: bool | None = None,
58
    expect100: bool = False,
59
    raise_for_status=None,
60
    read_until_eof: bool = True,
61
    proxy=None,
62
    proxy_auth=None,
63
    timeout=None,
64
    verify_ssl: bool | None = None,
65
    fingerprint: bytes | None = None,
66
    ssl_context=None,
67
    ssl=None,
68
    proxy_headers=None,
69
    trace_request_ctx=None,
70
    read_bufsize: int | None = None,
71
):
72
    # NOTE: timeout clamps existing connect and read timeouts.  We cannot
73
    # set the default to None because we need to detect if the user wants
74
    # to use the existing timeouts by setting timeout to None.
75
 
76
    if self.closed:
77
        raise RuntimeError("Session is closed")
78
 
79
    ssl = _merge_ssl_params(ssl, verify_ssl, ssl_context, fingerprint)
80
 
81
    if data is not None and json is not None:
82
        raise ValueError("data and json parameters can not be used at the same time")
83
    elif json is not None:
84
        data = payload.JsonPayload(json, dumps=self._json_serialize)
85
 
86
    history = []
87
    version = self._version
88
    params = params or {}
89
 
90
    # Merge with default headers and transform to CIMultiDict
91
    headers = self._prepare_headers(headers)
92
    proxy_headers = self._prepare_headers(proxy_headers)
93
 
94
    try:
95
        url = self._build_url(str_or_url)
96
    except ValueError as e:
97
        raise InvalidURL(str_or_url) from e
98
 
99
    skip_headers = set(self._skip_auto_headers)
100
    if skip_auto_headers is not None:
101
        for i in skip_auto_headers:
102
            skip_headers.add(istr(i))
103
 
104
    if proxy is not None:
105
        try:
106
            proxy = URL(proxy)
107
        except ValueError as e:
108
            raise InvalidURL(proxy) from e
109
 
110
    if timeout is None:
111
        real_timeout = self._timeout
112
    elif not isinstance(timeout, ClientTimeout):
113
        real_timeout = ClientTimeout(total=timeout)  # type: ignore[arg-type]
114
    else:
115
        real_timeout = timeout
116
    # timeout is cumulative for all request operations
117
    # (request, redirects, responses, data consuming)
118
    tm = TimeoutHandle(self._loop, real_timeout.total)
119
    handle = tm.start()
120
 
121
    if read_bufsize is None:
122
        read_bufsize = self._read_bufsize
123
 
124
    traces = []
125
 
126
    timer = tm.timer()
127
    try:
128
        with timer:
129
            url, auth_from_url = strip_auth_from_url(url)
130
            if auth and auth_from_url:
131
                raise ValueError(
132
                    "Cannot combine AUTH argument with credentials encoded in URL"
133
                )
134
 
135
            if auth is None:
136
                auth = auth_from_url
137
            if auth is None:
138
                auth = self._default_auth
139
            # It would be confusing if we support explicit
140
            # Authorization header with auth argument
141
            if auth is not None and hdrs.AUTHORIZATION in headers:
142
                raise ValueError(
143
                    "Cannot combine AUTHORIZATION header "
144
                    "with AUTH argument or credentials "
145
                    "encoded in URL"
146
                )
147
 
148
            all_cookies = self._cookie_jar.filter_cookies(url)
149
 
150
            if cookies is not None:
151
                tmp_cookie_jar = CookieJar()
152
                tmp_cookie_jar.update_cookies(cookies)
153
                req_cookies = tmp_cookie_jar.filter_cookies(url)
154
                if req_cookies:
155
                    all_cookies.load(req_cookies)
156
 
157
            if proxy is not None:
158
                proxy = URL(proxy)
159
            elif self._trust_env:
160
                with suppress(LookupError):
161
                    proxy, proxy_auth = get_env_proxy_for_url(url)
162
 
163
            req = self._request_class(
164
                method,
165
                url,
166
                params=params,
167
                headers=headers,
168
                skip_auto_headers=skip_headers,
169
                data=data,
170
                cookies=all_cookies,
171
                auth=auth,
172
                version=version,
173
                compress=compress,
174
                chunked=chunked,
175
                expect100=expect100,
176
                loop=self._loop,
177
                response_class=self._response_class,
178
                proxy=proxy,
179
                proxy_auth=proxy_auth,
180
                timer=timer,
181
                session=self,
182
                ssl=ssl,
183
                proxy_headers=proxy_headers,
184
                traces=traces,
185
            )
186
 
187
            req.response = resp = req.response_class(
188
                req.method,
189
                req.original_url,
190
                writer=None,
191
                continue100=req._continue,
192
                timer=req._timer,
193
                request_info=req.request_info,
194
                traces=req._traces,
195
                loop=req.loop,
196
                session=req._session,
197
            )
198
            from js import Headers, fetch
199
 
200
            from pyodide.ffi import to_js
201
 
202
            body = None
203
            if req.body:
204
                body = to_js(req.body._value)
205
            jsheaders = Headers.new()
206
            for k, v in headers.items():
207
                jsheaders.append(k, v)
208
            jsresp = await fetch(
209
                str(req.url), method=req.method, headers=jsheaders, body=body
210
            )
211
            resp.version = version
212
            resp.status = jsresp.status
213
            resp.reason = jsresp.statusText
214
            # This is not quite correct in handling of repeated headers
215
            resp._headers = CIMultiDict(jsresp.headers)
216
            resp._raw_headers = tuple(tuple(e) for e in jsresp.headers)
217
            resp.content = Content(jsresp)
218
 
219
        # check response status
220
        if raise_for_status is None:
221
            raise_for_status = self._raise_for_status
222
 
223
        if raise_for_status is None:
224
            pass
225
        elif callable(raise_for_status):
226
            await raise_for_status(resp)
227
        elif raise_for_status:
228
            resp.raise_for_status()
229
 
230
        # register connection
231
        if handle is not None:
232
            if resp.connection is not None:
233
                resp.connection.add_callback(handle.cancel)
234
            else:
235
                handle.cancel()
236
 
237
        resp._history = tuple(history)
238
 
239
        for trace in traces:
240
            await trace.send_request_end(
241
                method, url.update_query(params), headers, resp
242
            )
243
        return resp
244
 
245
    except BaseException as e:
246
        # cleanup timer
247
        tm.close()
248
        if handle:
249
            handle.cancel()
250
            handle = None
251
 
252
        for trace in traces:
253
            await trace.send_request_exception(
254
                method, url.update_query(params), headers, e
255
            )
256
        raise
257
 
258
 
259
ClientSession._request = _request