PythonIDE Docs
中文
简体中文

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 并打印仓库名:

python
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 示例

请求放在按钮回调里,加载、成功、失败都写回界面。

python
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

#网络状态

python
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) — 通用请求。

快捷方法:getpostputdeletepatch,参数相同。

python
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 对象:

属性/方法说明
statusHTTP 状态码
ok200–299 时为 True
text响应正文字符串
json()解析 JSON;非 JSON 会抛异常
headers响应头字典
python
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 主动按块消费,不会先把完整响应放进内存。

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 事件会抛出带稳定 codeStreamError,不会无限累积内存。

普通 request/get/post 同样限制为 128 个 header、64 KiB header 总量和 16 MiB 文本/JSON body;header 名大小写不敏感且不得重复,也不能自行设置 URLSession 管理的 HostContent-LengthConnectionTransfer-Encoding。重定向继续按原始能力租约校验,HTTPS 不会降级到 HTTP,跨 origin 时会移除认证、Cookie、API key 和 token/secret 类 header。更大的上传应使用后台传输 V2,而不是在 Python 与 JSON 之间复制整块数据。

#下载

download(url, dest_path, timeout=120) — 下载到本地路径,成功返回 True

python
ok = network.download(
    "https://httpbin.org/json",
    "report.json",
    timeout=30,
)

下载失败返回 False;路径需脚本可写。配合 photossave_video 时,先 download 再传本地路径。


#常见错误

错误写法后果修正
body() 里发请求每次刷新都联网放进按钮回调
不检查 response.ok把错误当成功解析先判断 okjson()
对非 JSON 直接 json()解析异常try/except 或改读 text
流打开后不关闭原生请求与缓冲占用资源使用 with 或在 finally 调用 close()
read() 无限制接收长流可能占用过多内存用迭代器;确需完整内容时设置合适的 max_bytes
蜂窝网络自动下大文件流量浪费检查 is_expensive() 并提示用户

#相关文档

文档用途
photos下载视频后 save_video
websocket双向实时消息
后台下载可恢复、可离开前台的文件任务
weatherWeatherKit 联网查询
原生能力入口MiniApp 场景配方

#预期效果

运行示例后,界面应出现文档描述的目标结果;若与预期不符,先看「失败路径」并按返回值或日志排查。