私のプロジェクトにワークフローエンジンを実装しようとしています。主な目的は移植可能なアプリケーションを作成することです。何か、私は将来的に他のプロジェクトに配置して、プロジェクトのさまざまなモデルにワークフローを添付して動作させることができます。信号とCelery-djangoを使用したDjangoワークフローエンジン
アプローチを考えようとしましたが、完璧な設定ではないようです。プロジェクト内にワークフローアプリケーションを作成し、2種類のモデルを添付し、ワークフロー(ワークフロー、ステップ、アクション)の設定を含むものと、インスタンス/トランザクションを含む他のモデルを考えました。
以下は、私は私のモデル設計が同様に強化する必要があるかもしれないと思う私のワークフロー/ 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)
と私も持っているワークフロー/ 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を挿入することができました。もしそうなら、私は特定のワークフローをトリガーします。
アクションについては、アクションを開発する必要があることを理解しています。たぶん私は、ワークフローのアプリケーションの下にアクションフォルダを作成する必要があります、各アクションをクラスに入れます。それぞれが電子メールを送信し、アーカイブし、データベースの値を更新する特定のタスクを実行します。
もし車輪を改革しているのであれば、すでに開発されていることがあれば誰でもお勧めできますそれを通過する。
敬具、例えば
あなたの質問は言い換えることができると思います。あなたが本当に尋ねるものは明確ではありません。 – roman