我正在努力在我的项目中实现工作流引擎,我尝试的主要目的是创建一个可移植应用程序。有些东西,我将来可以放入任何其他项目,然后将工作流附加到我项目中的不同模型并让它工作。
我试图想出一种方法,但它似乎不是完美的设置。我想在我的项目中创建一个工作流应用程序,附加两种模型,一些模型将包含工作流的设置(工作流、步骤、操作),而其他模型将包含实例/事务。
下面是我的workflow/models.py
from django.db import models
from django.contrib.contenttypes.models import ContentType
from django.contrib.contenttypes import generic
import signals
################################
# Workflow engine models
################################
class Signal_Listener(models.Model):
LISTENING_ON = (
('I', 'INSERT'),
('U', 'UPDATE'),
('D', 'DELETE'),
('S', 'SELECT'),
('A', 'ANY QUERY'),
('L', 'ANY DDL'),
)
table_name = models.CharField(max_length=100)
listening_to = models.CharField(max_length=1, choices=LISTENING_ON)
class Meta:
unique_together = (("table_name", "listening_to"),)
def __unicode__(self):
return '%s - %s' % (self.table_name, self.listening_to)
class Action(models.Model):
ACTION_TYPE_CHOICES = (
('P', 'Python Script' ),
('C', 'Class name' ),
)
name = models.CharField(max_length=100)
action_type = models.CharField(max_length=1, choices=ACTION_TYPE_CHOICES)
audit_obj = generic.GenericRelation('core.Audit', editable=False)
class Steps(models.Model):
sequence = models.IntegerField()
Action = models.ForeignKey(Action)
Signal_Listener = models.ForeignKey(Signal_Listener)
class Process(models.Model):
## TODO: Document
# Processes class is used to store information about the process itself.
# Or in another word, the workflow name.
WF_TYPE_LIST = (
('P', 'Python-API'),
)
name = models.CharField(max_length=30)
is_active = models.BooleanField()
wf_type = models.CharField(max_length=1, choices=WF_TYPE_LIST)
audit_obj = generic.GenericRelation('core.Audit', editable=False)
listening_to = models.ForeignKey(Steps)
################################
# Workflow transactions models
################################
class Instance(models.Model):
## TODO: Document
# Re
INSTANCE_STATUS = (
('I', 'In Progress' ),
('C', 'Cancelled' ),
('A', 'Archived' ), # Old completed tasks can be archived
('P', 'Pending' ),
('O', 'Completed' )
)
id = models.CharField(max_length=200, primary_key=True)
status = models.CharField(max_length=1, choices=INSTANCE_STATUS, db_index=True)
audit_obj = generic.GenericRelation('core.Audit', editable=False)
def save(self, *args, **kwargs):
# on new records generate a new uuid
if self.id is None or self.id.__len__() is 0:
import uuid
self.id = uuid.uuid4().__str__()
super(Instances, self).save(*args, **kwargs)
class Task(models.Model):
TASK_STATUS = (
('S', 'Assigned' ),
('I', 'In Progress' ),
('P', 'Pending' ),
('C', 'Cancelled' ),
('A', 'Archived' ), # Old completed tasks can be archived
('O', 'Completed' )
)
name = models.CharField(max_length=100)
instance = models.ForeignKey(Instance)
status = models.CharField(max_length=1, choices=TASK_STATUS)
bio = models.CharField(max_length=100)
audit_obj = generic.GenericRelation('core.Audit', editable=False)
我还有一个 workflow/signals.py
"""
Workflow Signals
Will be used to monitor all inserts, update, delete or select statements
If an action is attached to that particular table, it will be inserted Celery-Tasks distribution.
"""
from django.db.models.signals import post_save, post_delete
from django.core.cache import cache
def workflow_post_init_listener(sender, **kwargs):
try:
if cache.get('wf_listner_cache_%s' % kwargs['instance']._meta.db_table):
pass
else:
record = 'Signal_Listener'.objects.get(table_name__exact=kwargs['instance']._meta.db_table)
# am not sure what to do next!
except 'Signal_Listener'.DoesNotExist:
# TODO: Error logging
pass
post_save.connect(workflow_post_init_listener, dispatch_uid="workflow.models.listener.save")
我认为我的模型设计可能也需要改进。我可以在几个场景中使用,我想从批准周期开始,例如,我可以在 signal_listener 中插入 table_name/model_name 以监视新插入。如果是这样,我将触发特定的工作流程。
至于行动,我明白行动需要被开发。也许我需要在工作流应用程序下创建一个 Action 文件夹,将每个 Action 放在一个类中。每个人都将完成发送电子邮件、存档、更新数据库值等特定任务
如果我正在重新发明轮子并且已经开发出这样的东西,任何人都可以提出建议,我会非常乐意经历它。
最好的问候,
最佳答案
例如,您可以查看 zope.wfmc (http://pypi.python.org/pypi/zope.wfmc):这是一个工作流管理联盟的实现 可以在 XPDL 上定义的工作流。
关于python - 使用信号和 Celery-django 的 Django 工作流引擎,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/4624247/