专栏首页小程序-云开发微信小程序-云开发 数据库http接口封装
原创

微信小程序-云开发 数据库http接口封装

最近学习一下微信小程序的云开发,作为serverless架构,着实省去很多麻烦,省的搞https证书了,也不用写api,但是总是要建后台系统的,云开发的数据库是nosql,为了用一下分页,我就封装了一下接口,如下

import requests
import json
from flask_sqlalchemy import Pagination

class WeChatCloudDB(object):
    """
    微信小程序云开发相关接口
    https://developers.weixin.qq.com/miniprogram/dev/wxcloud/reference-http-api/
    """
    def __init__(self, env, access_token):
        self.env = env
        self.access_token = "access_token=" + access_token
        self.query = ''
        self.url = ''
        self.method = ''

    def paginate(self, page=None, per_page=None, max_per_page=100):
        if page is None:
            page = 1

        if per_page is None:
            per_page = 20

        if max_per_page is not None:
            per_page = min(per_page, max_per_page)

        if page < 1:
            page = 1

        if per_page < 0:
            per_page = 20

        self.query += '.limit({0}).skip({1})'.format(per_page, (page-1)*per_page)
        items = self.commit()
        return Pagination(self, page, per_page, self.total, items)

    def commit(self):
        url = self.url + '?' + self.access_token
        data = {'env': self.env, 'query': self.query+'.get()' if self.method=='get' else self.query}
        response = requests.post(url, data=json.dumps(data))
        results = json.loads(response.text)
        if results and results.get('errcode')==0:
            pager = results.get('pager')
            if pager:
                self.total = pager.get('Total')
            if results.get('data'):
                return results.get('data')
            if results.get('modified'):
                return results.get('modified')
            if results.get('deleted'):
                return results.get('deleted')
            if results.get('count'):
                return results.get('count')
            if results.get('id_list'):
                return results.get('id_list')
            
        else:
            print(results.get('errcode'), results.get('errmsg'))
            return []
        
    def get(self, table, _id, **kwargs):
        self.method = 'get'
        self.url = 'https://api.weixin.qq.com/tcb/databasequery'
        self.query = 'db.collection("{}")'.format(table)
        if _id:
            self.query += '.doc("{}")'.format(_id)
        elif kwargs:
            filter_str = ''
            for k,v in kwargs.items():
                if isinstance(v, str):
                    filter_str += '{0}:"{1}"'.format(k, v)
                elif isinstance(v, bool):
                    filter_str += '{0}:{1}'.format(k, v and 'true' or 'false')
                else:
                    filter_str += '{0}:{1}'.format(k, v)

            self.query += '.where({%s})' % filter_str
        return self
   
    def update_one(self, table, _id, **kwargs):
        self.method = 'update'
        self.url = 'https://api.weixin.qq.com/tcb/databaseupdate'
        self.query = 'db.collection("{0}").doc("{1}")'.format(table, _id)
        if kwargs:
            filter_list = []
            for k,v in kwargs.items():
                if isinstance(v, str):
                    filter_list.append('{0}:"{1}"'.format(k, v))
                elif isinstance(v, bool):
                    filter_list.append('{0}:{1}'.format(k, v and 'true' or 'false'))
                else:
                    filter_list.append('{0}:{1}'.format(k, v))
            filter_str = ','.join(filter_list)
            self.query += '.update({data:{%s}})' % filter_str
        return self
    
    def delete_one(self, table, _id):
        self.method = 'delete'
        self.url = 'https://api.weixin.qq.com/tcb/databasedelete'
        self.query = 'db.collection("{0}").doc("{1}").remove()'.format(table, _id)
        return self

    def add_one(self, table, **kwargs):
        self.method = 'add'
        self.url = 'https://api.weixin.qq.com/tcb/databaseadd'
        self.query = 'db.collection("{}")'.format(table)
        if kwargs:
            filter_list = []
            for k,v in kwargs.items():
                if isinstance(v, str):
                    filter_list.append('{0}:"{1}"'.format(k, v))
                elif isinstance(v, bool):
                    filter_list.append('{0}:{1}'.format(k, v and 'true' or 'false'))
                else:
                    filter_list.append('{0}:{1}'.format(k, v))
            filter_str = ','.join(filter_list)
            self.query += '.add({data:[{%s}]})' % filter_str
        return self 
    
    def count(self, table, **kwargs):
        self.method = 'count'
        self.url = 'https://api.weixin.qq.com/tcb/databasecount'
        self.query = 'db.collection("{}")'.format(table)
        if kwargs:
            filter_str = ''
            for k,v in kwargs.items():
                if isinstance(v, str):
                    filter_str += '{0}:"{1}"'.format(k, v)
                elif isinstance(v, bool):
                    filter_str += '{0}:{1}'.format(k, v and 'true' or 'false')
                else:
                    filter_str += '{0}:{1}'.format(k, v)

            self.query += '.where({%s})' % filter_str
        self.query += '.count()' 
        return self

link:https://github.com/xiaoyeqiannian/alita/blob/master/util/wx_api.py

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • python实现坦克大战

    砸漏
  • PaddleX助力无人驾驶:基于YOLOv3的车辆检测和车道线分割实战

    无人驾驶汽车利用传感器技术、信号处理技术、通讯技术和计算机技术等,通过集成视觉、激光雷达、超声传感器、微波雷达、GPS、里程计、磁罗盘等多种车载传感器来辨识汽车...

    用户1386409
  • 【iOS】教你用ZFPlayer+KTVHTTPCache搭建缓存,预加载的播放器

    mgr实现ZFPlayerMediaPlayback协议,然后在初始化时,开启本地服务器

    MapleYe
  • Python+PyQt5实现灭霸响指功能

    之前给大家分享过灭霸命令GitHub地址:https://github.com/hotvulcan/Thanos.sh,本文使用Python+PyQt5实现灭霸...

    砸漏
  • 使用Python生成基础验证码教程

    pillow是Python平台事实上的图像处理标准库。PIL功能非常强大,但API却非常简单易用。 所以我们使用它在环境里做图像的处理。

    步履不停凡
  • Pyspider框架 —— Python爬虫实战之爬取 V2EX 网站帖子

    这篇文章本是我暑假时写的,可是自己懒啊,最近自己又在捣鼓 python 了,然后蹭有机会然后就把这篇文章写下来了,后期应该还有爬取知乎爬虫文章,期待吧,写原创文...

    zhisheng
  • iOS原生的搜索:UISearchController

    王大锤
  • python+pygame游戏开发之使用Py2exe打包游戏

    最近在用python+pygame 开发游戏,写完以后在分享给朋友玩的时候遇到了很大的问题,只有搭建了环境才能运行python脚本。

    马三小伙儿
  • 超级简单的虚拟机(Python 实现)

    我们这次实现的简单虚拟机,和计算机的 cpu 有点类似。无非就是取指令,执行指令之类的操作。

    用户2870857
  • python的tkinter编程(九)Text多行文本框的详细解读

    一天不写程序难受

扫码关注云+社区

领取腾讯云代金券