Python implementation of the arkfbp
Project description
arkfbp-py
arkfbp-py is the python implementation of the arkfbp.
installation
arkfbp-py需要 Python 3.6+ 及Django 2.0+ 的版本支持。
pip3 install arkfbp (暂不可用)
or
pip3 install git+https://github.com/longguikeji/arkfbp-py.git@zzr/basic
Dev installation
python3 setup.py install
Quick Start
1、新建名为demo的项目:
arkfbp-py startproject demo
2、在项目根目录下，新建名为app1的应用:
arkfbp-py startapp app1
3、移动到demo/app1/flows目录下，新建名为flow1的流，并设置类型 --class:
arkfbp-py createflow flow1 --class view
4、移动到demo/app1/flows/flow1/nodes目录下，新建名为node1的节点,并设置类型 --class和标识 --id:
arkfbp-py createnode node1 --class function --id node1
5、在Node1的run方法示例如下:
def run(self, *args, **kwargs):
print(f'Hello, Node1!')
return 'hello arkfbp'
6、demo/app1/flows/flow1的main.py示例如下:
from arkfbp.node import StartNode, StopNode
from arkfbp.graph import Graph
# Editor your flow here.
from arkfbp.flow import ViewFlow
from app1.flows.flow1.nodes.node1 import Node1
class Main(ViewFlow):
def create_nodes(self):
return [
{
'cls': StartNode,
'id': 'start',
'next': 'node1'
},
{
'cls': Node1,
'id': 'node1',
'next': 'stop'
},
{
'cls': StopNode,
'id': 'stop'
}
]
7、在demo/arkfbp/routes/demo.json中配置路由信息:
{
"namespace": "demo/v1/",
"routes": [
{
"flow1/": {
"get": "app1.flows.flow1"
}
}
]
}
8、迁移路由信息，其中参数--topdir可指定路由配置信息所在目录，参数--urlfile可指定迁移后的文件所在路径，默认会在项目settings.py文件所在路径查找并生成文件:
python3 manage.py migrateroute --topdir demo --urlfile demo/demo_urls.py
9、将8中生成的url文件，配置到项目的demo/urls.py中。
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path('admin/', admin.site.urls),
path('', include('demo.demo_urls'))
]
10、尝试运行流flow1:
python3 manage.py runflow --flow app1.flows.flow1.main --input {\"username\": \"admin\"} --http_method post --header {\"Authorization\": \"token\"}
11、使用django原生方式启动server。
python3 manage.py runserver 0.0.0.0:8000
Advanced usage
GlobalHookFlow（已废弃）
全局钩子式工作流运行的场景适用于：
1）服务进行路由之前（self.before_route）
2）所有工作流运行之前（self.before_flow）
3）所有工作流运行之后（self.after_flow）
4）抛出异常之前（self.before_exception）
简单使用
1、创建全局钩子式工作流，在项目根目录创建hook.py文件(仅为示例)
from arkfbp.flow import GlobalHookFlow
class HookFlow(GlobalHookFlow):
def create_nodes(self):
return [
{
'cls': StartNode,
'id': 'start',
'next': 'stop'
},
{
'cls': StopNode,
'id': 'stop'
}
]
def set_mount(self):
self.before_flow = True
2、在set_mount()方法中设置想要开启钩子的位置。
def set_mount(self):
"""
设置为在所有工作流运行之前执行全局钩子流
"""
self.before_flow = True
3、将钩子流配置到项目的settings.py文件的MIDDLEWARE变量中。
INSTALLED_APPS = [
...
]
MIDDLEWARE = [
...
'hook.HookFlow',
'hook.HookFlow1',
'hook.HookFlow2',
]
HookFlow的执行顺序
GlobalHookFlow的执行顺序与django原生Middleware执行顺序一致，
before_route()、before_flow()的执行顺序依次为从上至下；after_flow()、before_exception()则为从下至上。
New GlobalHookFlow
全新的钩子流现已可以使用。
简单使用
1、在demo/hook/文件夹下创建一个全局钩子流，并设置类型 --class。
arkfbp-py createflow hook1 --class view
2、创建节点Node1（过程略），并编辑。
class Node1(FunctionNode):
id = 'node1'
def run(self, *args, **kwargs):
print(f'Hello, Hook!')
return None
3、在demo/arkfbp/hooks/hook.json中设置流的执行位置。
{
"before_route": ["hook.hook1"],
"before_flow": [],
"before_exception": [],
"before_response": []
}
4、这样在每次路由之前，都会先进入hook1这个流进行处理。
详解
全局钩子式工作流运行的场景适用于：
1）接口路由之前（before_route）
2）工作流运行之前（before_flow）
3）返回响应之前（before_response）
4）抛出异常之前（before_exception）
列表中流的摆放顺序，即为执行顺序。
Flow Hook
1、流创建成功后
def created(inputs, *args, **kwargs):
pass
2、流初始化之前
def before_initialize(inputs, *args, **kwargs):
pass
3、流初始化之后
def initialized(inputs, *args, **kwargs):
pass
4、流执行之前
def before_execute(inputs, *args, **kwargs):
pass
5、流执行之后
def executed(inputs, ret, *args, **kwargs):
pass
6、流被销毁之前
def before_destroy(inputs, ret, *args, **kwargs):
pass
ShutDown Flow
Flow Shutdown
现在，你可以通过flow.shutdown(outputs, **kwargs)方法，来随时随地的停止工作流的运行
如果你使用ViewFlow来定义流，那么可指定返回的response的状态码response_status，例如：
class Main(ViewFlow):
def create_nodes(self):
return [
{
'cls': StartNode,
'id': 'start',
'next': 'node1'
},
{
'cls': Node1,
'id': 'node1',
'next': 'stop'
},
{
'cls': StopNode,
'id': 'stop'
}
]
def before_initialize(inputs, *args, **kwargs):
self.shutdown('Flow Error！', response_status=400)
Node Shutdown
同样，你也可以通过node.flow.shutdown(outputs, **kwargs)方法，来随时随地的停止工作流的运行。
如果你使用ViewFlow来定义流，那么可指定返回的response的状态码response_status，例如：
class Node1(FunctionNode):
id = 'node1'
def run(self, *args, **kwargs):
print(f'Hello, Hook 1!')
self.flow.shutdown('Flow Error！', response_status=400)
Flow State
Flow Steps
flow.steps为一个dict，其中包含以node_id为key、以node_instance为value的数据。
现在你可以在任何一个节点，从node.state.steps中，获取指定的已运行的node。
node1 = node.state.steps.get('node1', None)
ViewFlow inputs
ViewFlow的inputs为原生的django的WSGIRequest对象，ViewFlow在此基础上为inputs对象增加了data、extra_data、str属性。
DataSet
ds属性将原生WSGIRequest对象的GET和POST的数据合并为一个dict。
extra_ds
你可以在extra_ds中存放你想要传递下去的任何数据。
str
str包含了请求体中的字符串信息。
注意：你可以随意为inputs增加任何属性，例如：
inputs.attr = {}
这样你就为inputs增加了attr的属性
Feature For CLI
Create Flow
现在你可以通过指定目录和基类来创建一个工作流，--topdir参数代表创建流的所在目录，--class参数代表工作流期望继承的基类流。
python3 manage.py createflow flow1 --topdir demo/flows --class base
或者
arkfbp-py createflow flow1 --topdir demo/flows --class base
详解：--class 参数可选值如下
{
'base': 'Flow',
'view': 'ViewFlow',
'hook': 'GlobalHookFlow',
}
也可通过命令行获取相关信息
arkfbp-py createflow -h
Create Node
现在你可以通过指定目录和基类来创建一个流节点，--topdir参数代表创建节点的所在目录，--class参数代表节点期望继承的基类节点, --id参数代表节点在流中的唯一标识。
python3 manage.py createnode node1 --topdir demo/flows/flow1/nodes --class base --id node1
或者
arkfbp-py createnode node1 --topdir demo/flows/flow1/nodes --class base --id node1
详解：--class 参数可选值如下
{
'base': 'Node',
'start': 'StartNode',
'stop': 'StopNode',
'function': 'FunctionNode',
'if': 'IFNode',
'loop': 'LoopNode',
'nop': 'NopNode',
'api': 'APINode',
'test': 'TestNode',
'trigger_flow': 'TriggerFlowNode',
}
也可通过命令行获取相关信息
arkfbp-py createnode -h
TestFlow
Create Flow
1、 通过Quick Start中的第3步新建一个工作流，新建的工作流的名称必须以test开头。
2、 将该工作流main.py模块里Main函数的父类ViewFlow修改为Flow。
3、 将from arkfbp.flow import ViewFlow修改为from arkfbp.flow import Flow。
这样就得到一个测试流
测试流的main.py如下：
from arkfbp.flow import Flow
from arkfbp.node import StartNode, StopNode
from app1.flows.testt1.nodes.node1 import Node1
# Editor your flow here.
class Main(Flow):
def create_nodes(self):
return [
{
'cls': StartNode,
'id': 'start',
'next': 'node1'
},{
'cls': Node1,
'id': 'node1',
'next': 'stop'
},{
'cls': StopNode,
'id': 'stop'
}
]
Create node
1、 通过Quick Start中的第4步新建一个节点。
2、 将新建节点对应python文件里节点类的父类FunctionNode改为TestNode。
3、 新建节点对应python文件里from arkfbp.node import FunctionNode修改为from arkfbp.node import TestNode。
这样就得到一个测试节点
测试节点node1如下：
from arkfbp.node import TestNode
# Editor your node here.
class Node1(TestNode):
def run(self, *args, **kwargs):
print(f'Hello, Node1!')
测试节点使用
1、 setUp函数
测试节点的setUp函数将在测试用例执行之前调用，可用于准备数据等。
def setUp(self):
print('before start test')
2、 tearDown函数
测试节点的tearDown函数在测试用例全部执行之后调用。
def tearDown(self):
print('after finish test')
3、 测试用例
测试用例为以test_开头的函数。
def test_one(self):
pass
4、 断言
测试节点支持python自带断言和django unittest的断言方法。
def test_one(self):
assert 1==1
def test_two(self):
self.assertEqual(1,1)
5、 调用其他测试流
在一个测试用例中可以调用其他测试流，得到被调用测试流的结果。调用方式如下：
from arkfbp.node import TestNode
from app1.flows.testt1.main import Main
class Node1(TestNode):
def test_other_testflow(self):
self.get_outputs(Main(),inputs={},http_method='get')
首先需要先从被调用测试流的main模块中引入Main类，然后调用函数get_outputs。
函数get_outputs有三个参数，第一个参数为被调用测试流Main类的实例，即Main()；第二个参数为输入的数据，字典类型；第三个参数为调用测试流的方法，为get
Run Flow
运行指定目录下测试流
1、 在项目目录下新建python 文件
2、 引入executer模块
3、 调用函数start_testflows运行测试流
函数start_testflows有一个参数，表示指定的目录，传入相对路径、绝对路径均可。运行指定工作流如下：
from arkfbp import executer
print(executer.FlowExecuter.start_testflows('./app1/flows/'))
若想运行全部测试流也可通过命令实现。在manage.py文件所在目录下输入命令python3 manage.py flowtest，即可直接运行所有测试流
Extension CLI
此部分内容适用于可视化插件开发相关人员
AddNode
在流的图定义（create_nodes）中同步一个已知的节点信息。
python3 manage.py ext_addnode --flow <flow_name> --class <node_class> --id <node_id> --next <next_node_id> --alias <node_alias> --x <coord_x> --y <coord_y>
示例
python3 manage.py ext_addnode --flow app1.flows.flow1 --class app1.flows.flow1.nodes.node1.Node1 --id node1 --next node2 --alias Flow1_Node1 --x 123.123456 --y 123.123456
如果使用arkfbp-py命令，需指定--topdir参数，其代表项目的绝对根路径：
arkfbp-py ext_addnode --flow app1.flows.flow1 --class app1.flows.flow1.nodes.node1.Node1 --id node1 --next node2 --alias Flow1_Node1 --x 123.123456 --y 123.123456 --topdir /Users/user/Development/demo
详解
参数flow代表流的路径以.分隔，具体到流的文件夹名称；参数id代表节点的唯一标识；参数class代表相关节点的路径以.分隔，具体到类名；参数next代表后继节点的id；参数alias代表在import时，指定的节点类的别名；参数x和y分别代表插件中的x、y坐标。
参数id、flow和class是必选，其他可选，不选则默认参数为None，你也可通过命令行获取相关信息：
arkfbp-py ext_addnode -h
UpdateNode
在流的图定义（create_nodes）中修改一个已知的节点信息。
python3 manage.py ext_updatenode --flow <flow_name> --class <node_class> --id <node_id> --next <next_node_id> --alias <node_alias> --x <coord_x> --y <coord_y>
如果使用arkfbp-py命令，需指定--topdir参数，其代表项目的绝对根路径：
arkfbp-py ext_updatenode --flow app1.flows.flow1 --class app1.flows.flow1.nodes.node2.Node2 --id node1 --next node3 --alias Flow1_Node2 --x 123.123456 --y 123.123456 --topdir /Users/user/Development/demo
详解
参数flow代表流的路径以.分隔，具体到流的文件夹名称；参数id代表目标节点的唯一标识，用于指定修改的目标节点；参数class代表节点类型，其路径以.分隔并具体到类名，用于修改目标节点的类型；参数next代表后继节点的id，用于修改目标节点的后继节点；参数alias代表在import时，指定的节点类的别名，用于修改目标节点的类型别名；参数x和y分别代表插件中的x、y坐标，用于修改目标节点在插件中的坐标。
当你想要将next设置为None的时候，可以在传递参数时指定--next为undefined即可。
参数id、flow是必选，其他可选，不选则默认不更改相应参数。你也可通过命令行获取相关信息：
arkfbp-py ext_updatenode -h
RemoveNode
在流的图定义（create_nodes）中删除一个已知的节点信息，并自动更新前驱后继节点的连接信息。
python3 manage.py ext_removenode --flow <flow_name> --id <node_id>
如果使用arkfbp-py命令，需指定--topdir参数，其代表项目的绝对根路径：
arkfbp-py ext_removenode --flow app1.flows.flow1 --id node1 --topdir /Users/user/Development/demo
详解
参数flow代表流的路径以.分隔，具体到流的文件夹名称；参数id代表目标节点的唯一标识，用于指定删除的目标节点；
参数id、flow是必选，其他可选。你也可通过命令行获取相关信息：
arkfbp-py ext_removenode -h
special usages
csrf
若想局部禁用或模拟csrf，只需要重写指定flow的Main Class的dispatch方法。示例如下：
from arkfbp.flow import ViewFlow
from arkfbp.node import StartNode, StopNode
from django.views.decorators.csrf import csrf_exempt
class Main(ViewFlow):
def create_nodes(self):
return [{
'cls': StartNode,
'id': 'start',
'next': 'stop',
'x': None,
'y': None
}，
{
'cls': StopNode,
'id': 'stop',
'next': None,
'x': None,
'y': None
}]
@csrf_exempt
def dispatch(self, request, *args, **kwargs):
return super(Main, self).dispatch(request, *args, **kwargs)
AuthTokenNode
现在可以使用AuthTokenNode来快速搭建您的用户名+密码验证流程，示例如下：
from arkfbp.node import AuthTokenNode
class VerifyPassword(AuthTokenNode):
def get_ciphertext(self):
return 'ciphertext'
def before_execute(self, *args, **kwargs):
self.username_field = 'USERNAME'
self.password_field = 'PASSWORD'
详解
其中，get_ciphertext()用于自定义从存储后端获取加密的数据；get_key()可自定义返回的token值，默认为生成一个新的token值；
你也可以通过before_execute()等run()方法运行前的钩子来自定义username_field和password_field来指定获取账号名和账号密码的字段名称；
AuthTokenNode在run()运行后默认返回一个长度为40的token字符串。
Auto-generated code
编辑 meta-config
meta-config最外层结构如下：
{
"name": "",
"type": "",
"module": {},
"meta": {},
"permission": {},
"api": {}
}
name
meta_config的名称，唯一标识（推荐和文件名相同）。
{
"name": "meta_config_name"
}
type
前端组件类型。
{
"type": “table"
}
module
model类及meta文件的具体路径。
"module": {
"user": {
"model": "arkid_meta.models.user.User"
},
"util": {
"meta": "automation.util"
}
}
permission
权限校验相关的路径。
{
"permission": {
"role": "demo.permission.role"
}
}
其中role表示别名即命名空间，demo.permission.role指定的为role相关的meta config的JSON文件，实例如下：
{
"admin": {
"title": "管理员",
"flow": "demo.permission.role.admin"
}
}
其中admin为权限角色名称，title为权限名字，flow指定了具体校验时需要运行的工作流。
使用方法
在api配置中增加permission字段来标识需要用到的permission。
{
"api": {
"user/": {
"post": {
"name": "新建用户",
"type": "create",
"request": {},
"response": {},
"permission": ["role.admin"] # role为上述的命名空间，admin为文件中指定的admin角色。
}
}
}
}
meta
包含了model所有的字段信息及校验规则，书写方式分为module导入，或者自定义。
{
"meta": {
"field_1": {
"title": "title_1",
"type": {
"field_type": {}
}
}
}
}
field_1
展示的字段名称，并不代表model中原始的字段名称。
title_1
字段的名称，用于前端展示。
field_type
字段的类型，目前支持string、integer、float、object、array。
{
"meta": {
"field_1": {
"title": "title_1",
"required": true, # 必须接受此参数
"type": {
"string": {
"read_only"：false， # 只读
"write_only"：true，# 只写
"min_length": 10, # 字符串最小的长度
"max_length": 50, # 字符串最大的长度
}
}
}
}
}
object field type
"field": {
"title": "title",
"type": {
"object": {
"field_1": "field_1",
"field_2": "field_2",
"field_3": "field_3",
}
}
}
array object type
"field": {
"title": "查询结果列表",
"type": {
"array": {
"array_item": "field_1"
}
}
}
api
接口定义。
"meta_name/<index>/": { # url，index为位置参数
"get": { # 接口的请求方法
"name": "update_meta_name", # 接口的名称
"type": "retrieve", # 接口的默认类型
"index": { # 位置参数的配置
"id": { # 位置参数名称
"src": "model_user.id" # 配置来源
}
},
"pagination": { # 分页配置
"enabled": true, # 是否启用
"page_size_query_param": "page_size", # 传参的key名称，页面大小
"page_query_param": "page", # 传参的key名称，页码
"count_param": "count", # 记录总数的名称
"results_param": "results", # 结果的名称
"next_param": "next", # 下一页的名称
"previous_param": "previous", # 上一页的名称
"paginated_response": "utils.custom_response" # 自定义分页response，需清楚具体pagination node的response实现
},
"request": {}, # 接口需要接收的字段
"response": { # 接口需要返回的字段
"data": "items", # 表示本地meta中的配置
"error_code": "util.error_code", # 表示从module导入的配置
"error_message": "util.error_message" # 表示从module导入的配置
},
"debug": false # 是否输出debug信息，默认为true
},
"delete": {
"index": "index",
"name": "delete_meta_name",
"http_method": "delete",
"request": [],
"response": []
}
}
pagination response
若想自定义分页的数据结构，你需要用到.pagination内置用法来重构响应的数据结构。
{"meta":
"data": {
"required": false,
"type": {
"object": {
"total": ".pagination.count",
"page": ".pagination.page",
"page_size": ".pagination.page_size",
"items": "items"
}
}
}
}
permission flow
在api描述中定义permission并引入role字段中定义的角色,
其中admin.flow是用于校验权限的工作流，其输出值为布尔类型。
{
"role": {
"admin": {
"title": "管理员",
"flow": "flows.flow"
}
},
"api": {
"user/": {
"get": {
"name": "获取信息",
"type": "retrieve",
"request": {},
"response": {},
"debug": false,
"permission": ["admin"]
}
}
}
}
在开启系统默认
custom type for api
除了create、update、retrieve、delete四种系统提供的基本的数据处理引擎，你还可以进行自定义引擎的配置。 此时不需要指定response参数。
"custom/": {
"post": {
"name": "custom_1",
"type": "custom",
"flow": "flows.flow_1", # 指定自定义流的位置
"request": {}, # 接口需要接收的字段
}
}
详解：自定义流运行之前系统会根据request中的参数先进行数据校验， 之后将validate的_data及原始的request传给自定义的flow
配置meta_config
将meta_config文件与django结合，以达到自动生成项目的效果。
编写JSON文件
将所有的meta_config统一存放到项目的某一文件夹下。
demo
|_ automation
|_ meta_1.json
|_ ...
|_ meta_n.json
配置url
在django项目的主urls.py文件中增加一条路由
from django.contrib import admin
from django.urls import path, include
from arkfbp.common.automation.core import MetaConfigs
meta_dir = '/demo/automation'
urlpatterns = [
path('admin/', admin.site.urls),
path('arkfbp-admin/', include(MetaConfigs(meta_dir).get_urls()))
]
运行项目
python manage.py runserver
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.
