华为云Serverless版表格储存数据库

huawei_cloudtable.py

import urllib.request
import ssl
import json
import json
import base64

table_name = ''
token = ''
ep_url = 'https://hbase-cloudtable.cn-north-1.myhuaweicloud.com/v1/929213ae947c47de866553f0c025741d'

def db_del(index):
    global table_name
    global token
    global ep_url
    api_url = ep_url + '/' + table_name + '/' + index
    q = urllib.request.Request(api_url,method='DELETE')
    q.add_header('X-Auth-Token', token)
    q.add_header('Accept', 'application/json')
    context=ssl._create_unverified_context()
    a = urllib.request.urlopen(q,context=context).read()

def db_get(index):
    global table_name
    global token
    global ep_url
    api_url = ep_url + '/' + table_name + '/' + index
    q = urllib.request.Request(api_url,method='GET')
    q.add_header('X-Auth-Token', token)
    q.add_header('Accept', 'application/json')
    context=ssl._create_unverified_context()
    a = urllib.request.urlopen(q,context=context).read()
    a = a.decode('utf8').replace("'", '"')
    a = json.loads(a)
    a = a['Row'][0]['Cell']
    result = []
    for i in a:
        sub_dict = {'value' : str(base64.b64decode(i['$']),encoding="utf-8"),'key' : str(base64.b64decode(i['column']),encoding="utf-8")}
        result.append(sub_dict)
    return result;

def db_put(index,key,value):
    global table_name
    global token
    global ep_url
    api_url = ep_url + '/' + table_name + '/' + index
    q = urllib.request.Request(api_url,method='PUT')
    q.add_header('X-Auth-Token', token)
    q.add_header('Content-Type', 'application/json')
    context=ssl._create_unverified_context()
    dict_x = {'Row': [{'key': str(base64.b64encode(bytes(index,encoding = "utf8")),encoding = "utf-8"), 'Cell': \
             [{'column' : str(base64.b64encode(bytes('A:' + key,encoding = "utf8")),encoding = "utf-8"),\
             '$' : str(base64.b64encode(bytes(value,encoding = "utf8")),encoding = "utf-8")  }]}]}
    dict_x = json.dumps(dict_x)
    dict_x = bytes(dict_x,'utf8')
    a = urllib.request.urlopen(q,dict_x,context=context).read()

def db_init(username,password):
  global table_name
  global token
  global ep_url
  data = { 
    "auth": { 
      "identity": { 
        "methods": [ 
          "password" 
        ], 
        "password": { 
          "user": { 
            "name": username, 
            "password": password, 
            "domain": { 
              "name": username 
            } 
          } 
        } 
      }, 
      "scope": { 
        "project": { 
          "name": "cn-north-1"  
        } 
      } 
    } 
  }

  data = json.dumps(data)
  data = bytes(data,'utf8')
  request = urllib.request.Request('https://iam.cn-north-1.myhuaweicloud.com/v3/auth/tokens')
  result = urllib.request.urlopen(request,data).info()

  result = str(dict(result))
  result = eval(result)

  token = result['X-Subject-Token']
  
  q = urllib.request.Request(ep_url)
  q.add_header('Content-Type', 'application/json')
  q.add_header('X-Auth-Token', token)
  context=ssl._create_unverified_context()
  a = urllib.request.urlopen(q,context=context).read()
  a = a.decode('utf8').replace("'", '"')
  a = json.loads(a)
  table_name = a['table'][0]['name']

if __name__ == '__main__':
  db_init("username","password")
  db_put('xxx','ccc','abd') #index key value
  db_put('xxx','ddd','123') #index key value
  m = db_get('xxx') #index
  print('DB Return:')
  print(m)
  db_del('xxx')
  try:
   m = db_get('xxx')
  except urllib.error.HTTPError :
    print('DB Key Not Found!')
    pass