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):
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:
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
|