apppublic/appPublic/oauth_client.py

68 lines
1.9 KiB
Python
Raw Normal View History

2024-02-29 14:15:25 +08:00
import json
from appPublic.httpclient import HttpClient, RESPONSE_TEXT, RESPONSE_JSON, RESPONSE_BIN,RESPONSE_FILE
from appPublic.argsConvert import ArgsConvert
class OAuthClient:
def __init__(self, desc):
assert desc.get('data')
assert desc.get('mapis')
self.desc = desc
self.data = desc.get('data')
2024-02-29 18:49:03 +08:00
self.mapis = desc.get('mapis')
2024-02-29 14:15:25 +08:00
self.mkeys = self.mapis.keys()
self.ac = ArgsConvert('${', '}')
2024-02-29 18:49:03 +08:00
async def __call__(self, mapi, params):
2024-02-29 14:15:25 +08:00
if mapi not in self.mkeys:
raise Exception(f'{mapi} not defined')
d = self.mapis[mapi]
assert d.get('url')
method = d.get('method', 'GET')
headers = {}
for h in d.get('headers',{}):
headers[h.get('name')] = h.get('value')
myparams = {}
for p in d.get('params', {}):
2024-02-29 18:49:03 +08:00
myparams[p.get('name')] = p.get('value')
2024-02-29 14:15:25 +08:00
url = d.get('url')
_params = self.datalize(myparams, params)
_headers = self.datalize(headers, params)
2024-02-29 18:49:03 +08:00
response_type = d.get('response_type', RESPONSE_JSON)
2024-02-29 14:15:25 +08:00
hc = HttpClient()
2024-02-29 18:49:03 +08:00
resp_data = await hc.request(url, method, response_type=response_type,
params=_params,
headers=_headers)
if resp_data is None:
raise Exception(f'request error:{url=}\n{method=}\n{response_type=}\n{_params=}\n{_headers=}')
errfield = d.get('error_field')
err = resp_data.get(errfield)
if err:
msg = resp_data.get(d.get('error_msg_field'))
raise Exception(f'Error:{url}\n{msg}')
respdesc= d.get('resp_set_data', [])
2024-02-29 14:15:25 +08:00
for r in respdesc:
2024-02-29 18:49:03 +08:00
v = resp_data[r['field']]
k = r['name']
self.data[k] = v
2024-02-29 14:15:25 +08:00
return resp_data
def datalize(self, dic, data={}):
mydata = self.data.copy()
mydata.update(data)
s1 = self.ac.convert(dic, mydata)
return s1
def is_error(self, resp_data, error_field):
if resp_data.get(error_feield):
return True
def set_data(self, resp_data, data_desc):
for dd in data_desc:
f = dd['field']
n = dd['name']
if resp_data.get(f):
self.data[n] = resp_data[f]