不知不觉距离上一篇日志已经隔了整整一个暑假啊,想必学校里的数学课都已经上完 两章内容了吧现在,一直在打酱油的日子过得还真快捏~

好啦言归正传,最近做了一下OAuth登录,所以不断在和各种平台的API打交道,有国 产的也有进口的,有用REST接口的也有用其他RPC协议的……咳咳,吐槽留 着另写一篇文吧。今天的主题是对付HTTP和REST接口的Python神器requests模 块。既然是神器,无论文档还是扩展都是一流地多,我就说点我自己看文 档没看出来的东西吧~

第一个程序

既然文档里有Quickstart,就从那里开始好了。接口好像很简单嘛,先写个 简化版的Github登录呗:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def sign_in(username, password):
    jsons = json.dumps({
        'scopes': ['gist'],
        'note': AUTH_NOTE,
    }).encode('utf8')
    r = requests.post(
        get_api_url('/authorizations'),
        data=jsons,
        headers=API_HEADERS,
        auth=(username, password))
    json_obj = r.json()
    r.close()

    token = None

    if 'errors' in json_obj:
        auth_exists = False
        for e in json_obj['errors']:
            if e['code'] == 'already_exists':
                auth_exists = True
                break

        if auth_exists:
            token = get_token(username, password, AUTH_NOTE)
    else:
        message = check_message(json_obj)
        if message is not None:
            logger.error(_('Authentication failed: %s'), message)
        else:
            token = Token(token_id=json_obj['id'], token=json_obj['token'])

    return token


def get_token(username, password, note):
    r = requests.get(
        get_api_url('/authorizations'),
        headers=API_HEADERS,
        auth=(username, password))
    auth_list = r.json()
    r.close()

    message = check_message(auth_list)
    if message is not None:
        logger.error(_('Authentication failed: %s'), message)
        return None

    for a in auth_list:
        if a['note'] == note:
            return Token(token_id=a['id'], token=a['token'])
    return None

OK,sign_in首先发了一个创建token的POST请求,然后做错误处理:如果token已经存在 的话就用get_token把旧token拿过来。不知道各位看到这里会怎么想,但是当时写完这 货之后我隐隐觉得哪里不妥:测试该怎么写?

这些代码和requests模块紧紧地耦合在一起了,我写的测试除了测我自己的代码之外,还 要顺带把requests也测一遍吗?这不坑爹吗?

重构

我开始翻requests源码,想要找到一个可以更改请求行为的扩展点,像覆盖个什么类的方 法就可以让requests.get(...)直接返回测试数据之类的。回想起来我也不清楚当时的 第一反应为什么是这个,也许是前段时间受了aiohttp影响的关系。

源码看下去之后发现requests.get(...)其实是建了一个requests.Session对象,然 后再通过这个session发的请求。但是requests.get(...)这类函数是直接写死了调用 requests.Session这个类的,所以别说碰了,我看都没看到那个session,请求就已经 发出去了。

如果没有可以改变session行为的接口,应该就意味着我可以不调用requests.get(...), 而是自己在程序里构造一个session拿来用吧?回去看文档,赫然发现Advanced主题 下第一节就讲的是Session对象……

文档Quickstart后面接的就是Advanced哦……你们判断什么内容是Advanced的标准也太奇怪 了吧……Session对象什么的怎么样都应该放在Quickstart那边吧……果然只看Quickstart什么 的还真的不能愉快地start起来啊……

根据文档说明,Session对象可以保存任意header、query string甚至是cookies,然后 添加到它所发出的每个请求里。既然你能存,那就让你存吧~重构后我把认证信息和特 殊header都放到一个Session子类里:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class GistBasicAuthSession(requests.Session):
    def __init__(self, username, password):
        super().__init__()

        headers = {
            'Accept': 'application/vnd.github.v3+json',
        }
        self.headers.update(headers)
        self.auth = (username, password)


def sign_in(session):
    jsons = json.dumps({
        'scopes': ['gist'],
        'note': AUTH_NOTE,
    }).encode('utf8')
    r = session.post(get_api_url('/authorizations'), data=jsons)
    json_obj = r.json()
    r.close()

    token = None

    if 'errors' in json_obj:
        auth_exists = False
        for e in json_obj['errors']:
            if e['code'] == 'already_exists':
                auth_exists = True
                break

        if auth_exists:
            token = get_token(session, AUTH_NOTE)
    else:
        message = check_message(json_obj)
        if message is not None:
            logger.error(_('Authentication failed: %s'), message)
        else:
            token = json_obj['token']

    return token


def get_token(session, note):
    r = session.get(get_api_url('/authorizations'))
    auth_list = r.json()
    r.close()

    message = check_message(auth_list)
    if message is not None:
        logger.error(_('Authentication failed: %s'), message)
        return None

    for a in auth_list:
        if a['note'] == AUTH_NOTE:
            return a['token']
    return None

这样,对requests的依赖变成了对session接口的依赖,在测试的时候我就可以 mock一个session对象扔进去。这其实就是依赖注入,程序不应该直接依赖一段 具体的代码,而应该依赖抽象类,或者依赖代码的引用,这样具体代码随时可 以被替换而不需要更改依赖者的逻辑。

写到这突然想起来,很多年前我还有在写Linux和Windows驱动的时候,看到的 各种driver对象和device对象,那也正是内核的依赖被抽象之后得到的东西, 只是那时候我没有关心过里面的关系,反而觉得很自然。

后记

文档真的不能只看Quickstart……你觉得自己很聪明,但是写文档的人不一定这 么想。

另一方面,写文档的时候拜托在第一页的例子就给出best practice,或者至少 单独用一节来说明。我相信在一个程序里不断看到这种挂满参数的代码会满恶 心的:

1
2
3
4
r = requests.get(
    get_api_url('/authorizations'),
    headers=API_HEADERS,
    auth=(username, password))

反之这样就好很多:

1
r = session.get(get_api_url('/authorizations'))

既然要做,为什么不第一次就把事情做对呢?既然提供了这样的便利,为什么 不让别人第一次就能用上呢?