点击上方“Deephub Imba”,关注公众号,好文章不错过 !
现在几乎每个大语言模型都在提供 API 服务,写 Python 脚本调用各种接口的场景只会越来越多。可真正上手的时候,很多人还是打开文档滚动半个小时复制粘贴一段勉强能用的代码片段。
下面是十个处理 API 的技巧,能省下不少时间,让脚本更健壮,也许还能让那个自认为是 REST 高手的同事高看一眼。
1、自动重试失败的请求(因为 API 总喜欢随机失败)
API 有时候会闹脾气:超时、限流、随机的 500 错误。与其像疯子一样不停刷新,不如加一段重试逻辑:
importrequestsfromrequests.adaptersimportHTTPAdapterfromurllib3.util.retryimportRetrysession=requests.Session() retry=Retry(total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504]) session.mount('http://', HTTPAdapter(max_retries=retry)) session.mount('https://', HTTPAdapter(max_retries=retry)) response=session.get("https://api.example.com/data") print(response.json())backoff_factor=1 意味着每次重试等待的时间会更长——1 秒、2 秒、4 秒;不用自己手写指数退避。
2、在本地缓存 API 响应(不要浪费请求次数)
API 有严格的速率限制时,缓存能解决大部分问题,Python 让这件事变得很简单:
importrequestsimportrequests_cacherequests_cache.install_cache('api_cache', expire_after=3600) r=requests.get('https://api.github.com/users/octocat') print(r.json())测试代码的时候不会每次都轰炸 GitHub 的服务器。
3、环境变量优于硬编码密钥
用 python-dotenv 和 .env 文件就不会把 API 密钥直接写在脚本里(API_KEY = "123abcSECRET"),
importosfromdotenvimportload_dotenvload_dotenv() API_KEY=os.getenv("API_KEY")4、用 Pydantic 做 Schema 校验(信任,但也要验证)
API 有时候会撒谎。期望拿到的是 "price": 12.5,结果拿到的却是 "price": "twelve"。Pydantic 能让人保持理智:
frompydanticimportBaseModelclassItem(BaseModel): name: strprice: floatdata= {"name": "Apple", "price": "12.5"} item=Item(**data) print(item.price) # 12.5 as float5、无烦恼的分页处理
大多数 API 会对结果分页。与其手动拼接 URL,不如优雅地循环:
importrequestsdeffetch_all(url): whileurl: r=requests.get(url).json() yieldfromr['results'] url=r.get('next') foriteminfetch_all("https://swapi.dev/api/people/"): print(item['name'])干净,Pythonic,不用写没完没了的 while True。
6、像高手一样处理 WebSocket
有些 API 会推送实时数据流,比如股票、加密货币、聊天机器人。不要轮询,去订阅:
importwebsocketimportjsondefon_message(ws, message): data=json.loads(message) print("Live:", data) ws=websocket.WebSocketApp("wss://ws-feed.pro.coinbase.com", on_message=on_message) ws.run_forever()曾经用这个方法搭过一个实时比特币行情器,没有延迟,没有浪费的请求,就是纯粹的实时数据。
7、对自己的调用做速率限制(讲点礼貌)
像机关枪一样疯狂发请求,很多 API 会直接把你封掉。用 ratelimit可以给自己限速:
fromratelimitimportlimits, sleep_and_retry@sleep_and_retry@limits(calls=5, period=60) # 5 requests per minute defcall_api(): print("API call")这就像是在脚本里设置减速带。
8、并行 API 调用(因为等待很无聊)
为什么要一个接一个地等 50 个请求跑完?用 concurrent.futures:
importrequestsfromconcurrent.futuresimportThreadPoolExecutorurls= [f"https://api.github.com/users/{u}"foruin ["octocat", "torvalds", "pjhyett"]] deffetch(url): returnrequests.get(url).json() withThreadPoolExecutor() asexecutor: results=list(executor.map(fetch, urls)) print([r['login'] forrinresults])靠这个方法把一个原本要跑 2 分钟的脚本缩短到了 10 秒。
9、用于开发环境的 API 模拟(没网络?没问题)
有时候是 API 挂了,用 responses 模拟接口,测试能通过,代码能运行,不用依赖不稳定的外部服务器:
importrequestsimportresponsesurls= [f"https://api.github.com/users/{u}"foruin ["octocat", "torvalds", "pjhyett"]] deffetch(url): returnrequests.get(url).json() withThreadPoolExecutor() asexecutor: results=list(executor.map(fetch, urls)) print([r['login'] forrinresults]10、 轻松搞定 GraphQL
REST 不错,但现在 GraphQL 也无处不在了。Python 的 gql 库让这件事变得毫不费力:
fromgqlimportgql, Clientfromgql.transport.requestsimportRequestsHTTPTransporttransport=RequestsHTTPTransport(url="https://countries.trevorblades.com/") client=Client(transport=transport, fetch_schema_from_transport=True) query=gql("""{ country(code: "US") { name capital } }""") result=client.execute(query) print(result)以前一直觉得 GraphQL 让人头疼,现在它不过是另一个 Python 字典而已。
by Steve M.
喜欢就关注一下吧:
点个 在看 你最好看!