170 lines
5.4 KiB
Python
170 lines
5.4 KiB
Python
"""Generic, database-driven workflow engine.
|
|
|
|
The engine understands only generic concepts: steps, transitions, conditions,
|
|
and versions. It has no knowledge of "washing", "drying", or any other
|
|
business-specific term.
|
|
"""
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.core.errors import ApiError
|
|
|
|
|
|
class WorkflowError(ApiError):
|
|
def __init__(self, message: str, details=None, code: str = "WORKFLOW_INVALID"):
|
|
super().__init__(code, message, details)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Condition registry. New condition types can be added without touching the
|
|
# rest of the system.
|
|
# ---------------------------------------------------------------------------
|
|
CONDITION_HANDLERS = {}
|
|
|
|
|
|
def register_condition(key: str):
|
|
def decorator(fn):
|
|
CONDITION_HANDLERS[key] = fn
|
|
return fn
|
|
|
|
return decorator
|
|
|
|
|
|
@register_condition("step_selected")
|
|
def _cond_step_selected(ctx, cond):
|
|
return cond.get("step") in ctx.get("selected_step_keys", [])
|
|
|
|
|
|
@register_condition("step_completed")
|
|
def _cond_step_completed(ctx, cond):
|
|
return cond.get("step") in ctx.get("completed_step_keys", [])
|
|
|
|
|
|
@register_condition("weight_gte")
|
|
def _cond_weight_gte(ctx, cond):
|
|
try:
|
|
return float(ctx.get("net_weight") or 0) >= float(cond.get("value", 0))
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
@register_condition("weight_lte")
|
|
def _cond_weight_lte(ctx, cond):
|
|
try:
|
|
return float(ctx.get("net_weight") or 0) <= float(cond.get("value", 0))
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
@register_condition("payment_status")
|
|
def _cond_payment_status(ctx, cond):
|
|
return ctx.get("payment_status") == cond.get("value")
|
|
|
|
|
|
@register_condition("customer_category")
|
|
def _cond_customer_category(ctx, cond):
|
|
return ctx.get("customer_category") == cond.get("value")
|
|
|
|
|
|
@register_condition("role")
|
|
def _cond_role(ctx, cond):
|
|
return cond.get("value") in ctx.get("user_roles", [])
|
|
|
|
|
|
@register_condition("field")
|
|
def _cond_field(ctx, cond):
|
|
return ctx.get("fields", {}).get(cond.get("key")) == cond.get("value")
|
|
|
|
|
|
def evaluate_condition(cond, context) -> bool:
|
|
handler = CONDITION_HANDLERS.get(cond.get("type"))
|
|
if handler is None:
|
|
return False
|
|
try:
|
|
return bool(handler(context, cond))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def evaluate_conditions(conditions, context) -> bool:
|
|
if not conditions:
|
|
return True
|
|
return all(evaluate_condition(c, context) for c in conditions)
|
|
|
|
|
|
class WorkflowEngine:
|
|
"""Validates and publishes workflow versions and evaluates transitions."""
|
|
|
|
def validate_version(self, version) -> bool:
|
|
errors = []
|
|
steps = list(version.steps.all())
|
|
if not steps:
|
|
errors.append("workflow has no steps")
|
|
if not any(s.is_start for s in steps):
|
|
errors.append("missing start step")
|
|
if not any(s.is_terminal for s in steps):
|
|
errors.append("missing terminal step")
|
|
step_ids = {s.id for s in steps}
|
|
transitions = list(version.transitions.select_related("from_step", "to_step").all())
|
|
for t in transitions:
|
|
if t.from_step_id not in step_ids or t.to_step_id not in step_ids:
|
|
errors.append(f"broken transition: {t}")
|
|
if self._has_cycle(steps, transitions):
|
|
errors.append("circular dependency detected")
|
|
if errors:
|
|
raise WorkflowError("Workflow validation failed", {"errors": errors})
|
|
return True
|
|
|
|
def _has_cycle(self, steps, transitions) -> bool:
|
|
adj = {s.id: [] for s in steps}
|
|
for t in transitions:
|
|
adj.setdefault(t.from_step_id, []).append(t.to_step_id)
|
|
WHITE, GRAY, BLACK = 0, 1, 2
|
|
color = {s.id: WHITE for s in steps}
|
|
|
|
def dfs(u):
|
|
color[u] = GRAY
|
|
for v in adj.get(u, []):
|
|
if v not in color:
|
|
continue
|
|
if color[v] == WHITE and dfs(v):
|
|
return True
|
|
if color[v] == GRAY:
|
|
return True
|
|
color[u] = BLACK
|
|
return False
|
|
|
|
for s in steps:
|
|
if color[s.id] == WHITE and dfs(s.id):
|
|
return True
|
|
return False
|
|
|
|
@transaction.atomic
|
|
def publish(self, workflow, version):
|
|
self.validate_version(version)
|
|
version.status = "active"
|
|
version.published_at = timezone.now()
|
|
version.save(update_fields=["status", "published_at"])
|
|
workflow.active_version = version
|
|
workflow.status = "active"
|
|
workflow.save(update_fields=["active_version", "status"])
|
|
return version
|
|
|
|
def start_steps(self, version):
|
|
return list(version.steps.filter(is_start=True, enabled=True))
|
|
|
|
def available_next_steps(self, version, current_step, context):
|
|
"""Return the list of steps reachable from `current_step` given context."""
|
|
result = []
|
|
qs = version.transitions.filter(from_step=current_step).select_related("to_step")
|
|
for t in qs:
|
|
if evaluate_conditions(t.conditions, context):
|
|
result.append(t.to_step)
|
|
return result
|
|
|
|
def can_transition(self, version, from_step, to_step, context) -> bool:
|
|
t = version.transitions.filter(from_step=from_step, to_step=to_step).first()
|
|
if t is None:
|
|
return False
|
|
return evaluate_conditions(t.conditions, context)
|