-
Notifications
You must be signed in to change notification settings - Fork 19
Expand file tree
/
Copy pathfetch.py
More file actions
218 lines (173 loc) · 6.74 KB
/
fetch.py
File metadata and controls
218 lines (173 loc) · 6.74 KB
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
"""
This module provides a Python-friendly interface to the
[browser's fetch API](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API),
returning native Python data types and supporting directly awaiting the promise
and chaining method calls directly on the promise.
```python
from pyscript.fetch import fetch
url = "https://api.example.com/data"
# Pattern 1: Await the response, then extract data.
response = await fetch(url)
if response.ok:
data = await response.json()
else:
raise NetworkError(f"Fetch failed: {response.status}")
# Pattern 2: Chain method calls directly on the promise.
data = await fetch(url).json()
```
"""
import json
import js
from pyscript.util import as_bytearray
class _FetchResponse:
"""
Wraps a JavaScript Response object with Pythonic data extraction methods.
This wrapper ensures that data returned from fetch is, if possible, in
native Python types rather than JavaScript types.
"""
def __init__(self, response):
self._response = response
def __getattr__(self, attr):
"""
Provide access to underlying Response properties like ok, status, etc.
"""
return getattr(self._response, attr)
async def arrayBuffer(self):
"""
Get response body as a buffer (memoryview or bytes).
Returns a memoryview in MicroPython or bytes in Pyodide, representing
the raw binary data.
"""
buffer = await self._response.arrayBuffer()
if hasattr(buffer, "to_py"):
# Pyodide conversion.
return buffer.to_py()
# MicroPython conversion.
return memoryview(as_bytearray(buffer))
async def blob(self):
"""
Get response body as a JavaScript Blob object.
Returns the raw JS Blob for use with other JS APIs.
"""
return await self._response.blob()
async def bytearray(self):
"""
Get response body as a Python bytearray.
Returns a mutable bytearray containing the response data.
"""
buffer = await self._response.arrayBuffer()
return as_bytearray(buffer)
async def json(self):
"""
Parse response body as JSON and return Python objects.
Returns native Python dicts, lists, strings, numbers, etc.
"""
return json.loads(await self.text())
async def text(self):
"""
Get response body as a text string.
"""
return await self._response.text()
class _FetchPromise:
"""
Wraps the fetch promise to enable direct method chaining.
This allows calling response methods directly on the fetch promise:
`await fetch(url).json()` instead of requiring two separate awaits.
This feels more Pythonic since it matches typical usage patterns
Python developers have got used to via libraries like `requests`.
"""
def __init__(self, promise):
self._promise = promise
# To be resolved in the future via the setup() static method.
promise._response = None
# Add convenience methods directly to the promise.
promise.arrayBuffer = self.arrayBuffer
promise.blob = self.blob
promise.bytearray = self.bytearray
promise.json = self.json
promise.text = self.text
@staticmethod
def setup(promise, response):
"""
Store the resolved response on the promise for later access.
"""
promise._response = _FetchResponse(response)
return promise._response
async def _get_response(self):
"""
Get the cached response, or await the promise if not yet resolved.
"""
if not self._promise._response:
await self._promise
return self._promise._response
async def arrayBuffer(self):
response = await self._get_response()
return await response.arrayBuffer()
async def blob(self):
response = await self._get_response()
return await response.blob()
async def bytearray(self):
response = await self._get_response()
return await response.bytearray()
async def json(self):
response = await self._get_response()
return await response.json()
async def text(self):
response = await self._get_response()
return await response.text()
def fetch(url, **options):
"""
Fetch a resource from the network using a Pythonic interface.
This wraps JavaScript's fetch API, returning Python-native data types
and supporting both direct promise awaiting and method chaining.
The function takes a `url` and optional fetch `options` as keyword
arguments. The `options` correspond to the JavaScript fetch API's
[RequestInit dictionary](https://developer.mozilla.org/en-US/docs/Web/API/RequestInit),
and commonly include:
- `method`: HTTP method (e.g., `"GET"`, `"POST"`, `"PUT"` etc.)
- `headers`: Dict of request headers.
- `body`: Request body (string, dict for JSON, etc.)
The function returns a promise that resolves to a Response-like object
with Pythonic methods to extract data:
- `await response.json()` to get JSON as Python objects.
- `await response.text()` to get text data.
- `await response.bytearray()` to get raw data as a bytearray.
- `await response.arrayBuffer()` to get raw data as a memoryview or bytes.
- `await response.blob()` to get the raw JS Blob object.
It's also possible to chain these methods directly on the fetch promise:
`data = await fetch(url).json()`
The returned response object also exposes standard properties like
`ok`, `status`, and `statusText` for checking response status.
```python
# Simple GET request.
response = await fetch("https://api.example.com/data")
data = await response.json()
# Method chaining.
data = await fetch("https://api.example.com/data").json()
# POST request with JSON.
response = await fetch(
"https://api.example.com/users",
method="POST",
headers={"Content-Type": "application/json"},
body=json.dumps({"name": "Alice"})
)
result = await response.json()
# Check response status codes.
response = await fetch("https://api.example.com/data")
if response.ok:
# Status in the range 200-299.
data = await response.json()
elif response.status == 404:
print("Resource not found")
else:
print(f"Error: {response.status} {response.statusText}")
```
"""
# Convert Python dict to JavaScript object.
js_options = js.JSON.parse(json.dumps(options))
# Setup response handler to wrap the result.
def on_response(response, *_):
return _FetchPromise.setup(promise, response)
promise = js.fetch(url, js_options).then(on_response)
_FetchPromise(promise)
return promise