Workflow Engine¶

Todo (New, Open)
  ├→ In Progress
  ├→ On Hold / Blocked
  └→ Done (Resolved)

In Progress
  ├→ Resolved
  ├→ Todo (Open)
  └→ Blocked

Done (Resolved)
  ├→ Closed
  └→ Todo (Open, if reopening needed)

Closed
  └→ (Terminal - no further transitions)

# Status to category mapping
STATUS_TO_CATEGORY = {
    'new': 'todo',
    'open': 'todo',
    'in_progress': 'in_progress',
    'on_hold': 'in_progress',
    'blocked': 'in_progress',
    'resolved': 'done',
    'closed': 'done',
}

# Transition definitions for UI
WORKFLOW_TRANSITIONS = {
    'todo': {
        'in_progress': {'label': 'Start Progress', 'style': 'primary'},
        'done': {'label': 'Resolve', 'style': 'success'},
    },
    'in_progress': {
        'todo': {'label': 'Reopen', 'style': 'warning'},
        'done': {'label': 'Resolve', 'style': 'success'},
    },
    'done': {
        'todo': {'label': 'Reopen', 'style': 'warning'},
        'done': {'label': 'Close', 'style': 'secondary'},
    },
}

TICKET_STATUS_TRANSITIONS = {
    'todo': {'in_progress', 'waiting', 'closed'},
    'in_progress': {'waiting', 'resolved', 'closed'},
    'waiting': {'in_progress', 'resolved'},
    'resolved': {'closed', 'in_progress'},
    'closed': set(),  # Terminal
}

def is_valid_transition(old_status, new_status):
    """Check if transition is allowed."""
    if old_status == new_status:
        return True
    return new_status in TICKET_STATUS_TRANSITIONS.get(old_status, set())

class TicketForm(ModelForm):
    def clean_status(self):
        new_status = self.cleaned_data.get('status')

        # Get current ticket status
        if self.instance.pk:
            current = Ticket.objects.get(pk=self.instance.pk)
            old_status = current.status

            # Check if transition is valid
            if not is_valid_transition(old_status, new_status):
                raise ValidationError(
                    f'Cannot transition from {old_status} to {new_status}'
                )

        return new_status

class Ticket(Model):
    def save(self, *args, **kwargs):
        # Set status category
        self.status_category = STATUS_TO_CATEGORY.get(self.status, 'todo')

        # If updating existing ticket, validate transition
        if self.pk:
            old = Ticket.objects.get(pk=self.pk)
            if old.status != self.status:
                if not is_valid_transition(old.status, self.status):
                    raise ValidationError('Invalid status transition')

        super().save(*args, **kwargs)

    def get_available_transitions(self):
        """Get valid next statuses."""
        category = self.status_category
        transitions = WORKFLOW_TRANSITIONS.get(category, {})
        return list(transitions.keys())

from tickets.workflows import is_valid_transition
from tickets.models import Ticket

ticket = Ticket.objects.get(pk=1)

# Check if transition is valid
if is_valid_transition(ticket.status, 'in_progress'):
    ticket.status = 'in_progress'
    ticket.save()
else:
    print('Invalid transition!')

ticket = Ticket.objects.get(pk=1)

# In template
{{ ticket.get_available_transitions }}  # ['in_progress', 'resolved']

from tickets.workflow import apply_transition

ticket = Ticket.objects.get(pk=1)

# Apply transition with additional context
apply_transition(
    ticket=ticket,
    new_status='in_progress',
    updated_by=request.user,
    comment='Starting work on this issue'
)

@receiver(post_save, sender=Ticket)
def on_ticket_status_change(sender, instance, **kwargs):
    """Triggered when ticket status changes."""

    # Automatically assign to default group
    if instance.status == 'new':
        instance.assigned_to = Group.objects.get(name='Support').users.first()
        instance.save()

    # Send notification
    if instance.status == 'resolved':
        notify_requester(instance, 'Your ticket has been resolved')

    # Archive old closed tickets
    if instance.status == 'closed':
        archive_if_old(instance, days=30)

def can_close_ticket(ticket):
    """Check if ticket can be closed."""

    # Must be resolved first
    if ticket.status != 'resolved':
        return False

    # Must have no open subtasks
    if ticket.subtasks.filter(status='open').exists():
        return False

    # Cannot close high-priority tickets without approval
    if ticket.priority == 'critical':
        return ticket.requires_review

    return True

class RecurringTaskTemplate(Model):
    name = CharField(max_length=255)
    pattern = CharField(choices=[
        ('daily', 'Daily'),
        ('weekly', 'Weekly'),
        ('monthly', 'Monthly'),
    ])
    template_ticket = ForeignKey(Ticket, on_delete=CASCADE)
    is_active = BooleanField(default=True)

@shared_task
def create_recurring_tickets():
    """Run every minute, creates tickets from active templates."""

    for template in RecurringTaskTemplate.objects.filter(is_active=True):
        if should_create_ticket(template):
            ticket = Ticket.objects.create(
                title=template.template_ticket.title,
                description=template.template_ticket.description,
                priority=template.template_ticket.priority,
            )

            RecurringTicketRun.objects.create(
                template=template,
                ticket=ticket,
                run_at=timezone.now()
            )

class Ticket(Model):
    # Timestamps
    created_at = DateTimeField(auto_now_add=True)
    updated_at = DateTimeField(auto_now=True)

    # Lifecycle milestones (v0.6.0+)
    first_responded_at = DateTimeField(null=True, blank=True)
    resolved_at = DateTimeField(null=True, blank=True)
    closed_at = DateTimeField(null=True, blank=True)

def save(self, *args, **kwargs):
    # Track first response
    if self.status in ['in_progress', 'resolved'] and not self.first_responded_at:
        self.first_responded_at = timezone.now()

    # Track resolution
    if self.status == 'resolved' and not self.resolved_at:
        self.resolved_at = timezone.now()

    # Track closure
    if self.status == 'closed' and not self.closed_at:
        self.closed_at = timezone.now()

    super().save(*args, **kwargs)

class Ticket(Model):
    # SLA targets (v0.6.0+)
    response_target = DateTimeField(null=True, blank=True)
    resolution_target = DateTimeField(null=True, blank=True)

def check_sla_breach(ticket):
    """Check if ticket has breached SLA."""
    now = timezone.now()

    # Response SLA
    if not ticket.first_responded_at and now > ticket.response_target:
        return 'Response SLA breached'

    # Resolution SLA
    if not ticket.closed_at and now > ticket.resolution_target:
        return 'Resolution SLA breached'

    return None

# ✅ Good
if ticket.status == STATUS_IN_PROGRESS:
    ...

# ❌ Bad
if ticket.status == 'in_progress':  # Magic string
    ...

def test_cannot_transition_from_closed():
    """Closed tickets should not transition further."""
    ticket = Ticket.objects.create(..., status='closed')

    with self.assertRaises(ValidationError):
        ticket.status = 'open'
        ticket.save()

@shared_task
def process_bulk_status_change(ticket_ids, new_status):
    """Bulk status change in background."""
    for ticket_id in ticket_ids:
        ticket = Ticket.objects.get(pk=ticket_id)
        if is_valid_transition(ticket.status, new_status):
            ticket.status = new_status
            ticket.save()