你好，我是坚持分享干货的 EarlGrey，翻译出版过《Python编程无师自通》、《Python并行计算手册》等技术书籍。
如果我的分享对你有帮助，请关注我，一起向上进击。
作者：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）
class SAMIService(Service): _instance_lock = threading.Lock()def __new__(cls, *args, **kwargs):if not hasattr(SAMIService, "_instance"):with SAMIService._instance_lock:if not hasattr(SAMIService, "_instance"): SAMIService._instance = object.__new__(cls)return SAMIService._instancedef __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) @staticmethoddef get_service_info(): api_url = 'open.volcengineapi.com' service_info = ServiceInfo(api_url, {}, Credentials('', '', 'sami', 'cn-north-1'), 10, 10)return service_info @staticmethoddef get_api_info(): api_info = {"GetToken": ApiInfo("POST", "/", {"Action": "GetToken", "Version": "2021-07-27"}, {}, {}), }return api_infodef common_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_jsonexcept Exception as e: res = str(e)try: res_json = json.loads(res)return res_jsonexcept: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 里用 基本是要共享状态的，而用了 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() 的函数啥事都要自己来干。
既然如此不要把它放在类里面了，直接拉出来写成一个函数。
def common_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_jsonexcept Exception as e:# 后面的太可怕了，不要学 ...还记得直接用 requests 怎么发送和接收 JSON 响应吗？
res = requests.post(url, json=body)res_json = res.json()好优雅，好舒服，这么优雅舒服的库怎么被他包成了这样？不要忘了一开始提出的问题，要对请求签名。我们看看 Service.json() 的实现。
def json(self, api, params, body):if not (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() 这个方法：
class VolcSession(requests.Session):def send(self, request, **kwargs):# new_sign 具体实现略，照抄即可 new_sign(request, service_info, credentials)return super().send(request, **kwargs)（重载 Session.prepare_request() 也是一样的效果，区别是在 super() 返回的对象上修改）不知对开始的问题你们心目中的方案是不是这样。
但是，我说但是了，这里最好的方法利用，requests.auth，他的签名是这样的：
class AuthBase:"""Base that all auth implementations derive from"""def __call__(self, r):raise NotImplementedError("Auth hooks must be callable.")接收一个唯一对象 r，这个就是即将要发送的请求，并返回一个新的请求，你可以对它作任何修改，这不就是我们要做的事情吗？签名所需的其他信息，可以作为 __init__ 的初始化参数。那么就可以改写成：
class VolcAuth(AuthBase):def __init__(self, service_info, credentials): self.service_info = service_info self.credentials = credentialsdef __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.pyhttps://github.com/volcengine/volc-sdk-python/blob/main/volcengine/base/Service.py总结这个 SDK 写成这样，可能是直接从别的语言直译过来的。不知从事 code review 的 @piglei 如何看待，能不能过你这关。如果阅读本文的你恰好就是维护这个 SDK 的人被我中伤了我深表抱歉，并绝对不改。
- EOF -
文章已经看到这了，别忘了在右下角点个“赞”和“在看”鼓励哦~
推荐阅读 点击标题可跳转1、Python 项目工程化最佳实践
2、Python 可以比 C 还要快！
3、streamlit，一个超强的 Python 库
4、豆瓣8.9分的C++经典之作，免费送！
5、Python 3.12 版本有什么变化？
最近我开了一家淘宝店，名字叫【打破壁垒】，主打程序开发相关付费素材、工具的共享类商品，帮助大家降低试错和使用成本。欢迎大家关注。
告诉你更多细节干货
欢迎围观我的朋友圈
👆每天更新所想所悟
