Rainbond python cloud native development base library
Project description
Rainbond Python
基于 Rainbond 云原生平台和 Flask 框架的 Python 云原生开发基础库。
使用说明
快速开始创建一个 Python 云原生组件:
rainbond -c demo-component
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)
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, [('Content-Type', 'application/json; charset=utf-8')
内部组件或外部客户端通过 /api/1.0/demo?page_size=10¤t=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)
如果写入失败,会直接返回异常响应。
文档是否存在
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)
通用方法
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
)时间。
参考
- Restful API : 具体的组件API开发标准
- 12 Factor : 符合十二要素的才是云原生应用
- RainBond : 一个开源的云原生平台
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.