Frost's Blog
2384 字
12 分钟
友好的 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。

下面的代码是我直接从火山引擎的接口文档里截取的。

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._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
    def get_service_info():
        api_url = 'open.volcengineapi.com'
        service_info = ServiceInfo(api_url, {},
                                   Credentials('', '', 'sami', 'cn-north-1'), 10, 10)
        return service_info

    @staticmethod
    def get_api_info():
        api_info = {
            "GetToken": ApiInfo("POST", "/", {"Action": "GetToken", "Version": "2021-07-27"}, {}, {}),
        }
        return api_info

    def 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_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_keysecret_key,那么直接传入,不必后面再 set_ak 了。上面这个写法和之前继承 + staticmethod 的效果完全一样。 好了现在除了 common_json_handler() 以外这个类的成员全被我干掉了,需要注意到 api_info 里仿佛包含的是一些请求相关的信息,依次分别是 method, path, bodyheaders 之类的东西。下面我们来看看怎么改掉这个函数。

common_json_handler() 唯一用到的 Service 的方法是 self.json(),从名字猜测这是一个接收 JSON 响应的方法,注意到 body 和 response 都分别经过了 json.dumpsjson.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_json
    except 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,但仿佛 requests 没有 before_request 这个钩子(曾经有),那么接下来考虑的是重载,由于这个签名方法是应用在 request 对象上的,所以不同在 getpost 之前做文章,因为这两个方法都还没产生 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 class 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 = 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 里做了一个针对 httpxAuth 实现,和 requestsAuth 作用差不多,有兴趣的话甚至可以用一个 Auth 同时支持 httpxrequests 两个库。

https://github.com/frostming/tetos/blob/15a039f15feda2a3f7ffba7c441b5438f22a6ee4/src/tetos/volc.py#L25-L86

比较一下,这个实现 62 行,加上不超过两行的调用,实现了原来 SignerV4.py 207 行,加上 Service.py 290 行,近 500 行,还没算上 import 的公共函数,十倍的差距。可见阅读库的文档,理清逻辑,是可以大大节省代码量的。

总结#

这个 SDK 写成这样,可能是直接从别的语言直译过来的。不知从事 code review 的 @piglei 如何看待,能不能过你这关。如果阅读本文的你恰好就是维护这个 SDK 的人被我中伤了我深表抱歉,并绝对不改。

友好的 Python:封装和复用
https://frostming.com/2024/friendly-python-reuse/
作者
Frost Ming
发布于
2024-05-09
许可协议
CC BY-NC-SA 4.0