大家好，又见面了，我是你们的朋友全栈君。

研究背景

自己的项目都是python3.6开发。想使用蓝鲸的流程系统，真是千难万难。魔改路上真是一路坎坷。由于BK-SOPS需要结合蓝鲸的一整套服务才能够运行，所以单独把标准运维的流程系统抽出来然后融合进自己的系统。

看看蓝鲸标准运维的功能

• 多元接入支持：标准运维对接了蓝鲸通知、作业平台、配置平台等服务，作为官方标准插件库提供服务，还支持用户自定义接入企业内部系统，定制 开发标准插件。
• 可视化流程编排：通过拖拽方式组合标准插件节点到一个流程模板。
• 多种流程模式：支持标准插件节点的串行、并行，支持子流程，可以根据全局参数自动选择 分支执行，节点失败处理机制可配置。
• 参数引擎：支持参数共享，支持参数替换。
  可交互的任务执行：任务执行中可以随时暂停、继续、撤销，节点失败后可以重试、跳过。
• 通用权限管理：通过配置平台同步业务角色，支持流程模板的使用权限控制。

改造日志

2020年10月20日

有博友留言问能不能独立部署，是可以的。下面是我之前研究的代码。能运行并且自定义节点什么的，我没有改完。欢迎大家一起学习，分享出来我之前研究的源码

https://github.com/CyberWong/django_pipeline

2019年08月23日

这里用到比较多的就是元类。
元类就是在构建类时候运行的，蓝鲸的组件就是在构建类时候把组件写到一个公共类，然后构造流程时候通过get方法，返回组件类。

代码一

from pipeline.component_framework.component import Component
class TestComponent(Component):
    name = 'test'
    code = 'test'
    bound_service = TestService
    form = 'test.js'

这就是个标准的test组件。继承Component，看看Component的结构

from pipeline.component_framework.base import ComponentMeta
class Component(object,metaclass=ComponentMeta):

    def __init__(self, data_dict):
        self.data_dict = data_dict

    @classmethod
    def outputs_format(cls):
        outputs = cls.bound_service().outputs()
        outputs = map(lambda oi: oi._asdict(), outputs)
        return outputs

    def clean_execute_data(self, context):
        return self.data_dict

    def data_for_execution(self, context, pipeline_data):
        data_dict = self.clean_execute_data(context)
        inputs = { 
   }

        for key, tag_info in data_dict.items():
            if tag_info is None:
                raise ComponentDataLackException('Lack of inputs: %s' % key)

            inputs[key] = get_variable(key, tag_info, context, pipeline_data)

        return DataObject(inputs)

    def service(self):
        return self.bound_service()

Python2和Python3使用元类的语法不一样，sops用的是Python2，这里需要修改一下

# python2
class Component(object):
    __metaclass__ = ComponentMeta

在python3里面，Python3默认就是新式类，声明的语法如下：

# python3
class Component(object,metaclass=ComponentMeta):
# or
class Component(metaclass=ComponentMeta):

研究了好久为什么通过ComponentLibrary获取不到自己定义的组件。ComponentLibrary就是一个组件库，Component类初始化时候会到ComponentLibrary组件库类中，注册的方法就是通过元类，也就是上面的ComponentMeta

class ComponentMeta(type):
def __new__(cls, name, bases, attrs):
super_new = super(ComponentMeta, cls).__new__
# Also ensure initialization is only performed for subclasses of Model
# (excluding Model class itself).
parents = [b for b in bases if isinstance(b, ComponentMeta)]
# 为了确保只对子类初始化
if not parents:
return super_new(cls, name, bases, attrs)
# Create the class
# 创建一个类，并动态引入
module_name = attrs.pop('__module__')
new_class = super_new(cls, name, bases, { 
'__module__': module_name})
module = importlib.import_module(new_class.__module__)
# Add all attributes to the class
attrs.setdefault('desc', '')
for obj_name, obj in attrs.items():
setattr(new_class, obj_name, obj)
# check
if not getattr(new_class, 'name', None):
raise ValueError("component %s name can't be empty" %
new_class.__name__)
if not getattr(new_class, 'code', None):
raise ValueError("component %s code can't be empty" %
new_class.__name__)
if not getattr(new_class, 'bound_service', None) or not issubclass(new_class.bound_service, Service):
raise ValueError("component %s service can't be empty and must be subclass of Service" %
new_class.__name__)
if not getattr(new_class, 'form', None):
setattr(new_class, 'form', None)
# category/group name
# 在这里我们可以发现，可以自己自定义组名。使用__group_name__
group_name = getattr(
module, "__group_name__",
new_class.__module__.split(".")[-1].title()
)
setattr(new_class, 'group_name', group_name)
new_name = u"%s-%s" % (group_name, new_class.name)
# 自定义icon
# category/group name
group_icon = getattr(
module, "__group_icon__",
''
)
setattr(new_class, 'group_icon', group_icon)
if not getattr(module, '__register_ignore__', False):
# 就是在这里
ComponentLibrary.components[new_class.code] = new_class
# try:
# ComponentModel.objects.update_or_create(
# code=new_class.code,
# defaults={ 

# 'name': new_name,
# 'status': __debug__,
# }
# )
# except Exception as e:
# if not isinstance(e, ProgrammingError):
# logging.exception(e)
return new_class

下面再看看ComponentLibrary这个东西,很有趣的一种设计方法。

class ComponentLibrary(object):
components = { 
}
def __new__(cls, *args, **kwargs):
# 
component_code = kwargs.get('component_code', None)
if args:
component_code = args[0]
if not component_code:
raise ValueError('please pass a component_code in args or kwargs: '
'ComponentLibrary(\'code\') or ComponentLibrary(component_code=\'code\')')
if component_code not in cls.components:
raise ComponentNotExistException('component %s does not exist.' %
component_code)
return cls.components[component_code]
@classmethod
def get_component_class(cls, component_code):
return cls.components.get(component_code)
@classmethod
def get_component(cls, component_code, data_dict):
return cls.get_component_class(component_code)(data_dict)

受益匪浅。大牛的设计思想果然牛逼

蓝鲸bk-sops源码学习

一：流程图前端一
二：流程图前端二之节点创建
三：流程组件注册实现原理