peterlu14
124 lines
5.1 KiB
Python
124 lines
5.1 KiB
Python
import requests, json
|
|
from requests.adapters import HTTPAdapter
|
|
from requests.packages.urllib3.util.retry import Retry
|
|
from ._identify import get_api_jwt, set_api_jwt
|
|
|
|
api_url = '127.0.0.1'
|
|
|
|
class ApiRequests():
|
|
def __init__(self, controller_id = None):
|
|
self._controller_id = controller_id
|
|
self._header = {}
|
|
self._requests = requests.session()
|
|
|
|
# self.existDevice('aaa')
|
|
# retry_strategy = Retry(
|
|
# total=5,
|
|
# status_forcelist=[429, 500, 502, 503, 504],
|
|
# method_whitelist=["HEAD", "GET", "OPTIONS"]
|
|
# )
|
|
# adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
# self._http = requests.Session()
|
|
# self._http.mount("https://", adapter)
|
|
# self._http.mount("http://", adapter)
|
|
# self._requests.keep_alive = False
|
|
# self._requests.mount('http://', HTTPAdapter(max_retries=3))
|
|
self.setupJwt()
|
|
def setupJwt(self):
|
|
try:
|
|
if get_api_jwt() is None:
|
|
res = self._requests.get('http://' + api_url + ':3000/public/auth/get_cookie', timeout = 3)
|
|
|
|
if res.status_code == 200:
|
|
set_api_jwt(res.text)
|
|
self._header['Authorization'] = 'Bearer ' + get_api_jwt()
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def createDevice(self, name, serial_number, mac_address):
|
|
data = {
|
|
"name": str(name),
|
|
"serial_number": str(serial_number),
|
|
"mac_address": str(mac_address),
|
|
}
|
|
try:
|
|
response = self._requests.post('http://' + api_url + ':3000/api/device/create', data, headers= self._header)
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def existDevice(self, mac_address):
|
|
try:
|
|
response = self._requests.get('http://' + api_url + ':3000/api/device/get_by_mac/' + mac_address)
|
|
if len(response.json()) > 0:
|
|
return False
|
|
return True
|
|
except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as e:
|
|
pass
|
|
|
|
def updateDeviceByAttr(self, attrs, values, datas:dict):
|
|
try:
|
|
res = self._requests.post('http://' + api_url + ':3000/api/device/update_by_attr/' + str(attrs) + '/' + str(values), datas, headers= self._header, timeout = 3)
|
|
return True
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def updateDeviceByID(self, id:int, attrs:dict):
|
|
try:
|
|
res = self._requests.post('http://' + api_url + ':3000/api/device/update_by_id/' + str(id), attrs, headers= self._header, timeout = 3)
|
|
return True
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def updateDeviceAll(self, attrs):
|
|
try:
|
|
res = self._requests.post('http://' + api_url + ':3000/api/device/update/', attrs, headers= self._header, timeout = 3)
|
|
return True
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def getDeviceByAttr(self, attrs:str, values:str):
|
|
try:
|
|
res = self._requests.get('http://' + api_url + ':3000/api/device/get_by_attr/' + str(attrs) + '/' + str(values), headers= self._header, timeout = 3)
|
|
return res.json()
|
|
except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as e:
|
|
print('getDeviceByAttr', e)
|
|
pass
|
|
|
|
def getDeviceAll(self):
|
|
try:
|
|
res = self._requests.get('http://' + api_url + ':3000/api/device/get/all', headers= self._header, timeout = 3)
|
|
return res.json()
|
|
except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as e:
|
|
print('getDeviceAll', e)
|
|
pass
|
|
|
|
def createController(self, attrs):
|
|
try:
|
|
response = requests.post('http://' + api_url + ':3000/api/controller/create', attrs, headers= self._header)
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass
|
|
|
|
def getControllerAll(self):
|
|
try:
|
|
res = self._requests.get('http://' + api_url + ':3000/api/controller/get/all', headers= self._header, timeout = 3)
|
|
return res.json()
|
|
except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as e:
|
|
print('getControllerAll', e)
|
|
pass
|
|
|
|
def existController(self, mac_address):
|
|
try:
|
|
res = self._requests.get('http://' + api_url + ':3000/api/controller/get_by_mac/' + mac_address, headers= self._header, timeout = 3)
|
|
if len(res.json()) > 0:
|
|
return True
|
|
return False
|
|
except (requests.exceptions.ConnectionError, json.decoder.JSONDecodeError) as e:
|
|
print('existController', e)
|
|
pass
|
|
|
|
def updateControllerByAttr(self, attrs = None, values = None , datas:dict = None):
|
|
try:
|
|
res = self._requests.post('http://' + api_url + ':3000/api/controller/update_by_attr/' + str(attrs) + '/' + str(values), datas, headers= self._header, timeout = 3)
|
|
return True
|
|
except requests.exceptions.ConnectionError as e:
|
|
pass |