作者：frostming
原文地址：
https://frostming.com/2024/friendly-python-reuse
最近寫了一個 TTS(Text to Speach) 庫
Tetos
，為的就是統一各種雲 TTS 服務的呼叫介面，讓使用者可以用同一套程式碼，只需要變動參數就可以在不同的 TTS 間切換。
計畫地址：https://github.com/frostming/tetos
在實作過程中，我翻閱了很多雲 TTS 服務的介面文件，發現它們介面的設計大相徑庭，有的是 RESTful，有的是偽 RESTful，有的文件裏甚至只讓你用 SDK，沒有 HTTP 介面說明。
本來嘛，我做的工作就是讓使用者可以不用做這些工作，但本篇文章還是想主要吐槽一下火山引擎的介面，和它的 SDK 設計。所以這篇可能不能叫【友好的 Python】了，可以當吐槽大會來看。
提出問題
假設你是一名公有雲廠商 Python SDK 的開發者，你們的介面有一個非常復雜的驗簽機制，你人微言輕，不能質疑，只能按照上面交給你的文件來做。那麽你會怎麽設計這個 SDK 給使用者使用？進一步，不如我們脫離簽名的具體細節，把它抽象出來：
sign(request, randomData, secrets) -> signedRequest
簽名的輸入有三個：HTTP 請求、現場隨機生成的數據，和金鑰數據。輸出是簽名的請求，這個簽名可能修改了請求頭，或是請求體，我們不管它，總之後續就用這個新的請求執行。假如這個 SDK 支持的是
requests
庫，你會怎麽設計呢？不妨先帶著這個思考，來
吃一口屎
看一下火山引擎的 SDK。
下面的程式碼是我直接從火山引擎的介面文件裏截取的。（ https://www.volcengine.com/docs/6489/71995#python ）
classSAMIService(Service):
_instance_lock = threading.Lock()
def__new__(cls, *args, **kwargs):
ifnot hasattr(SAMIService, "_instance"):
with SAMIService._instance_lock:
ifnot hasattr(SAMIService, "_instance"):
SAMIService._instance = object.__new__(cls)
return SAMIService._instance
def__init__(self):
self.service_info = SAMIService.get_service_info()
self.api_info = SAMIService.get_api_info()
super(SAMIService, self).__init__(self.service_info, self.api_info)
@staticmethod
defget_service_info():
api_url = 'open.volcengineapi.com'
service_info = ServiceInfo(api_url, {},
Credentials('', '', 'sami', 'cn-north-1'), 10, 10)
return service_info
@staticmethod
defget_api_info():
api_info = {
"GetToken": ApiInfo("POST", "/", {"Action": "GetToken", "Version": "2021-07-27"}, {}, {}),
}
return api_info
defcommon_json_handler(self, api, body):
params = dict()
try:
body = json.dumps(body)
res = self.json(api, params, body)
res_json = json.loads(res)
return res_json
except Exception as e:
res = str(e)
try:
res_json = json.loads(res)
return res_json
except:
raise Exception(str(e))
if __name__ == '__main__':
sami_service = SAMIService()
sami_service.set_ak(ACCESS_KEY)
sami_service.set_sk(SECRET_KEY)
req = {"appkey": APPKEY, "token_version": AUTH_VERSION, "expiration": 3600}
resp = sami_service.common_json_handler("GetToken", req)
try:
print("response task_id=%s status_code=%d status_text=%s expires_at=%s\n\t token=%s" % (
resp["task_id"], resp["status_code"], resp["status_text"], resp["expires_at"], resp["token"]))
except:
print("get token failed, ", resp)
大病得治
這是一個獲取 Token 的請求，最後使用的是
common_json_handler()
這個函式。一眼看去，你發現一點都不像正常的 Python HTTP 呼叫風格，你以為他是祖傳自建的 HTTP 輪子，但其實不是，它底層還是
requests
，那麽為什麽 SDK 會變得這麽畸形呢？
我們先忽略
set_ak()
,
Singleton
這種從別的語言過來的在 Python 裏毫無必要的寫法，並且也忽略他在
except Exception
邏輯裏返回正常響應的行為（我得咬著後槽牙才能忍，這麽寫是要浸豬籠的）。
我第一個反對的是，為什麽要用繼承 +
staticmethod
的方法來寫，我們知道 Python 裏用 class 基本是要共享狀態的，而用了
staticmethod
就沒得共享了，那麽為什麽不能直接改成下面這樣？
api_url = 'open.volcengineapi.com'
service_info = ServiceInfo(
api_url, {},
Credentials('', '', 'sami', 'cn-north-1'), 10, 10
)
api_info = {
"GetToken": ApiInfo("POST", "/", {"Action": "GetToken", "Version": "2021-07-27"}, {}, {}),
}
sami_service = Service(service_info, api_info)
...
並且閱讀程式碼可知
Credentials
的頭兩個參數就是
access_key
和
secret_key
，那麽直接傳入，不必後面再
set_ak
了。上面這個寫法和之前繼承 +
staticmethod
的效果完全一樣。
好了現在除了
common_json_handler()
以外這個類的成員全被我幹掉了，需要註意到
api_info
裏仿佛包含的是一些請求相關的資訊，依次分別是
method
,
path
,
body
和
headers
之類的東西。下面我們來看看怎麽改掉這個函式。
common_json_handler()
唯一用到的 Service 的方法是
self.json()
，從名字猜測這是一個接收 JSON 響應的方法，註意到 body 和 response 都分別經過了
json.dumps
和
json.loads
，等於這個名為
json()
的函式啥事都要自己來幹。
既然如此不要把它放在類裏面了，直接拉出來寫成一個函式。
defcommon_json_handler(service, api, body):
params = dict()
try:
body = json.dumps(body)
res = service.json(api, params, body)
res_json = json.loads(res)
return res_json
except Exception as e:
# 後面的太可怕了，不要學
...
還記得直接用
requests
怎麽發送和接收 JSON 響應嗎？
res = requests.post(url, json=body)
res_json = res.json()
好優雅，好舒服，這麽優雅舒服的庫怎麽被他包成了這樣？不要忘了一開始提出的問題，要對請求簽名。我們看看
Service.json()
的實作。
defjson(self, api, params, body):
ifnot (api in self.api_info):
raise Exception("no such api")
api_info = self.api_info[api]
r = self.prepare_request(api_info, params)
r.headers['Content-Type'] = 'application/json'
r.body = body
SignerV4.sign(r, self.service_info.credentials)
url = r.build()
resp = self.session.post(url, headers=r.headers, data=r.body,
timeout=(self.service_info.connection_timeout, self.service_info.socket_timeout))
if resp.status_code == 200:
return json.dumps(resp.json())
else:
raise Exception(resp.text.encode("utf-8"))
好家夥，難怪我要自己
json.loads()
呢，
json.dumps(resp.json())
來來來，你過來我保證不打死你。
接著看，這裏出現了關鍵的
SignerV4.sign()
，參數是一個
自己
生成的 request 物件，和上面我抽象的差不多，需要一些請求的資訊和金鑰。這也是為什麽要一個如此奇怪的
api_info
，因為這是簽名需要用的請求的資訊，只好單獨傳遞。
好了問題找到了，搞這麽奇怪，就是因為他自己弄了個請求物件，然後又要費勁把它變成
requests
接受的物件（
r.build()
拿 URL 及
r.headers
,
r.body
）。
那麽請問下，為什麽不能用
requests
內部的請求物件去生成簽名？反正最終是要靠
requests
發送請求，要有的資訊這全都有。就好比你跑馬拉松，補給點都是在跑道必經之處，想象一下你要喝個水還要專門跑岔路去補給點，怎生一個臥槽。
盡量不要自己封裝新的物件，因為你要拷貝原有的內容。
那麽現在要做的事情就清楚了，就是要在請求前修改
requests
即將要發送的請求物件，給它加上簽名資訊。
這其實是一種 interceptor，
requests
有什麽機制實作這個需求呢？我第一想到的是 Event Hook(
https://requests.readthedocs.io/en/latest/user/advanced/#event-hooks
)，但仿佛
requests
沒有
before_request
這個勾點（曾經有），那麽接下來考慮的是多載，由於這個簽名方法是套用在 request 物件上的，所以不同在
get
，
post
之前做文章，因為這兩個方法都還沒產生 request 物件呢，可以多載
Session.send()
這個方法：
classVolcSession(requests.Session):
defsend(self, request, **kwargs):
# new_sign 具體實作略，照抄即可
new_sign(request, service_info, credentials)
return super().send(request, **kwargs)
（多載
Session.prepare_request()
也是一樣的效果，區別是在
super()
返回的物件上修改）不知對開始的問題你們心目中的方案是不是這樣。
但是，我說但是了，這裏最好的方法利用，
requests.auth
，他的簽名是這樣的：
classAuthBase:
"""Base class that all auth implementations derive from"""
def__call__(self, r):
raise NotImplementedError("Auth hooks must be callable.")
接收一個唯一物件
r
，這個就是即將要發送的請求，並返回一個新的請求，你可以對它作任何修改，這不就是我們要做的事情嗎？簽名所需的其他資訊，可以作為
__init__
的初始化參數。那麽就可以覆寫成：
classVolcAuth(AuthBase):
def__init__(self, service_info, credentials):
self.service_info = service_info
self.credentials = credentials
def__call__(self, r):
# new_sign 具體實作略，照抄即可，區別是自訂的 request 物件改成了 requests 的
new_sign(r, self.service_info, self.credentials)
return r
只需要這一個小小的物件即可。利用庫的已存在的數據結構的好處是，我們能最大化保持原來的庫的介面，因為請求方法我們沒有任何侵入。用這個 Auth 物件請求的方法是：
auth = VolcAuth(service_info, credentials)
res = requests.post(url, json=body, auth=auth)
這樣
post()
方法裏的所有參數，包括
data
,
files
,
headers
你可以任意使用，就像用
requests
一樣去調火山的介面，你還可以把建立一個帶 auth 的
Session
，這樣後面呼叫就不用每次都傳
auth
了。無感親膚，就像冰絲內褲一樣。
對一個庫的多載或修改，修改面要越小越好，並盡可能利用庫本身提供的擴充套件方式。
這與上面的方案相比，上面需要繼承
Session
，而利用的
AuthBase
本來就是提供給你擴充套件的，而且建立的物件 Auth 比 Session 小得多。只有當庫擴充套件能力不足時，才考慮前面的方式，一直到無能為力，甚至動用 monkey patch 這種武器。
這裏面的細微優劣，就像你想要車的某個高級功能，你是希望得到一個插到任何車上都能用的零件，還是一台升級好的車，且你不知道它改了哪裏呢？
參考實作
我在 Tetos 裏做了一個針對
httpx
的
Auth
實作，和
requests
的
Auth
作用差不多，有興趣的話甚至可以用一個
Auth
同時支持
httpx
和
requests
兩個庫。
https://github.com/frostming/tetos/blob/15a039f15feda2a3f7ffba7c441b5438f22a6ee4/src/tetos/volc.py#L25-L86
比較一下，這個實作 62 行，加上不超過兩行的呼叫，實作了原來 SignerV4.py 207 行，加上 Service.py 290 行，近 500 行，還沒算上 import 的公共函式，十倍的差距。可見閱讀庫的文件，理清邏輯，是可以大大節省程式碼量的。
https://github.com/volcengine/volc-sdk-python/blob/main/volcengine/auth/SignerV4.py
https://github.com/volcengine/volc-sdk-python/blob/main/volcengine/base/Service.py
總結
這個 SDK 寫成這樣，可能是直接從別的語言直譯過來的。不知從事 code review 的 @piglei 如何看待，能不能過你這關。如果閱讀本文的你恰好就是維護這個 SDK 的人被我中傷了我深表抱歉，並絕對不改。
