简易流水线表单设计

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

from django.db import models

def local_timestamp(timestamp):
    """Convert datetime field to local timestamp."""
    return timestamp.astimezone(
        timezone("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S")


class Pipeline(models.Model):
    """Pipeline task template."""

    # 用于channel队列名称，结果需要放到rabbit对应的{name}_result channel中
    name = models.CharField(max_length=100, verbose_name='工作流标识符', unique=True)
    nickname = models.CharField(max_length=100, verbose_name='工作流显示名称')
    p_tag = models.CharField(max_length=50,
                             verbose_name='标签',
                             default='default')

    p_creator = models.CharField(max_length=50, verbose_name='创建人')
    # json format string
    html_form_json = models.TextField(verbose_name='表单类型')
    # 0: 禁用 1: 启用
    status = models.IntegerField(verbose_name='工作流可用状态', default=1)
    description = models.CharField(max_length=100, verbose_name='工作流描述')
    create_timestamp = models.DateTimeField(auto_now_add=True,
                                            verbose_name="创建时间戳")
    update_timestamp = models.DateTimeField(auto_now=True,
                                            verbose_name="更新时间戳")

    def to_json(self):
        """Json format of pipeline."""
        return {
            "id": self.id,
            "name": self.name,
            "nickname": self.nickname,
            "p_tag": self.p_tag,
            "p_creator": self.p_creator,
            "html_form_json": self.html_form_json,
            "status": self.status,
            "description": self.description,
            "create_timestamp": str(self.create_timestamp),
            "update_timestamp": str(self.update_timestamp)
        }


class PipelineTask(models.Model):
    """Pipeline actual task."""

    p_id = models.IntegerField(verbose_name='工作流id')
    data_json = models.TextField(verbose_name='表单数据')
    trigger = models.CharField(verbose_name="触发人", max_length=50, default='')
    # json format string
    result = models.TextField(verbose_name='任务结果')
    # -1: 就绪状态，等待运行 0: 立即放到运行队列 1: 运行中 2: 运行完成 3: 程序异常终止 4: 用户终止 5: 运行失败
    status = models.IntegerField(verbose_name='工作流任务状态')
    create_timestamp = models.DateTimeField(auto_now_add=True,
                                            verbose_name="创建时间戳")
    update_timestamp = models.DateTimeField(auto_now=True,
                                            verbose_name="更新时间戳")

    def to_json(self):
        """Json format of pipeline task."""
        return {
            "id": self.id,
            "p_id": self.p_id,
            "trigger": self.trigger,
            "data_json": self.data_json,
            "result": self.result,
            "status": self.status,
            "create_timestamp": local_timestamp(self.create_timestamp),
            "update_timestamp": local_timestamp(self.update_timestamp)
        }


class PipelineTaskComment(models.Model):
    """Pipeline task comments."""

    pt_id = models.IntegerField(verbose_name='工作流task id')
    creator = models.CharField(verbose_name='评论人',
                               max_length=50,
                               default='robot')
    c_type = models.CharField(verbose_name='评论类型', max_length=30)
    c_message = models.TextField(verbose_name='评论内容')
    create_timestamp = models.DateTimeField(auto_now_add=True,
                                            verbose_name="创建时间戳")

    def to_json(self):
        """Json format of pipeline task comments."""
        return {
            "id": self.id,
            "creator": self.creator,
            "pt_id": self.pt_id,
            "c_type": self.c_type,
            "c_message": self.c_message,
            "create_timestamp": local_timestamp(self.create_timestamp)
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

def AddPipelineTask(params):
    """Add pipeline_task with needed param."""
    if 'p_id' not in params or 'status' not in params:
        return fast_return(10835, extra=params['p_id'])
    p_id = params["p_id"]
    status = params["status"]
    data_json = params.get('data_json', '{}')

    # only add ready and send to running
    if status not in [0, -1]:
        return fast_return(10942, extra=status)

    pipeline_obj = Pipeline.objects.filter(id=p_id).first()
    if not pipeline_obj:
        logger.info("pipeline不存在 ", p_id)
        return fast_return(10933, extra=p_id)
    else:
        try:
            pipeline_task_obj = PipelineTask.objects.create(**params)
            # send to running task queue
            if status == 0:
                # will send to rabbit mq queue with pipeline name group
                logger.info('add pipeline task ' + str(pipeline_obj.name) +
                            str(data_json))
                try:
                    # add a comment to task detail page
                    comment = {
                        "pt_id": pipeline_task_obj.id,
                        "c_type": "auto",
                        "creator": "robot",
                        "c_message": "任务已经触发，即将运行!"
                    }
                    AddPipelineTaskComment(comment)

                    # send task info to mq
                    rb = RabbitMQ(host=RabbitMQ_HOST)
                    rb.send(exchange='',
                            queue=pipeline_obj.name,
                            durable=True,
                            body=json.dumps({
                                'args': pipeline_task_obj.data_json,
                                'task_id': pipeline_task_obj.id,
                                'pipeline_id': pipeline_obj.id,
                                'pipeline_name': pipeline_obj.name
                            }))
                except Exception as e:
                    logger.error('pipeline task send to mq error', str(e))
                finally:
                    return fast_return(code=10945,
                                       extra=pipeline_obj.name,
                                       data=pipeline_task_obj.to_json())

        except Exception as e:
            return fast_return(10935, extra=str(e))
        finally:
            pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Date: Mon Sep  5 09:57:26 2022
# Author: liuliancao <liuliancao@gmail.com>
"""Description: Rabbit mq related."""
import pika


class RabbitMQ(object):
    """Rabbit MQ related."""

    connection = None

    def __init__(self, host="127.0.0.1"):
        """Rabbitmq init related."""
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host))

    def send(self, queue='', durable=False, exchange='', body='{}'):
        """Rabbitmq send message."""
        ret = {"code": 0, "data": None, "message": "init"}
        if queue:
            try:
                with self.connection.channel() as ch:

                    ch.queue_declare(queue=queue, durable=durable)

                    if exchange:
                        ch.exchange_declare(exchange=exchange, durable=durable)

                    if durable:
                        sr = ch.basic_publish(exchange=exchange,
                                              routing_key=queue,
                                              body=body,
                                              properties=pika.BasicProperties(
                                                  delivery_mode=2, ))
                    else:
                        sr = ch.basic_publish(exchange=exchange,
                                              routing_key=queue,
                                              body=body)
                ret['code'] = 200
                ret['message'] = 'add message successfully ' + queue
                ret['data'] = sr
            except Exception as e:
                ret['message'] = 'add message failed ' + str(e)
                ret['data'] = None
                ret['code'] = 500
            finally:
                return ret
        return None

    def receive(self,
                queue='',
                durable=False,
                callback=None,
                auto_ack=True,
                exchange=''):
        """Rabbitmq receive message."""
        if queue:
            with self.connection.channel() as ch:
                if exchange:
                    ch.exchange_declare(exchange=exchange)
                ch.queue_declare(queue=queue, durable=durable)
                ch.basic_consume(queue=queue,
                                 auto_ack=True,
                                 on_message_callback=callback)
                ch.start_consuming()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Date: Mon Sep  5 17:42:12 2022
# Author: liuliancao <liuliancao@gmail.com>
"""Description: Rabbitmq related."""
from django.core.management.base import BaseCommand

class ResultCatcher(object):
    """Pipeline result catcher."""

    def listen(self):
        """Fetch the result from the rabbitmq and update."""
        rb = RabbitMQ(host=RabbitMQ_HOST)
        rb.receive(queue=RabbitMQ_PipelineResultQueue,
                   durable=True,
                   callback=callback)


class Command(BaseCommand):
    """Used for manage.py command rabbitmq."""

    help = 'Run rabbit mq consumer for result catch.'

    def handle(self, *args, **kwargs):
        """Run rabbitmq."""
        print("running events pipeline result catcher...")
        rt = ResultCatcher()
        rt.listen()