友好的 Python：封裝和復用

2024-06-19碼農

點選上方 " Python人工智慧技術 " 關註，星標或者置頂

22點24分準時推播，第一時間送達

後台回復「 大禮包 」，送你特別福利

編輯：樂樂 | 來自：frostming

連結：https://frostming.com/2024/friendly-python-reuse

大家好，我是Python人工智慧技術

最近寫了一個 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 這種武器。

這裏面的細微優劣，就像你想要車的某個高級功能，你是希望得到一個插到任何車上都能用的零件，還是一台升級好的車，且你不知道它改了哪裏呢？