docs: 同步日常维护指南并补齐遗漏的板卡配置
This commit is contained in:
91
mixly/boards/default/micropython/build/lib/tiny_webdb.py
Normal file
91
mixly/boards/default/micropython/build/lib/tiny_webdb.py
Normal file
@@ -0,0 +1,91 @@
|
||||
import urequests as requests
|
||||
|
||||
|
||||
def url_quote(s):
|
||||
safe = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_.-"
|
||||
s = str(s)
|
||||
res = bytearray()
|
||||
for b in s.encode('utf-8'):
|
||||
if b in safe:
|
||||
res.append(b)
|
||||
else:
|
||||
res.extend(b'%' + b'%02X' % b)
|
||||
return res.decode()
|
||||
|
||||
|
||||
class TinyWebDB:
|
||||
def __init__(self, *args):
|
||||
if len(args) == 1:
|
||||
self.init_with_mqtt(*args)
|
||||
else:
|
||||
self.init_with_user(*args)
|
||||
|
||||
def init_with_user(self, url, username, password):
|
||||
self._api_url = ""
|
||||
self._username = username
|
||||
self._password = password
|
||||
self.set_url(url)
|
||||
|
||||
def init_with_mqtt(self, mqtt_client):
|
||||
self._api_url = ""
|
||||
url, username, password = mqtt_client.get_server_info()
|
||||
self.set_url('https://{}/tinydb'.format(url))
|
||||
self._username = username
|
||||
self._password = password
|
||||
|
||||
def update(self, key, value):
|
||||
key = url_quote(str(key))
|
||||
value = url_quote(str(value))
|
||||
result = self._request("update", "tag={}&value={}".format(key, value))
|
||||
if "status" in result and result["status"] == "error":
|
||||
raise RuntimeError(result["message"])
|
||||
|
||||
def get(self, key):
|
||||
key = url_quote(str(key))
|
||||
result = self._request("get", "tag={}".format(key))
|
||||
if "status" in result and result["status"] == "error":
|
||||
raise RuntimeError(result["message"])
|
||||
return result["value"]
|
||||
|
||||
def count(self):
|
||||
result = self._request("count")
|
||||
if "status" in result and result["status"] == "error":
|
||||
raise RuntimeError(result["message"])
|
||||
return int(result["count"])
|
||||
|
||||
def search(self, no=1, count=1, tag='', dtype='both'):
|
||||
no = str(no)
|
||||
count = str(count)
|
||||
tag = url_quote(tag)
|
||||
result = self._request("search", "no={}&count={}&tag={}&type={}".format(no, count, tag, dtype))
|
||||
if "status" in result and result["status"] == "error":
|
||||
raise RuntimeError(result["message"])
|
||||
return result["data"]
|
||||
|
||||
def delete(self, key):
|
||||
key = url_quote(str(key))
|
||||
result = self._request("delete", "tag={}".format(key))
|
||||
if "status" in result and result["status"] == "error":
|
||||
raise RuntimeError(result["message"])
|
||||
|
||||
def set_url(self, url):
|
||||
if url[-1] != '/':
|
||||
url += '/'
|
||||
self._api_url = url
|
||||
|
||||
def _request(self, op, param=""):
|
||||
data = "user={}&secret={}&action={}".format(self._username, self._password, op)
|
||||
if param:
|
||||
data += '&' + param
|
||||
try:
|
||||
headers = {
|
||||
"Content-Type": "application/x-www-form-urlencoded"
|
||||
}
|
||||
response = requests.post(self._api_url, data=data, headers=headers)
|
||||
result = {}
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
response.close()
|
||||
return result
|
||||
except Exception as e:
|
||||
raise RuntimeError("API request failed or WiFi is not connected", e)
|
||||
Reference in New Issue
Block a user