network
原生 HTTP 请求、连接状态和下载。
原生 HTTP 客户端:发起 GET/POST 等请求、按块读取 HTTP/SSE 响应、判断网络状态、下载文件到本地路径。
边界:普通get()/request()会把响应体放入内存,适合 REST API 和小响应;长响应、SSE、NDJSON 或需要边下载边处理的内容使用stream()。双向实时通信使用 websocket,可恢复的后台文件下载使用 background_download。
#模块概览
| 项 | 说明 |
|---|---|
| 导入 | import network |
| 适合做什么 | 拉取 JSON API、提交表单、HTTP/SSE 流、检查联网、下载文件 |
| 调用时机 | 放在按钮回调或加载任务;不要写在 AppUI body() 中 |
| 推荐顺序 | is_connected() 提示 → get/post → 检查 response.ok → 解析 json() |
| 蜂窝/低数据 | is_expensive() / is_constrained() 为真时避免自动下大文件 |
#快速开始
下面脚本检查网络,请求 GitHub API 并打印仓库名:
import network
if not network.is_connected():
print("当前无网络")
else:
response = network.get(
"https://api.github.com/repos/python/cpython",
timeout=15,
)
if response.ok:
data = response.json()
print(data["full_name"], "⭐", data["stargazers_count"])
else:
print("请求失败:", response.status, response.text[:200])
#AppUI 示例
请求放在按钮回调里,加载、成功、失败都写回界面。
import appui
import network
state = appui.State(
status="未请求",
network="—",
detail="点击按钮开始",
)
def refresh_network_info():
if not network.is_connected():
state.network = "离线"
return
conn = network.connection_type()
flags = []
if network.is_expensive():
flags.append("计费网络")
if network.is_constrained():
flags.append("低数据模式")
suffix = f"({' · '.join(flags)})" if flags else ""
state.network = f"{conn}{suffix}"
def fetch_repo():
refresh_network_info()
if not network.is_connected():
state.batch_update(status="离线", detail="当前没有可用网络")
return
state.status = "请求中..."
response = network.get(
"https://api.github.com/repos/python/cpython",
timeout=15,
)
if not response.ok:
state.batch_update(status="请求失败", detail=str(response.status))
return
try:
data = response.json()
except Exception as error:
state.batch_update(status="解析失败", detail=str(error))
return
stars = data.get("stargazers_count", "—")
state.batch_update(
status="请求成功",
detail=f"{data.get('full_name', '')} · ⭐ {stars}",
)
def body():
return appui.NavigationStack(
appui.Form([
appui.Section("网络", [
appui.LabeledContent("连接", value=state.network),
appui.LabeledContent("状态", value=state.status),
appui.Text(state.detail).foreground_color("secondaryLabel"),
]),
appui.Section("操作", [
appui.Button("获取 CPython 仓库信息", action=fetch_repo)
.button_style("bordered_prominent"),
], footer="需要网络;真机测试最可靠。"),
]).navigation_title("网络请求")
)
appui.run(body, state=state)
#API 参考
#速查
| API | 作用 |
|---|---|
is_connected() | 是否有可用网络 → bool |
connection_type() | wifi / cellular / ethernet / none / other |
is_expensive() | 是否计费网络(蜂窝/热点) |
is_constrained() | 是否低数据模式 |
get/post/put/delete/patch(...) | HTTP 快捷方法 → Response |
request(method, url, ...) | 通用 HTTP 请求 |
stream(method, url, ...) | 有界、可取消的 HTTP 响应流 → StreamResponse |
download(url, dest_path) | 下载文件 → bool |
#网络状态
import network
if network.is_connected():
print(network.connection_type())
if network.is_expensive():
print("计费网络,避免大流量")
| API | 说明 |
|---|---|
is_connected() | 当前能否联网 |
connection_type() | 连接类型字符串 |
is_expensive() | 蜂窝或热点等 |
is_constrained() | 系统低数据模式 |
#HTTP 请求
request(method, url, headers=None, body=None, json=None, timeout=30) — 通用请求。
快捷方法:get、post、put、delete、patch,参数相同。
response = network.get("https://httpbin.org/get", timeout=15)
response = network.post(
"https://httpbin.org/post",
json={"title": "Hello"},
headers={"Accept": "application/json"},
timeout=20,
)
Response 对象:
| 属性/方法 | 说明 |
|---|---|
status | HTTP 状态码 |
ok | 200–299 时为 True |
text | 响应正文字符串 |
json() | 解析 JSON;非 JSON 会抛异常 |
headers | 响应头字典 |
if response.ok:
data = response.json()
else:
print(response.status, response.text[:200])
#HTTP 流
stream(method, url, *, headers=None, body=None, json=None, timeout=30, read_timeout=30, buffer_limit=1048576) — 等待响应头后立即返回 StreamResponse,正文由 Python 主动按块消费,不会先把完整响应放进内存。
import network
with network.stream(
"GET",
"https://example.com/events",
headers={"Accept": "text/event-stream"},
read_timeout=45,
) as response:
if not response.ok:
print("HTTP 错误:", response.status)
else:
for event in response.iter_sse():
print(event.get("event", "message"), event["data"])
StreamResponse 会对慢消费者施加背压:原生缓冲达到上限后暂停接收,Python 消费后继续。离开 with、调用 close() / cancel(),或当前 MiniApp 运行停止时,原生任务都会取消。
| 属性/方法 | 说明 |
|---|---|
status / ok / headers / url | 响应元数据 |
closed | 流是否已结束或取消 |
iter_bytes(chunk_size=65536, timeout=None) | 逐块返回 bytes;块大小必须为 1 字节至 1 MiB |
iter_text(...) | 用增量解码器返回文本,正确处理跨块 UTF-8 字符 |
iter_lines(..., max_line_chars=1048576) | 逐行返回文本,移除行尾换行符;无换行超长内容会在 1 MiB 上限处失败 |
iter_sse(..., max_event_chars=1048576) | 解析 Server-Sent Events;单事件有 1 MiB 聚合上限 |
read(max_bytes=67108864, ...) | 有明确内存上限地读取剩余正文,返回 bytes |
json(max_bytes=67108864, ...) | 在上限内读取并解析 JSON |
close() / cancel() | 幂等取消并释放原生任务 |
buffer_limit 可设为 64 KiB–4 MiB,只控制原生待消费缓冲,不是完整响应大小;每个运行最多 8 条流、全局最多 64 条,request body 最大 16 MiB、headers 最大 64 KiB。read() 默认 64 MiB且不得超过 256 MiB。等待下一块超过 read_timeout 会抛 TimeoutError;超长行或 SSE 事件会抛出带稳定 code 的 StreamError,不会无限累积内存。
普通 request/get/post 同样限制为 128 个 header、64 KiB header 总量和 16 MiB 文本/JSON body;header 名大小写不敏感且不得重复,也不能自行设置 URLSession 管理的 Host、Content-Length、Connection、Transfer-Encoding。重定向继续按原始能力租约校验,HTTPS 不会降级到 HTTP,跨 origin 时会移除认证、Cookie、API key 和 token/secret 类 header。更大的上传应使用后台传输 V2,而不是在 Python 与 JSON 之间复制整块数据。
#下载
download(url, dest_path, timeout=120) — 下载到本地路径,成功返回 True。
ok = network.download(
"https://httpbin.org/json",
"report.json",
timeout=30,
)
下载失败返回 False;路径需脚本可写。配合 photos 的 save_video 时,先 download 再传本地路径。
#常见错误
| 错误写法 | 后果 | 修正 |
|---|---|---|
在 body() 里发请求 | 每次刷新都联网 | 放进按钮回调 |
不检查 response.ok | 把错误当成功解析 | 先判断 ok 再 json() |
对非 JSON 直接 json() | 解析异常 | try/except 或改读 text |
| 流打开后不关闭 | 原生请求与缓冲占用资源 | 使用 with 或在 finally 调用 close() |
用 read() 无限制接收长流 | 可能占用过多内存 | 用迭代器;确需完整内容时设置合适的 max_bytes |
| 蜂窝网络自动下大文件 | 流量浪费 | 检查 is_expensive() 并提示用户 |
#相关文档
#预期效果
运行示例后,界面应出现文档描述的目标结果;若与预期不符,先看「失败路径」并按返回值或日志排查。