python_libraries_demo/operation-wechat-database-demo.py at master · CavinCao/python_libraries_demo · GitHub

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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-


__author__ = 'Cavin Cao'

'''
	功能：利用云开发http api操作小程序云数据库
'''

import requests
import json
import config

APP_ID = config.WECHAT_APP_ID
APP_SECRET = config.WECHAT_APP_SECRET
ENV=config.WECHAT_ENV
TEST_COLLECTION="test_collection"

HEADER = {'content-type': 'application/json'}

WECHAT_URL="https://api.weixin.qq.com/"

'''
获取小程序token
'''
def get_access_token():
    url='{0}cgi-bin/token?grant_type=client_credential&appid={1}&secret={2}'.format(WECHAT_URL,APP_ID,APP_SECRET)
    response =requests.get(url)
    result=response.json()
    print(result)
    return result['access_token']

'''
新增集合
'''
def add_collection(accessToken):
    url='{0}tcb/databasecollectionadd?access_token={1}'.format(WECHAT_URL,accessToken)
    data={
        "env":ENV,
        "collection_name":TEST_COLLECTION
    }
    response  = requests.post(url,data=json.dumps(data),headers=HEADER)
    print('1.新增集合：'+response.text)

'''
新增数据
'''
def add_data(accessToken):
    url='{0}tcb/databaseadd?access_token={1}'.format(WECHAT_URL,accessToken)
    query='''
    db.collection("test_collection").add({
        data:{
            key:1,
            value:"2345"
        }
    })
    '''

    data={
        "env":ENV,
        "query":query
    }
    response  = requests.post(url,data=json.dumps(data),headers=HEADER)
    print('2.新增数据：'+response.text)

'''
查询数据
'''
def query_data(accessToken):
    url='{0}tcb/databasequery?access_token={1}'.format(WECHAT_URL,accessToken)
    query='''
    db.collection("test_collection").limit(10).skip(1).get()
    '''

    data={
        "env":ENV,
        "query":query
    }
    response  = requests.post(url,data=json.dumps(data),headers=HEADER)
    print('3.查询数据：'+response.text)
    result=response.json()
    resultValue =json.loads(result['data'][0])
    return resultValue['_id']

'''
删除数据
'''
def delete_data(accessToken,id):
    url='{0}tcb/databasedelete?access_token={1}'.format(WECHAT_URL,accessToken)
    query='''db.collection("test_collection").doc("{0}").remove()'''.format(id)

    data={
        "env":ENV,
        "query":query
    }
    response  = requests.post(url,data=json.dumps(data),headers=HEADER)
    print('4.删除数据：'+response.text)

'''
删除集合
'''
def delete_collection(accessToken):
    url='{0}tcb/databasecollectiondelete?access_token={1}'.format(WECHAT_URL,accessToken)
    data={
        "env":ENV,
        "collection_name":TEST_COLLECTION
    }
    response  = requests.post(url,data=json.dumps(data),headers=HEADER)
    print('5.删除集合：'+response.text)

if __name__ =='__main__':
    #0.获取token
    accessToken=get_access_token()
    #1.新增集合：
    add_collection(accessToken)
    #2.新增数据
    add_data(accessToken)
    #3.查询数据
    id=query_data(accessToken)
    #4.删除数据
    delete_data(accessToken,id)
    #5.删除集合
    delete_collection(accessToken)