用Python优雅的封装Upsource HTTP API

发表于2020-04-28阅读时长21分

背景

Upsource是一个非常出色的CodeReview工具，在和其他系统联动时需要通过API进行访问，于是使用python对其API进行封装，过程中一步一步思考如何使用python装饰器简化封装，达到十分简洁的封装。
本文主要介绍利用动态生成函数和装饰器两种方法简化API封装的实现。

Upsource HTTP API说明

1
2
3
4
5
6
7
8
Upsource API is an RPC-style HTTP API. You can make calls using HTTP GET and POST. All data is sent and received as JSON. While the RPC methods don't enforce the use of a specific HTTP method, we recommend that you conform to HTTP semantics by using GET for retrieving data (e.g. getRevisionsList) and POST for modifying data (e.g. createReview).

All timestamps are Unix timestamps, the number of seconds that have elapsed since January 1, 1970 (midnight UTC/GMT).

To invoke a method of the Upsource API using HTTP GET, make the following request:

http://your-upsource-host/~rpc/methodName?params={JSON-encoded-params}
To make a POST request, use the same URL but pass the request payload in the POST body instead. A Content-Type of application/json should be set.

根据文档描述，每个接口的调用包含methodName和JSON格式的参数，于是我们开始用python进行封装。

简单实现

最简单的就是实现就是每个API加个函数

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
class Upsource:
    ENDPOINT = "https://upsource.xxxx.com"

    def __init__(self, user, passwd, retry_times = 3):
        self.__auth = HTTPBasicAuth(user, passwd)
        self.retry_times = retry_times

    # HTTP请求
    def _request(self, rpc_name, data):
        suffix = '/~rpc/{}'.format(rpc_name)
        url = self.ENDPOINT + re.sub('/+', '/', suffix)

        headers = {'Content-Type': 'application/json; charset=utf-8'}
        proxies = { "http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"}
        retry = self.retry_times

        info('{}, request: {}'.format(rpc_name, data))

        while retry > 0:
            retry -= 1
            resp = None
            try:
                resp = requests.post(url, data=json.dumps(data), auth=self.__auth, headers=headers, verify=False)
                if resp and resp.status_code == 200:
                    return json.loads(resp.text)['result']
                raise Exception('request failed')
            except Exception as e:
                error('request upsource failed!, rpc_name: {}, request: {}, resp: {}, error:{}'.format(rpc_name, data, resp.text, str(e)))
            time.sleep(1)

    # getBranchInfo
    def get_branch_info(self, project_id, branch):
        data = {'projectId': project_id,
                'branch': branch}
        return self._request('getBranchInfo', data)

上述实现每个API都需要实现一个函数做转换，有没有优化的空间？
下面我们介绍两种方法进行优化：

  • 动态生成函数
  • 装饰器

动态生成函数

核心思路

python使用locals()可以获取当前上下文
exec()可以执行在字符串中的python代码，所以可以动态生成函数，然后利用setattr()将生成的函数

实现探索

我们可以利用下划线转驼峰，将参数名的转换做优化，配合locals()获取参数列表
下划线转驼峰函数

1
2
def underline2hump(underline_str):
    return ''.join([x.capitalize() for x in underline_str.split('_') if x])
1
2
3
4
5
def get_branch_info(project_id, branch):
    print(locals())
    
>>> get_branch_info("123456", "master")
{'project_id': '123456', 'branch': 'master'}

于是，每个API函数可以优化成这样

1
2
3
4
5
6
7
def get_branch_info(self, project_id, branch):
    m = locals()
    m.pop('self')
    data = {}
    for k,v in iteritems(m):
        data[underline2hump(k)] = v
    return self._request(underline2hump(__name__), data)

函数体变成公共的了， 所以我们可以再写个公共函数，这样每个API都可以调用这个公共函数

1
2
3
4
5
def request(self, data):
    rpc_name = sys._getframe(1).f_code.co_name
    resp = self._request(rpc_name, data)
    info(pprint.pformat(resp))
    return resp

每个API函数调用request， 获取上一层调用名，请求参数透传；
注意: 这里不能用locals()，应为locals只是针对调用栈的本层命名空间，暂时没找到方法获取上一层调用函数的参数列表
所以每个API函数应该都有如下函数体

1
2
3
4
def api_name(self, params1, params2, ...):
    m = locals();
    m.pop('self');
    return self.request(m);

我们可以利用python的exec()将字符串代码执行，这样我们就可以动态生成形如上述格式的函数，并且函数名和参数可变

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
REQUEST_MAP = {
    'getProjectInfo': ['projectId'],
    'getCodeReviewPatterns': [],
    'getRevisionsList': ['projectId', 'limit', 'skip = 0', 'requestGraph = False'],
    'getRevisionsListFiltered': ['projectId', 'query', 'limit', 'skip = 0', 'requestGraph = False'],
    'getRevisionInfo': ['projectId', 'revisionId'],
    'getBranchInfo': ['projectId', 'branch'],
    'getBranchGraph': ['projectId', 'branch'],
    'getBranches': ['projectId', 'query', 'limit', 'sortBy = "updated"'],
    'findCommits': ['commits', 'requestChanges = False', 'limit = 10'],
    'getReviews': ['limit', 'query = "*"', 'sortBy = "updated"', 'projectId = "herohub-platform"', 'skip = 0'],
}
# dynamic gen class function from REQUEST_MAP
for k, v in self.REQUEST_MAP.items():
    fn_str = '''def {}(self, {}): m = locals(); m.pop('self'); return self.request(m);'''.format(k, ', '.join(v))
    info('class: {}, gen function: {}'.format(self.__class__.__name__, fn_str))
    exec(fn_str)
    setattr(Upsource, k, locals().get(k))

由此，我们只需要修改REQUEST_MAP就能实现API封装。
这个实现有个小缺点：由于函数是运行时生成的，所以编辑器没法索引定义，可能导致误报。

使用装饰器进行优化

核心思路

利用装饰器+inspect库获取函数参数名及参数值，进行参数预处理，然后进行调用

实现探索

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
import inspect

def request(func):
    def wrapper(*args, **kw):
        print("locals: {}".format(locals()))
        print("args: {}".format(args))
        print("kw: {}".format(kw))
        spec = inspect.getfullargspec(func)
        print(spec)
        
        data = {}
        for idx, k in enumerate(spec.args):
            if idx >= len(args):
                break
            data[k] = args[idx]
            
        for idx, k in enumerate(spec.args[len(args):]):
            data[k] = spec.defaults[idx]
            
        method = getattr(data['self'], 'request_internal')
        self = data.pop('self')
        return method(func.__name__, data)
    
    return wrapper

class HttpRequestWrap:
    def __init__(self):
        pass
    
    def request_internal(self, name, data):
        print("call request, name: {}, data: {}".format(name, data))
        
    @request
    def get_branch_info(self, project_id, branch_name):
        pass
    
    @request
    def get_branch_info2(self, project_id, branch_name=''):
        pass
h = HttpRequestWrap()
h.get_branch_info("123456", "master")
locals: {'args': (<__main__.HttpRequestWrap object at 0x10fd5b450>, '123456', 'master'), 'kw': {}, 'func': <function HttpRequestWrap.get_branch_info at 0x10fd5a4d0>}
args: (<__main__.HttpRequestWrap object at 0x10fd5b450>, '123456', 'master')
kw: {}
FullArgSpec(args=['self', 'project_id', 'branch_name'], varargs=None, varkw=None, defaults=None, kwonlyargs=[], kwonlydefaults=None, annotations={})
call request, name: get_branch_info, data: {'project_id': '123456', 'branch_name': 'master'}

h.get_branch_info2("123456")
locals: {'args': (<__main__.HttpRequestWrap object at 0x10fd5b450>, '123456'), 'kw': {}, 'func': <function HttpRequestWrap.get_branch_info2 at 0x10fd5a680>}
args: (<__main__.HttpRequestWrap object at 0x10fd5b450>, '123456')
kw: {}
FullArgSpec(args=['self', 'project_id', 'branch_name'], varargs=None, varkw=None, defaults=('',), kwonlyargs=[], kwonlydefaults=None, annotations={})
call request, name: get_branch_info2, data: {'project_id': '123456', 'branch_name': ''}

完整代码

动态生成

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-

import os
import sys
import re
import time
import requests
import json
import pprint
from requests.auth import HTTPBasicAuth

import urllib3
urllib3.disable_warnings()

from log import debug, info, warning, error, fatal

class Upsource:
    ENDPOINT = "https://upsource.xxxx.com"

    REQUEST_MAP = {
        'getProjectInfo': ['projectId'],
        'getCodeReviewPatterns': [],
        'getRevisionsList': ['projectId', 'limit', 'skip = 0', 'requestGraph = False'],
        'getRevisionsListFiltered': ['projectId', 'query', 'limit', 'skip = 0', 'requestGraph = False'],
        'getRevisionInfo': ['projectId', 'revisionId'],
        'getBranchInfo': ['projectId', 'branch'],
        'getBranchGraph': ['projectId', 'branch'],
        'getBranches': ['projectId', 'query', 'limit', 'sortBy = "updated"'],
        'findCommits': ['commits', 'requestChanges = False', 'limit = 10'],
        'getReviews': ['limit', 'query = "*"', 'sortBy = "updated"', 'projectId = "herohub-platform"', 'skip = 0'],
    }

    def __init__(self, user, passwd, retry_times = 3):
        self.__auth = HTTPBasicAuth(user, passwd)
        self.retry_times = retry_times

        # dynamic gen class function from REQUEST_MAP
        for k, v in self.REQUEST_MAP.items():
            fn_str = '''def {}(self, {}): m = locals(); m.pop('self'); return self.request(m);'''.format(k, ', '.join(v))
            info('class: {}, gen function: {}'.format(self.__class__.__name__, fn_str))
            exec(fn_str)
            setattr(Upsource, k, locals().get(k))

    def _request(self, rpc_name, data):
        suffix = '/~rpc/{}'.format(rpc_name)
        url = self.ENDPOINT + re.sub('/+', '/', suffix)

        headers = {'Content-Type': 'application/json; charset=utf-8'}
        proxies = { "http": "http://127.0.0.1:8080", "https": "http://127.0.0.1:8080"}
        retry = self.retry_times

        info('{}, request: {}'.format(rpc_name, data))

        while retry > 0:
            retry -= 1
            resp = None
            try:
                resp = requests.post(url, data=json.dumps(data), auth=self.__auth, headers=headers, verify=False)
                if resp and resp.status_code == 200:
                    return json.loads(resp.text)['result']
                raise Exception('request failed')
            except Exception as e:
                error('request upsource failed!, rpc_name: {}, request: {}, resp: {}, error:{}'.format(rpc_name, data, resp.text, str(e)))
            time.sleep(1)

    def request(self, data):
        rpc_name = sys._getframe(1).f_code.co_name
        resp = self._request(rpc_name, data)
        info(pprint.pformat(resp))
        return resp

    def getAllProjects(self, projectIdList):
        return self.request({'projectId': projectIdList})

if __name__ == '__main__':
    up = Upsource('admin', '123456')
    up.getProjectInfo("herohub-platform")
    up.getReviews(1000, 'branch:feature/20200316 and state: open')