rainbond-python 0.6.0

Rainbond python cloud native development base library

Rainbond Python

基于 Rainbond 云原生平台和 Flask 框架的 Python 云原生开发基础库。

使用说明

快速开始创建一个 Python 云原生组件：

rainbond -c demo-component

设置MongoDB的环境，以Windows为例：

$ set MONGODB_HOST=127.0.0.1
$ set MONGODB_PORT=27017

error_handler

通过 @app.errorhandler(xxx) 重新定义常用的 4xx 和 5xx 状态码的异常响应。

......
from flask import Flask, request, abort
from rainbond_python.error_handler import error_handler
......
app = Flask(__name__)
error_handler(app)
......

默认情况下，Flask 会自动捕获这些异常并返回响应，但是也可以通过 abort() 方法主动返回异常响应：

abort(412)

对于 200、4xx、5xx 状态码，目前做出以下约定：

• 4xx、5xx 的错误大部分交给 error_handler 封装方法处理
• 少数业务相关的 4xx、5xx 异常通过 abort(xxx) 方法主动处理
• 为了简单，忽略除 200 以外的其他 2xx 响应，且通过 return data, 200, [] 返回，要注意的是 data 只有是必要参数，后面两个参数可以不写，200, [] 就是默认值
• 业务代码通过 return '错误信息XXX', 500 返回自定义的异常响应

Parameter

处理请求与响应参数的通用类。

from rainbond_python.parameter import Parameter

获取请求参数

通过 Parameter 类实例，可以获取以下信息：

• parameter.method: 请求类型
• parameter.headers: 请求头
• parameter.param_url: URL中传递的参数
• parameter.param_json: Json请求中的参数
• parameter.param_form: 表单请求中的参数

所有信息均为字典类型，通过 json.dumps() 可以直接作为响应返回：

@app.route('/api/1.0/demo', methods=['GET', 'POST', 'PUT', 'DELETE'])
def api_demo():
    parameter = Parameter(request)
    if parameter.method == 'GET':
        return json.dumps(parameter.param_url, ensure_ascii=False), 200, []
    elif parameter.method == 'POST':
        return json.dumps(parameter.param_json, ensure_ascii=False), 200, []
    elif parameter.method == 'PUT':
        return json.dumps(parameter.param_json, ensure_ascii=False), 200, []
    elif parameter.method == 'DELETE':
        return json.dumps(parameter.param_json, ensure_ascii=False), 200, []

校验参数内容

通过 Parameter 类的 verification() 方法，可以判断参数字典是否符合要求：

    elif parameter.method == 'POST':
        param = parameter.verification(checking=parameter.param_json, verify={'name': str, 'age': int})

其中 checking 参数是需要校验的参数字典，通常传递 parameter.param_url、parameter.param_json 或 parameter.param_form。第二个 verify 参数则是校验内容字典，需要指定 参数名 和 参数类型 作为字典项。如果请求中包含可选参数，可以将该参数的名称及其默认值输入到 optional 参数中，例如可以设置 age 参数为空时，默认填充为 18 岁：

parameter.verification(checking=parameter.param_json, verify={'name': str, 'age': int}, optional={'age': 18})

如果判断失败，则直接返回异常响应，响应体中包含明确的提示信息。默认情况下，str 类型的 必选参数不能为空字符串，如果需要为空，可以通过 null_value=True 进行设置，或者将其作为可选参数处理。

DBConnect

处理 MongoDB 读写行为的通用类。

from rainbond_python.db_connect import DBConnect
db = DBConnect(db='db_name', collection='collection_name')

分页查询

仅支持 GET 请求，使用非常简单，直接把 Parameter 类的实例传递给 DBConnect 类的 find_paging() 方法即可：

@app.route('/api/1.0/demo', methods=['GET'])
def api_demo():
    parameter = Parameter(request)
    if parameter.method == 'GET':
        find_data = db.find_paging(parameter)
        return find_data, 200, []

内部组件或外部客户端通过 /api/1.0/demo?page_size=10&current=1&columns=["title"]&sort_order=[""]&filtered_value=["标题"] 即可访问，请求参数如下：

• page_size: 每页条数，从1开始计算
• current: 当前页数，从1开始计算
• columns: 受控列
• sort_order: 排序顺序（对应受控列），必须与受控列长度一致，""=不排序、asc=升序、desc=降序
• filtered_value: 筛选值（对应受控列），必须与受控列长度一致
• start_date: 可选，开始日期（区间查询），支持日期（2020-10-1）格式和时间戳（601481600）格式
• end_date: 可选，结束日期（区间查询），同上，必须成对出现

由于技术原因，目前不支持 int 型数据的模糊查询。

写文档

insert_dict = {'name': 'Xiao Ming', 'age': 23}
db.write_one_docu(docu=insert_dict)

如果写入失败，会直接返回异常响应，如果成功则会返回新数据的 _id 值。

文档是否存在

examine_dict = {'name': 'Xiao Ming'}
if db.does_it_exist(docu=examine_dict):
    print('Docu already exists')
else:
    print('Docu does not exist')

更新文档

同样的，如果更新失败，也会直接返回异常响应。

更新单个匹配文档

find_dict = {'name': 'Xiao Ming'}
modify_dict = {'$set': {'name': 'Xiao Hong'}}
db.update_docu(find_docu=find_dict, modify_docu=modify_dict)

更新全部匹配文档

find_dict = {'age': 23}
modify_dict = {'$set': {'name': '23 year old'}}
db.update_docu(find_docu=find_dict, modify_docu=modify_dict, many=True)

该方法会返回一个包含 matched_count 和 modified_count 即匹配/影响数据条数的字典。

查询文档

通过 find_docu() 标准查询方法时，无论查询单个还是多个，返回均是 list 类型数据。

查询单个匹配文档

find_dict = {'title': {'$regex': '标题'}}
find_data_list = db.find_docu(find_dict=find_dict, many=False)
print(find_data_list[0])

查询全部匹配文档

find_dict = {'title': {'$regex': '标题'}}
find_data_list = db.find_docu(find_dict=find_dict)
for find_data in find_data_list:
    print(find_data)

根据id查找文档

from rainbond_python.db_connect import DBConnect
db = DBConnect('unitest_rainbond_python', 'test_db_connect')
id = db.write_one_docu({'name': 'LaoXu'})
docu = db.find_docu_by_id(str(id))

# 当id不存在时，默认会使用abort抛出异常
fail_docu = db.find_docu_by_id('6008daa19223551b00548ded')
# 可以将raise_err=False时，id不存在会返回None
fail_docu = db.find_docu_by_id('6008daa19223551b00548ded',raise_err=False)

该方法返回记录字典，且把'_id'转换为了str类型

根据id列表查找文档

from rainbond_python.db_connect import DBConnect
db = DBConnect('unitest_rainbond_python', 'test_db_connect')
docu_list = db.find_docu_by_id_list(['6008daa19223551b00548ded','6008daa29223551b00548dee'])

该方法返回记录字典列表，且把'_id'转换为了str类型。当所有id不存在时，返回[]

通用方法

handle_date()

将 2020-10-1 或 601481600 即日期格式或时间戳格式的字符串，处理成 Python 的 datetime.datetime 数据：

from rainbond_python.tools import handle_date
print(handle_date(date='2020-10-1'))
print(handle_date(date='2020-10-31', date_type='end'))

通过 date_type 可以设置是日期的开始（start）还是一天的结束（end）时间。

handle_db_dict()

将 MongoDB 字典数据中的 _id 转换为 str 类型、时间转换成时间戳：

query_dict = self.mongo_collection.find_one({'title': {'$regex': '标题1'}})
handle_db_dict(query_dict)

handle_db_to_list()

将 MongoDB 的列表中的 _id 转换为 str 类型，并转换为字典列表（原db的id是ObjectId类型，转为json会报错）：

from rainbond_python.tools import handle_db_to_list
from rainbond_python.db_connect import DBConnect

def test_handle_db_to_list():
    db = DBConnect('unitest_rainbond_python', 'test_parameter')
    old_list = db.mongo_collection.find({})
    new_list = handle_db_to_list(old_list)
    print('new_list is a list of dict',new_list)

开发与测试

调试开发

基础调试代码的 demo.py 即 rainbond -c demo-component 命令创建项目中的 app.py 文件，是一个可以快速开始的基础代码项目。

在本地调试时，在 devs 目录下创建 dev_xxxxx.py，并复制 demo.py 文件里的代码，并在里面调试 rainbond_python 目录下的代码。（本地创建的 dev_*.py 文件会被忽略，不会被提交），同时要在开头处添加下面代码，以调用基础包中的代码：

......
import sys
sys.path.append('..')
from rainbond_python.parameter import Parameter
from rainbond_python.error_handler import error_handler
from rainbond_python.db_connect import DBConnect
......

单元测试

单元测试在 /tests/* 目录下

• 执行单元测试

$ pytest

参考

• Restful API : 具体的组件API开发标准
• 12 Factor : 符合十二要素的才是云原生应用
• RainBond : 一个开源的云原生平台