Python 接入企业微信 API

先决定同步还是异步，判断线不是 QPS

这个决定要在写第一行代码之前做完，因为它会沿着调用栈往外传染：一个 async 函数会要求它的调用方也是 async，一路传到框架层。中途改范式的代价，通常比一开始就选对高一个数量级。

判断只有两条。第一条：你的服务已经是什么形状？跑在 WSGI 上（Flask、Django 同步视图）就走同步，跑在 ASGI 上（FastAPI、Starlette）就走异步，跟着框架走，别在一个进程里养两套。第二条：这段逻辑的时间是不是几乎全花在等网络上，而且并发要上百？两个条件同时成立才值得单独为它上 asyncio，只成立一个，线程池更省心。

• 调用量在每分钟几十次这个量级 —— 异步省下的东西根本不是你的瓶颈。
• 链路里还有同步的数据库驱动或同步的第三方库，短期换不掉 —— 它们会把事件循环卡住，你等于付了复杂度没拿到收益。
• 团队里能看懂「事件循环被阻塞」这个症状的人不到一半 —— 这不是玩笑，async 的排障门槛比同步高一档，而故障不挑时间。

把企业微信 API 接进业务系统的多数场景 —— 后台任务发通知、事件回调触发回复、定时同步客户数据 —— 都落在上面这三条里。真正需要异步的是一个进程要同时维持大量出向调用或长连接的形态，先确认你在不在里面，再决定要不要付这个复杂度。

同步路线：requests 的三个默认值会坑你

对 wecomapi 这类统一 Bearer 鉴权的 REST 接口，同步客户端要配的只有三样：超时、连接池、重试语义。requests 好用到让人忘了它当初是给脚本设计的，这三样的默认值搬进常驻服务就是问题。

1. 1默认没有超时。不传 timeout 就是无限等，一个不返回的请求能挂住一个 worker 线程直到重启。timeout 要传二元组，把建连和读取分开设，读取那一段还得比调用你的上游的超时预算短。
2. 2不用 Session 就没有连接复用。每次 requests.post 都重新走一遍 TCP 与 TLS 握手，高频调用下这部分开销能超过业务处理本身。用 Session 加 HTTPAdapter，把每主机的连接上限调到和线程数匹配。
3. 3urllib3 的 Retry 对非幂等方法是保守的。想让 POST 参与重试要显式放开，而一旦放开，就必须自己保证这个请求带幂等键 —— 否则重试等于重复发送，而且是在你完全不知情的库内部发生的。

# 示意逻辑，精确字段与端点以线上接口文档为准
import requests
from requests.adapters import HTTPAdapter

session = requests.Session()
# pool_maxsize 是每个主机的连接上限，按线程数配；只调 pool_connections 没用
session.mount("https://", HTTPAdapter(pool_maxsize=16, max_retries=0))

resp = session.post(
    "https://manager.wecomapi.com/message/sendText",
    headers={"Authorization": f"Bearer {token}"},
    json={"guid": guid, "toId": to_id, "content": "Hello WeCom"},
    timeout=(3, 10),   # (建连, 读取)。不传就是无限等
)

resp.raise_for_status()   # HTTP 层
raw = resp.json()         # 业务层的成败再判一次，判定方式以文档为准

异步路线：httpx 与 asyncio 的四条硬规矩

httpx 用同一套 API 同时覆盖同步与异步，Client 和 AsyncClient 的调用形状基本一致。这是它相对 requests 最实在的价值：将来真要换范式时，改动集中在少数几处，而不是整个调用栈重写。选它的理由应该是这个，不是「它更快」。

• AsyncClient 要长期持有，放在应用的生命周期里或做成单例。每次请求新建一个 client 等于放弃连接池，还会在事件循环里堆一批还没关掉的连接。
• 超时分四段：建连、读、写、从连接池拿连接。第四段最容易漏，而它是池满时唯一的保护 —— 不配的话，池一满，所有协程一起悬在那里，看起来像死锁。
• asyncio.gather 不限并发。要配 Semaphore，并发数从连接池上限倒推。gather 一次铺开一万个协程，和同步里 for 循环怼一万个请求是同一种事故，只是发生得更快。
• 协程里绝不能出现阻塞调用。requests、time.sleep、同步 DB 驱动，任何一个混进来都会停掉整个事件循环。躲不掉的用 asyncio.to_thread 包出去，并给那个线程池单独的上限。

第四条是异步路线上最贵的错误，因为它不报错。一句阻塞调用混进协程，所有会话一起变慢，监控上看不到任何异常，错误率也是零，只有整体延迟在涨 —— 而涨的幅度取决于那句阻塞当天有多慢。上线前用一次真实压测把这类调用揪出来，比事后看火焰图便宜。

回调服务：Flask 与 FastAPI 各自怎么写 ACK

接 wecomapi 的事件回调时，入口该做的只有三件事：验签、持久化、返回 200。站内口径是先快速 ACK 再异步处理，这里补一条顺序上的细节 —— 持久化必须在 ACK 之前完成。ACK 之后再落库，进程恰好在这两步之间被杀掉，这条事件就凭空消失了，而且不会留下任何痕迹。

Flask 走 WSGI，请求返回之后进程里没有「继续跑」的地方。常见的 threading.Thread 起个后台线程的写法能跑，但 worker 回收、发布重启都会静默丢任务，而且丢的时候没有日志。正确形态是入口只写那三件事，重活交给独立消费者 —— Celery、RQ 或你自己写的进程都行，关键是它和 Web 进程的生命周期解耦。

FastAPI 的 BackgroundTasks 看起来正好合适，但它跑在同一个进程、同一个事件循环里，重启同样即丢，而且它执行时占的就是处理下一个回调的那份时间。判断标准很简单：允许丢的收尾动作可以放 BackgroundTasks，事件处理不属于允许丢的那一类。

# 示意逻辑，验签算法与字段以线上接口文档为准
@app.post("/wecom/callback")
async def callback(request: Request):
    raw = await request.body()          # 验签用原始字节
    if not verify_signature(raw, request.headers):
        return Response(status_code=401)

    await store.append(raw)             # 持久化在 ACK 之前，且要足够快
    return Response(status_code=200)    # 快速 ACK，重活交给独立消费者

验签一定要用 await request.body() 拿到的原始字节，不要拿解析后的 dict 重新 json.dumps 一遍。这个坑为什么会表现成「大部分报文正常、偶尔一条验不过」，站内讲 Node 接入那篇展开过，Python 侧的成因与症状完全一样。

多 worker 之后，进程内的东西全都不作数

用 gunicorn 起四个 worker，你的进程内状态就有四份。三样东西必须外置，否则它们在单机开发时全对、上线全错，而且错得很安静：

1. 1凭证缓存与刷新。四个进程各刷各的，还可能互相把对方的结果盖掉。这块的设计取舍见站内讲 Token 管理那篇，这里只强调一点：缓存和刷新锁都要放进程外。
2. 2限速器。进程内的令牌桶乘以进程数就是实际速率，四个 worker 各限每秒十次，对端看到的是每秒四十次。要么集中式计数，要么给每个 worker 静态切分。
3. 3幂等去重表。进程内的 set 在多 worker 下形同虚设，重复投递会有几分之一的概率漏过去 —— 概率型 bug，最难复现。

锁的选择也跟着范式走：threading.Lock 在协程之间不起作用，asyncio.Lock 跨进程无效，跨进程只能用外部存储上的锁。这三者混用是这块最常见的 bug，表现通常是「偶尔重复一次」，隔几天才出现一回，然后被当成灵异事件放着。

上线前的五条自检

1. 1所有出向请求都带了超时，且读取超时短于上游给你的预算。
2. 2连接池上限、线程数（或 Semaphore 并发数）三个值是对齐的，不是各写各的。
3. 3重试只在一层做，POST 参与重试的前提是带了幂等键。
4. 4回调入口只做验签、持久化、ACK，没有任何可能变慢或抛错的业务逻辑。
5. 5凭证缓存、限速器、去重表都在进程外，用多 worker 起一遍验证过。