arkfbp-py is the python implementation of the arkfbp.
arkfbp-py需要 Python 3.6+ 及Django 2.0+ 的版本支持。
pip3 install arkfbp (暂不可用)
or
pip3 install git+https://github.com/longguikeji/arkfbp-py.git@zzr/basic
python3 setup.py install
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
全局钩子式工作流运行的场景适用于：
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',
]
GlobalHookFlow的执行顺序与django原生Middleware执行顺序一致，
before_route()、before_flow()的执行顺序依次为从上至下；after_flow()、before_exception()则为从下至上。
全新的钩子流现已可以使用。
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）
列表中流的摆放顺序，即为执行顺序。
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
现在，你可以通过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.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.steps为一个dict，其中包含以node_id为key、以node_instance为value的数据。
现在你可以在任何一个节点，从node.state.steps中，获取指定的已运行的node。
node1 = node.state.steps.get('node1', None)
ViewFlow的inputs为原生的django的WSGIRequest对象，ViewFlow在此基础上为inputs对象增加了data、extra_data、str属性。
ds属性将原生WSGIRequest对象的GET和POST的数据合并为一个dict。
你可以在extra_ds中存放你想要传递下去的任何数据。
str包含了请求体中的字符串信息。
注意：你可以随意为inputs增加任何属性，例如：
inputs.attr = {}
这样你就为inputs增加了attr的属性
现在你可以通过指定目录和基类来创建一个工作流，--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
现在你可以通过指定目录和基类来创建一个流节点，--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
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'
}
]
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
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，即可直接运行所有测试流
此部分内容适用于可视化插件开发相关人员
在流的图定义（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
在流的图定义（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
在流的图定义（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
若想局部禁用或模拟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来快速搭建您的用户名+密码验证流程，示例如下：
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字符串。
meta-config最外层结构如下：
{
"name": "",
"type": "",
"module": {},
"meta": {},
"permission": {},
"api": {}
}
meta_config的名称，唯一标识（推荐和文件名相同）。
{
"name": "meta_config_name"
}
前端组件类型。
{
"type": “table"
}
model类及meta文件的具体路径。
"module": {
"user": {
"model": "arkid_meta.models.user.User"
},
"util": {
"meta": "automation.util"
}
}
权限校验相关的路径。
{
"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角色。
}
}
}
}
包含了model所有的字段信息及校验规则，书写方式分为module导入，或者自定义。
{
"meta": {
"field_1": {
"title": "title_1",
"type": {
"field_type": {}
}
}
}
}
展示的字段名称，并不代表model中原始的字段名称。
字段的名称，用于前端展示。
字段的类型，目前支持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, # 字符串最大的长度
}
}
}
}
}
"field": {
"title": "title",
"type": {
"object": {
"field_1": "field_1",
"field_2": "field_2",
"field_3": "field_3",
}
}
}
"field": {
"title": "查询结果列表",
"type": {
"array": {
"array_item": "field_1"
}
}
}
接口定义。
"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内置用法来重构响应的数据结构。
{"meta":
"data": {
"required": false,
"type": {
"object": {
"total": ".pagination.count",
"page": ".pagination.page",
"page_size": ".pagination.page_size",
"items": "items"
}
}
}
}
在api描述中定义permission并引入role字段中定义的角色,
其中admin.flow是用于校验权限的工作流，其输出值为布尔类型。
{
"role": {
"admin": {
"title": "管理员",
"flow": "flows.flow"
}
},
"api": {
"user/": {
"get": {
"name": "获取信息",
"type": "retrieve",
"request": {},
"response": {},
"debug": false,
"permission": ["admin"]
}
}
}
}
在开启系统默认
除了create、update、retrieve、delete四种系统提供的基本的数据处理引擎，你还可以进行自定义引擎的配置。 此时不需要指定response参数。
"custom/": {
"post": {
"name": "custom_1",
"type": "custom",
"flow": "flows.flow_1", # 指定自定义流的位置
"request": {}, # 接口需要接收的字段
}
}
详解：自定义流运行之前系统会根据request中的参数先进行数据校验， 之后将validate的_data及原始的request传给自定义的flow
将meta_config文件与django结合，以达到自动生成项目的效果。
将所有的meta_config统一存放到项目的某一文件夹下。
demo
|_ automation
|_ meta_1.json
|_ ...
|_ meta_n.json
在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