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
|