基于Python编写一个调用API的类

现在后端开发基本上都是写各种API提供给别人使用,我在日常工作里既写API,也经常调用别人写的API。

分享一下经常使用的调用API的模块。

看代码之前会有一些假设,可以帮助理解代码。

一些假设

假设我们有一个API是: http://127.0.0.1:8000/api/token , 这个详细信息可以参考simple jwt

我在这里给一个简单的接口文档,如下。

请求方法

POST

请求参数

在请求体中需要提供以下json格式的数据:

  • username: 用户名
  • password: 密码

示例:

1
{     "username": "<username>",     "password": "<password>" }

响应内容

如果认证成功,接口将返回如下格式的响应:

1
{     "access": "<Access_Token>",     "refresh": "<Refresh_Token>" }

其中:

  • <Access_Token>: 访问令牌,可以用来进行后续的受保护操作。
  • <Refresh_Token>: 刷新令牌,可以用来在访问令牌过期后获取新的访问令牌。

错误处理

  • 如果用户名或密码错误,会返回401 Unauthorized错误,并且具有描述性的错误信息。
  • 如果服务器内部发生错误,会返回500 Internal Server Error错误。

使用示例

请求

1
curl --header "Content-Type: application/json" \   --request POST \   --data '{"username":"admin", "password":"123456"}' \   http://localhost:8000/token/

响应

1
{     "access": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6Ikpv...",     "refresh": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6Ikpv..." }

APIConnection

有了上面的假设,现在就可以来看代码了。

代码如下,直接贴出了全部代码,需要解释的内容放在了注释中。

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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
import json
import logging
import os
import requests


logging.basicConfig(
format="%(asctime)s %(levelname)-8s %(message)s",
level=logging.INFO,
datefmt="%Y-%m-%d %H:%M:%S",
)


class APIConnection:
"""
Api Connection
"""

def __init__(self):
# 通过环境变量来获取后端的host,而不是硬编码
self.api_url = os.environ.get("BACKEND_API_URL", "http://localhost:8000/")

self.token = ""
self.headers = {
"Content-Type": "application/json",
"Authorization": "Bearer {}".format(self.token),
"Cache-Control": "no-cache",
}

def request_jwt(self):
"""
用于调用/api/token获取token,
调用的时候需要先从环境变量中获取用户名和密码,
获得token之后更新self.headers属性,便于后面发起请求的时候做认证
"""
self.headers = {
"Content-Type": "application/json",
"Cache-Control": "no-cache",
}

api_url = f"{self.api_url}api/token/"

data = {
"username": os.environ["SCRIPT_USER"],
"password": os.environ["SCRIPT_PASSWORD"],
}

res = requests.post(
api_url, data=json.dumps(data), headers=self.headers, timeout=60
)

if res.status_code == 200:
data = json.loads(res.text)

self.token = data["access"]

self.headers = {
"Content-Type": "application/json",
"Authorization": "Bearer {}".format(self.token),
"Cache-Control": "no-cache",
}

return self.token

logging.info("Invalid.", res.text)
return False

def get_data(self, path, limit=100):
"""
从一个指定的uri(也就是path参数)获取数据返回给前端,
默认是100条数据,
使用于不做任何过滤的情况
"""

self.request_jwt()

request_url = f"{self.api_url}{path}?limit={limit}"
response = requests.get(request_url, headers=self.headers, timeout=60)

if response.status_code == 200:
data = response.json()

return data

logging.error(
"statue code: {}. error message: {}".format(
response.status_code, response.text
)
)

return None

def post_data(self, item, api):
"""
新建一条数据
"""
self.request_jwt()
api_url = f"{self.api_url}{api}"

try:
response = requests.post(
api_url, data=json.dumps(item), headers=self.headers, timeout=60
)

if response.status_code == 403:
if self.request_jwt():
self.post_data_with_response(item, api)
else:
logging.error("Invalid Credentials")

elif response.status_code not in [200, 201]:
logging.error(response.text)
logging.error(f"{response.status_code} - unable to post data")

except Exception as err:
logging.error(err)

def update_data(self, api, item):
"""
更新数据,
区别于新建数据,是对已有的数据进行更新,要注意传入主键
"""

api_url = f"{self.api_url}{api}"
self.request_jwt()
try:
response = requests.patch(
url=api_url, data=json.dumps(item), headers=self.headers, timeout=60
)

if response.status_code == 403:
# try again
self.update_data(api, item)

elif response.status_code not in [200, 201]:
logging.info(response.status_code)

except Exception as err:
logging.error(err)
return err

return response

写这篇博客的时候发现这段代码写的真心不咋地,有以下几个问题

  1. 异常处理只有日志
  2. 每个方法都单独请求一次self.request_jwt(),既对后端造成没必要的压力,有增加了自身的耗时
  3. 可以使用request.Session()来保持一些header参数,并且利用连接池,可以提高性能

这大概就是写博客的意义所在吧,写作的时候其实就在是做复盘。