112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
"""Database-driven workflow definitions.
|
|
|
|
The engine knows only generic concepts (steps, transitions, conditions,
|
|
versions). Business processes (washing, drying, sorting, ...) are data, never
|
|
code. Do NOT add hard-coded step names or branching here.
|
|
"""
|
|
from django.db import models
|
|
|
|
from apps.core.models import TimeStampedModel, UUIDModel
|
|
|
|
|
|
class Workflow(UUIDModel, TimeStampedModel):
|
|
class Status(models.TextChoices):
|
|
DRAFT = "draft", "Draft"
|
|
ACTIVE = "active", "Active"
|
|
INACTIVE = "inactive", "Inactive"
|
|
ARCHIVED = "archived", "Archived"
|
|
|
|
name = models.CharField(max_length=200)
|
|
slug = models.SlugField(max_length=200, unique=True)
|
|
description = models.TextField(blank=True)
|
|
status = models.CharField(max_length=16, choices=Status.choices, default=Status.DRAFT)
|
|
active_version = models.OneToOneField(
|
|
"WorkflowVersion", null=True, blank=True, on_delete=models.SET_NULL, related_name="+"
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
return self.name
|
|
|
|
|
|
class WorkflowVersion(UUIDModel, TimeStampedModel):
|
|
class Status(models.TextChoices):
|
|
DRAFT = "draft", "Draft"
|
|
ACTIVE = "active", "Active"
|
|
INACTIVE = "inactive", "Inactive"
|
|
ARCHIVED = "archived", "Archived"
|
|
|
|
workflow = models.ForeignKey(Workflow, on_delete=models.CASCADE, related_name="versions")
|
|
version = models.PositiveIntegerField()
|
|
status = models.CharField(max_length=16, choices=Status.choices, default=Status.DRAFT)
|
|
published_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
ordering = ["-version"]
|
|
constraints = [
|
|
models.UniqueConstraint(fields=["workflow", "version"], name="uniq_workflow_version")
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.workflow.slug} v{self.version}"
|
|
|
|
|
|
class WorkflowStep(UUIDModel):
|
|
class StepType(models.TextChoices):
|
|
TASK = "task", "Task"
|
|
DECISION = "decision", "Decision"
|
|
INPUT = "input", "Input"
|
|
OUTPUT = "output", "Output"
|
|
INTEGRATION = "integration", "Integration"
|
|
|
|
version = models.ForeignKey(WorkflowVersion, on_delete=models.CASCADE, related_name="steps")
|
|
key = models.SlugField(max_length=100)
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
order = models.PositiveIntegerField(default=0)
|
|
step_type = models.CharField(max_length=16, choices=StepType.choices, default=StepType.TASK)
|
|
required = models.BooleanField(default=False)
|
|
enabled = models.BooleanField(default=True)
|
|
estimated_duration_minutes = models.PositiveIntegerField(default=0)
|
|
is_start = models.BooleanField(default=False)
|
|
is_terminal = models.BooleanField(default=False)
|
|
requires_customer_confirmation = models.BooleanField(default=False)
|
|
requires_payment = models.BooleanField(default=False)
|
|
triggers_invoice = models.BooleanField(default=False)
|
|
triggers_odoo_sync = models.BooleanField(default=False)
|
|
customer_visible = models.BooleanField(default=True)
|
|
operator_visible = models.BooleanField(default=True)
|
|
live_monitoring_enabled = models.BooleanField(default=False)
|
|
input_schema = models.JSONField(default=dict, blank=True)
|
|
output_schema = models.JSONField(default=dict, blank=True)
|
|
# Optional condition (dict) that gates whether this step is available.
|
|
availability_condition = models.JSONField(default=dict, blank=True)
|
|
# Canvas position for the visual workflow designer.
|
|
position_x = models.FloatField(default=0)
|
|
position_y = models.FloatField(default=0)
|
|
|
|
class Meta:
|
|
ordering = ["order"]
|
|
constraints = [
|
|
models.UniqueConstraint(fields=["version", "key"], name="uniq_step_key")
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.key} ({self.version})"
|
|
|
|
|
|
class WorkflowTransition(UUIDModel):
|
|
version = models.ForeignKey(WorkflowVersion, on_delete=models.CASCADE, related_name="transitions")
|
|
from_step = models.ForeignKey(WorkflowStep, on_delete=models.CASCADE, related_name="outgoing")
|
|
to_step = models.ForeignKey(WorkflowStep, on_delete=models.CASCADE, related_name="incoming")
|
|
# List of condition dicts, e.g. [{"type": "step_completed", "step": "drying"}]
|
|
conditions = models.JSONField(default=list, blank=True)
|
|
required_roles = models.ManyToManyField("rbac.Role", blank=True)
|
|
|
|
class Meta:
|
|
constraints = [
|
|
models.UniqueConstraint(fields=["version", "from_step", "to_step"], name="uniq_transition")
|
|
]
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.from_step.key} -> {self.to_step.key}"
|