aiohttps异步HTTPS库：uPyPI+MicroPython一键安装

弁言

做 MicroPython 嵌入式开辟的朋侪，谁没被​内存溢出​、​HTTP/HTTPS 哀求壅闭​、大文件传输崩掉这些破事折磨过？尤其是在 ESP32S3、树莓派 Pico W 这种资源有限的装备上，想对接云端 API、传个音频 / 图片，大概搞个 SSE 流式通讯，稍不留意就 OOM（内存溢出），调试到心态爆炸。
我踩了无数坑才磨出 aiohttps 这个库 —— 它完全基于 MicroPython 原生的 asyncio、socket 和 ssl 模块打造，​零外部依靠​，专门为嵌入式装备的内存瓶颈量身定做。焦点筹划就两个关键点：

• 一是​1KB 分块流式读写​，不管是下载大文件还是上传数据，内存峰值永世卡在 1KB，彻底告别 OOM；
  
• 二是​异步非壅闭​，哀求时每 5ms 就让出 CPU，绝不霸占资源，让装备能同时处置处罚多使命。
  

更知心的是，它直接办理了 MicroPython ure 正则剖析的递归溢出题目，用纯字符串操纵搞定 URL 剖析；还封装了 GET/POST/PUT/DELETE 全量 HTTP 方法，流式下载、SSE 逐行读取、大 Body 分块发送这些高频场景都给你做好了。不管你是给 ESP32S3 做智能终端应用，还是在 Pico W 上对接 LLM、IoT 平台，aiohttps 都能让你少走 90% 的弯路，把精神放在业务逻辑上，而不是和底层网络死磕。
原理

说白了，aiohttps 就是一套给嵌入式装备量身定做的、低内存、非壅闭的 HTTP/HTTPS 协议实现。
MicroPython 的 asyncio 变乱循环最怕壅闭：平凡的 socket.recv() 是同步的，一碰到没数据的情况，就会不停卡着等，把 CPU 占死，其他协程（好比你的 uPyOS 里的 UI 革新、传感器读取）全都会停摆。我是这么改的：

• 先把 socket 设为​非壅闭模式​（sock.setblocking(False)）：如许 recv() 没数据的时间不会卡死，只会返回空值大概抛非常。
  
• 然后套一层循环轮询：每次读不到数据，就 await asyncio.sleep_ms(5) 让出 CPU，给其他使命留时间片。
  
• 这个节奏和官方的 async_websocket_client 完全对齐，哪怕你的 uPyOS 里跑着 10 个联网应用，也不会被单个哀求拖垮。
  

嵌入式装备的内存太金贵了：Pico W 只有 100 多 KB 可用 RAM，ESP32S3 固然强，但也架不住一次性读个 1MB 的文件进内存，直接就炸了。以是焦点筹划就是 **“边用边扔” 的 1KB 分块战略 **：

• ​下载场景​：每次只从 socket 读 1KB 数据，立即写到文件里，写完就开释这块内存。不管文件是 10KB 还是 10MB，内存里永世只有 1KB 的缓存，绝不大概由于文件太大 OOM。
  
• ​上传场景​：假如是传文件，先读文件巨细填到 Content-Length 里，然后每次读 1KB 文件数据，分块发送，每发一块就 sleep_ms(5) 让出 CPU，不会把 socket 缓冲区堵死；传大 JSON/base64 图片的时间，也是按 1KB 分块写，制止大 payload 把非壅闭缓冲区撑爆。
  

MicroPython 自带的 ure 正则引擎是递归实现的，碰到带一堆鉴权参数的长 URL（凌驾 200 字符），直打仗发 maximum recursion depth exceeded 错误，连哀求都发不出去。以是我干脆​弃用了全部正则​，用纯字符串方法剖析 URL：

• 先用 str.startswith 判断是 HTTP 还是 HTTPS 协议；
  
• 用 str.find("//") 跳过协议头，再用 str.find("/") 分离 host 和 path；
  
• 末了用 str.find(":") 分离 host 和 port，全程零递归、零正则，再长的 URL 都不会崩。
  

而且，这里我也做了一些优化：

• 自动根据 URL 的 scheme 选协议：HTTP 直接明文毗连，HTTPS 自动走 TLS 加密，不消用户手动切换。
  
• TLS 握手用原生 ssl.wrap_socket，默认用 cert_reqs=ssl.CERT_NONE（不验证服务端证书），制止加载证书占内存 —— 究竟嵌入式装备存根证书太贫苦，用户要生产情况验证的话，本身改参数就行，默认给个低内存方案。
  
• 也办理了嵌入式 TLS 握手慢的题目：用 lwIP 软件 TLS 第一次握手要 2-4 秒，这个是硬件限定，我在文档里也写清晰了，制止用户误以为是库的 bug。
  

整个库只依靠 MicroPython 内置的 socket、ssl、asyncio 模块，没有任何第三方包。
测试

这里，各人在uPyPI（https://upypi.net/）中搜刮aiohttps：

点击复制安装下令：

粘贴到终端运行下载即可，然后将下面的代码烧录到测试的单片机上：

1. # Python env   : MicroPython v1.23.0
2. # -*- coding: utf-8 -*-
3. # @Time    : 2026/04/15
4. # @Author  : leeqingsui
5. # @File    : main.py
6. # @Description : aiohttps async HTTPS client test for MicroPython on Raspberry Pi Pico 2W
8. # ======================================== 导入相关模块 =========================================
10. import network
11. import asyncio
12. import time
13. import json
14. import ntptime
15. import aiohttps
17. # ======================================== 全局变量 ============================================
19. WIFI_SSID     = "Y/OURSPACE"
20. WIFI_PASSWORD = "qc123456789"
22. # ======================================== 功能函数 ============================================
24. def connect_wifi():
25.     """
26.     连接 WiFi 并返回网络对象。
28.     Returns:
29.         network.WLAN: 已连接的 WLAN 对象；连接失败时返回 None。
31.     ==========================================
33.     Connect to WiFi and return the network object.
35.     Returns:
36.         network.WLAN: Connected WLAN object; None if connection fails.
37.     """
38.     # 初始化 WiFi 为 STA 模式
39.     wlan = network.WLAN(network.STA_IF)
40.     wlan.active(True)
42.     # 避免重复连接
43.     if not wlan.isconnected():
44.         print("Connecting to WiFi: {}".format(WIFI_SSID))
45.         wlan.connect(WIFI_SSID, WIFI_PASSWORD)
47.         timeout = 15
48.         while not wlan.isconnected() and timeout > 0:
49.             time.sleep(1)
50.             timeout -= 1
51.             print("Connecting... {}s remaining".format(timeout))
53.         if wlan.isconnected():
54.             print("WiFi connected, IP: {}".format(wlan.ifconfig()[0]))
55.         else:
56.             print("WiFi connection failed")
57.             return None
58.     else:
59.         print("WiFi already connected")
61.     return wlan
64. def sync_ntp():
65.     """
66.     通过 NTP 同步系统时间。
68.     ==========================================
70.     Sync system time via NTP.
71.     """
72.     for host in ("ntp.aliyun.com", "ntp.tencent.com", "pool.ntp.org"):
73.         try:
74.             ntptime.host = host
75.             ntptime.settime()
76.             t = time.gmtime()
77.             print("NTP synced via {}: {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d} UTC".format(
78.                 host, t[0], t[1], t[2], t[3], t[4], t[5]))
79.             return
80.         except Exception as e:
81.             print("NTP failed ({}): {}".format(host, e))
82.     print("NTP sync unavailable.")
85. async def test_aiohttps():
86.     """
87.     aiohttps 库全量测试，使用 httpbin.org 作为公开测试服务器。
89.     Test 1: GET  -> json()         HTTPS 握手 + 全量读取 + JSON 解析
90.     Test 2: POST -> json()         str 请求体上传 + 服务端回显
91.     Test 3: GET  -> save()         流式下载写文件（4096 字节）
92.     Test 4: GET  -> status 404     非 200 状态码不抛异常
93.     Test 5: GET  -> text           resp.text 属性读取
94.     Test 6: POST -> bytes body     bytes 类型请求体
95.     Test 7: PUT  -> request()      非 GET/POST 方法
96.     Test 8: GET  -> http://        HTTP 明文连接
97.     Test 9: GET  -> iter_lines()   SSE 流式逐行读取
99.     ==========================================
101.     Full test for aiohttps using httpbin.org as the public test server.
102.     """
103.     # 1. 连接 WiFi
104.     if not connect_wifi():
105.         return
107.     # 2. 同步 NTP
108.     sync_ntp()
110.     print("--- aiohttps Test Start ---")
112.     # Test 1: HTTPS GET -> json()
113.     print("[1/8] GET https://httpbin.org/get")
114.     try:
115.         resp = await aiohttps.get("https://httpbin.org/get")
116.         print("  status:", resp.status)
117.         data = await resp.json()
118.         print("  origin IP:", data.get("origin", "?"))
119.         print("  [1/8] PASS" if resp.status == 200 else "  [1/8] FAIL")
120.     except Exception as e:
121.         print("  [1/8] ERROR:", e)
123.     # Test 2: HTTPS POST str body -> json()
124.     print("[2/8] POST https://httpbin.org/post (str body)")
125.     try:
126.         body = json.dumps({"device": "pico2w", "lib": "aiohttps"})
127.         resp = await aiohttps.post(
128.             "https://httpbin.org/post",
129.             headers={"Content-Type": "application/json"},
130.             data=body,
131.         )
132.         data = await resp.json()
133.         echoed = data.get("json", {})
134.         ok = echoed.get("device") == "pico2w" and echoed.get("lib") == "aiohttps"
135.         print("  [2/8] PASS" if ok else "  [2/8] FAIL (echo mismatch)")
136.     except Exception as e:
137.         print("  [2/8] ERROR:", e)
139.     # Test 3: HTTPS GET -> save() streaming download
140.     print("[3/8] GET https://httpbin.org/bytes/4096 -> save test.bin")
141.     try:
142.         resp = await aiohttps.get("https://httpbin.org/bytes/4096")
143.         n = await resp.save("test.bin")
144.         print("  saved:", n, "bytes")
145.         print("  [3/8] PASS" if n == 4096 else "  [3/8] FAIL (expected 4096, got {})".format(n))
146.     except Exception as e:
147.         print("  [3/8] ERROR:", e)
149.     # Test 4: non-200 status code (404)
150.     print("[4/8] GET https://httpbin.org/status/404")
151.     try:
152.         resp = await aiohttps.get("https://httpbin.org/status/404")
153.         resp.close()
154.         print("  status:", resp.status)
155.         print("  [4/8] PASS" if resp.status == 404 else "  [4/8] FAIL (expected 404)")
156.     except Exception as e:
157.         print("  [4/8] ERROR:", e)
159.     # Test 5: resp.text property
160.     print("[5/8] GET https://httpbin.org/encoding/utf8 -> text")
161.     try:
162.         resp = await aiohttps.get("https://httpbin.org/encoding/utf8")
163.         t = await resp.text
164.         print("  text length:", len(t))
165.         print("  [5/8] PASS" if len(t) > 0 else "  [5/8] FAIL (empty text)")
166.     except Exception as e:
167.         print("  [5/8] ERROR:", e)
169.     # Test 6: POST bytes body
170.     print("[6/8] POST https://httpbin.org/post (bytes body)")
171.     try:
172.         body = b"\x00\x01\x02\x03\xff"
173.         resp = await aiohttps.post(
174.             "https://httpbin.org/post",
175.             headers={"Content-Type": "application/octet-stream"},
176.             data=body,
177.         )
178.         await resp.text  # 消费响应体（二进制回显不能 json()）
179.         ok = resp.status == 200
180.         print("  [6/8] PASS" if ok else "  [6/8] FAIL")
181.     except Exception as e:
182.         print("  [6/8] ERROR:", e)
184.     # Test 7: PUT method via request()
185.     print("[7/8] PUT https://httpbin.org/put")
186.     try:
187.         resp = await aiohttps.request(
188.             "PUT", "https://httpbin.org/put",
189.             headers={"Content-Type": "application/json"},
190.             data=json.dumps({"action": "put_test"}),
191.         )
192.         data = await resp.json()
193.         ok = resp.status == 200 and data.get("json", {}).get("action") == "put_test"
194.         print("  [7/8] PASS" if ok else "  [7/8] FAIL")
195.     except Exception as e:
196.         print("  [7/8] ERROR:", e)
198.     # Test 8: HTTP (plain, non-HTTPS)
199.     print("[8/9] GET http://httpbin.org/get (plain HTTP)")
200.     try:
201.         resp = await aiohttps.get("http://httpbin.org/get")
202.         print("  status:", resp.status)
203.         data = await resp.json()
204.         print("  origin IP:", data.get("origin", "?"))
205.         print("  [8/9] PASS" if resp.status == 200 else "  [8/9] FAIL")
206.     except Exception as e:
207.         print("  [8/9] ERROR:", e)
209.     # Test 9: iter_lines() streaming SSE
210.     print("[9/9] GET https://httpbin.org/stream/3 -> iter_lines()")
211.     try:
212.         resp = await aiohttps.get("https://httpbin.org/stream/3")
213.         print("  status:", resp.status)
214.         count = 0
215.         async for line in resp.iter_lines():
216.             line = line.strip()
217.             if not line:
218.                 # 跳过空行（HTTP 分隔行）
219.                 continue
220.             count += 1
221.             print("  line {}: {} bytes".format(count, len(line)))
222.         print("  [9/9] PASS" if count >= 3 else "  [9/9] FAIL (expected >=3 lines, got {})".format(count))
223.     except Exception as e:
224.         print("  [9/9] ERROR:", e)
226.     print("--- aiohttps Test Done ---")
228. # ======================================== 自定义类 ============================================
230. # ======================================== 初始化配置 ===========================================
232. time.sleep(3)
233. print("FreakStudio: aiohttps async HTTPS client test")
235. # ========================================  主程序  ===========================================
237. if __name__ == "__main__":
238.     asyncio.run(test_aiohttps())

复制代码

上面案例中，在树莓派 Pico 2W 硬件平台上完成 aiohttps 异步 HTTPS 客户端全功能验证测试，步伐内置 WiFi 自动毗连、多源 NTP 时间同步功能，并通过 9 项尺度化测试用例，全面验证了库的焦点本领：支持 HTTPS/HTTP 双协议、GET/POST/PUT 全 HTTP 方法，可实现 JSON / 二进制数据收发、大文件流式下载、404 等非 200 状态码兼容、SSE 流式逐行读取等高频场景，全程异步非壅闭运行、内存占用极低，无任何第三方依靠。

免责声明：如果侵犯了您的权益，请联系站长及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金.