Files
pacnpal b508434574 Add state machine diagrams and code examples for ThrillWiki
- Created a comprehensive documentation file for state machine diagrams, detailing various states and transitions for models such as EditSubmission, ModerationReport, and Park Status.
- Included transition matrices for each state machine to clarify role requirements and guards.
- Developed a new document providing code examples for implementing state machines, including adding new state machines to models, defining custom guards, implementing callbacks, and testing state machines.
- Added examples for document approval workflows, custom guards, email notifications, and cache invalidation callbacks.
- Implemented a test suite for document workflows, covering various scenarios including approval, rejection, and transition logging.
2025-12-21 20:21:54 -05:00

12 KiB

State Machine System Documentation

Overview

ThrillWiki uses a sophisticated state machine system built on django-fsm integrated with the RichChoice system. This provides:

  • Type-safe state transitions with validation
  • Guard-based access control for transitions
  • Callback system for side effects
  • Automatic logging via django-fsm-log
  • RichChoice metadata for transition rules

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    State Machine Layer                       │
├─────────────────────────────────────────────────────────────┤
│  RichFSMField          │  StateMachineMixin                 │
│  - Choice validation   │  - Transition methods              │
│  - Metadata access     │  - State field management          │
├─────────────────────────────────────────────────────────────┤
│  Guards                │  Callbacks                          │
│  - PermissionGuard     │  - Pre-transition                  │
│  - OwnershipGuard      │  - Post-transition                 │
│  - AssignmentGuard     │  - Error handlers                  │
│  - StateGuard          │  - Notifications                   │
│  - MetadataGuard       │  - Cache invalidation              │
│  - CompositeGuard      │  - Related model updates           │
├─────────────────────────────────────────────────────────────┤
│  django-fsm            │  django-fsm-log                    │
│  - @transition         │  - StateLog model                  │
│  - can_proceed()       │  - Automatic logging               │
└─────────────────────────────────────────────────────────────┘

Key Components

RichFSMField

A custom FSM field that integrates with the RichChoice system:

from apps.core.state_machine import RichFSMField

class Ride(StateMachineMixin, TrackedModel):
    status = RichFSMField(
        choice_group="statuses",
        domain="rides",
        max_length=20,
        default="OPERATING"
    )

StateMachineMixin

Provides the base functionality for models with state machines:

from apps.core.state_machine import StateMachineMixin

class EditSubmission(StateMachineMixin, models.Model):
    state_field_name = "status"  # Required attribute
    
    # Transition methods are auto-generated from metadata
    # e.g., transition_to_approved(user=None)

Guards

Guards control who can perform transitions:

Guard Purpose
PermissionGuard Role and permission checks
OwnershipGuard Verify user owns the object
AssignmentGuard Verify user is assigned
StateGuard Validate current state
MetadataGuard Check required fields
CompositeGuard Combine guards with AND/OR

Callbacks

Callbacks execute side effects during transitions:

Callback Type When Executed
Pre-transition Before state change
Post-transition After state change
Error On transition failure
Notification Send emails/alerts
Cache Invalidate caches
Related Update related models

Models with State Machines

Moderation Domain

Model States Description
EditSubmission PENDING → APPROVED/REJECTED/ESCALATED User edit submissions
PhotoSubmission PENDING → APPROVED/REJECTED/ESCALATED Photo submissions
ModerationReport PENDING → UNDER_REVIEW → RESOLVED/DISMISSED Content reports
ModerationQueue PENDING → IN_PROGRESS → COMPLETED/CANCELLED Queue items
BulkOperation PENDING → RUNNING → COMPLETED/FAILED/CANCELLED Bulk actions

Parks Domain

Model States Description
Park OPERATING/CLOSED_TEMP/CLOSED_PERM/DEMOLISHED/RELOCATED Park lifecycle

Rides Domain

Model States Description
Ride OPERATING/CLOSED_TEMP/SBNO/CLOSING/CLOSED_PERM/DEMOLISHED/RELOCATED Ride lifecycle

Transition Metadata

RichChoice metadata defines transition behavior:

RichChoice(
    value="PENDING",
    label="Pending Review",
    metadata={
        'can_transition_to': ['APPROVED', 'REJECTED', 'ESCALATED'],
        'requires_moderator': False,
        'requires_admin_approval': False,
        'requires_assignment': False,
        'is_final': False,
        'color': 'yellow',
        'icon': 'clock'
    }
)

Metadata Fields

Field Type Description
can_transition_to List[str] Allowed target states
requires_moderator bool Requires MODERATOR role or higher
requires_admin_approval bool Requires ADMIN role or higher
requires_assignment bool Requires user to be assigned
is_final bool Terminal state (no transitions out)
zero_tolerance bool Requires SUPERUSER role
escalation_level str 'moderator', 'admin', or 'superuser'

Adding a New State Machine

Step 1: Define RichChoice Statuses

# apps/myapp/choices.py
from apps.core.choices import RichChoice, register_choices

WORKFLOW_STATUSES = [
    RichChoice(
        value="DRAFT",
        label="Draft",
        metadata={
            'can_transition_to': ['REVIEW', 'CANCELLED'],
            'requires_moderator': False,
            'is_final': False,
        }
    ),
    RichChoice(
        value="REVIEW",
        label="Under Review",
        metadata={
            'can_transition_to': ['APPROVED', 'REJECTED'],
            'requires_moderator': True,
        }
    ),
    RichChoice(
        value="APPROVED",
        label="Approved",
        metadata={
            'can_transition_to': [],
            'is_final': True,
        }
    ),
    RichChoice(
        value="REJECTED",
        label="Rejected",
        metadata={
            'can_transition_to': ['DRAFT'],  # Can resubmit
            'is_final': False,
        }
    ),
    RichChoice(
        value="CANCELLED",
        label="Cancelled",
        metadata={
            'can_transition_to': [],
            'is_final': True,
        }
    ),
]

register_choices('workflow_statuses', 'myapp', WORKFLOW_STATUSES)

Step 2: Add RichFSMField to Model

# apps/myapp/models.py
from apps.core.state_machine import RichFSMField, StateMachineMixin
from apps.core.models import TrackedModel

class Document(StateMachineMixin, TrackedModel):
    state_field_name = "status"
    
    title = models.CharField(max_length=255)
    content = models.TextField()
    
    status = RichFSMField(
        choice_group="workflow_statuses",
        domain="myapp",
        max_length=20,
        default="DRAFT"
    )
    
    # Transition methods are auto-generated:
    # - transition_to_review(user=None)
    # - transition_to_approved(user=None)
    # - transition_to_rejected(user=None)
    # - transition_to_cancelled(user=None)

Step 3: Add Wrapper Methods (Optional)

class Document(StateMachineMixin, TrackedModel):
    # ... fields ...
    
    def submit_for_review(self, user=None):
        """Submit document for review."""
        self.transition_to_review(user=user)
        self.save()
    
    def approve(self, user=None, notes=None):
        """Approve the document."""
        self.transition_to_approved(user=user)
        if notes:
            self.approval_notes = notes
        self.approved_at = timezone.now()
        self.approved_by = user
        self.save()

Step 4: Create Migration

uv run manage.py makemigrations myapp
uv run manage.py migrate

Step 5: Write Tests

# apps/myapp/tests.py
from django.test import TestCase
from django_fsm import TransitionNotAllowed

class DocumentTransitionTests(TestCase):
    def test_draft_to_review_transition(self):
        doc = Document.objects.create(title='Test', status='DRAFT')
        doc.transition_to_review(user=self.user)
        doc.save()
        self.assertEqual(doc.status, 'REVIEW')
    
    def test_approved_cannot_transition(self):
        doc = Document.objects.create(title='Test', status='APPROVED')
        with self.assertRaises(TransitionNotAllowed):
            doc.transition_to_rejected(user=self.moderator)

Guards Usage

PermissionGuard

from apps.core.state_machine.guards import PermissionGuard

# Require moderator role
guard = PermissionGuard(requires_moderator=True)

# Require admin role
guard = PermissionGuard(requires_admin=True)

# Require specific roles
guard = PermissionGuard(required_roles=['ADMIN', 'SUPERUSER'])

# Custom check
guard = PermissionGuard(
    custom_check=lambda instance, user: instance.department == user.department
)

OwnershipGuard

from apps.core.state_machine.guards import OwnershipGuard

# Default ownership check (created_by, user, submitted_by)
guard = OwnershipGuard()

# With moderator override
guard = OwnershipGuard(allow_moderator_override=True)

# Custom owner field
guard = OwnershipGuard(owner_fields=['author'])

CompositeGuard

from apps.core.state_machine.guards import CompositeGuard, PermissionGuard, OwnershipGuard

# Require moderator OR owner
guard = CompositeGuard([
    PermissionGuard(requires_moderator=True),
    OwnershipGuard()
], operator='OR')

# Require moderator AND assigned
guard = CompositeGuard([
    PermissionGuard(requires_moderator=True),
    AssignmentGuard()
], operator='AND')

Callbacks Usage

Registering Callbacks

from apps.core.state_machine.registry import state_machine_registry

# Register post-transition callback
@state_machine_registry.register_callback('myapp.Document', 'post_transition')
def on_document_approved(instance, from_state, to_state, user):
    if to_state == 'APPROVED':
        send_approval_notification(instance, user)
        invalidate_document_cache(instance)

Notification Callbacks

from apps.core.state_machine.callbacks import NotificationCallback

class ApprovalNotification(NotificationCallback):
    def execute(self, context):
        if context['to_state'] == 'APPROVED':
            send_email(
                to=context['instance'].author.email,
                template='document_approved',
                context={'document': context['instance']}
            )

Transition Logging

All transitions are automatically logged via django-fsm-log:

from django_fsm_log.models import StateLog

# Get transition history for an instance
logs = StateLog.objects.for_instance(document).order_by('timestamp')

for log in logs:
    print(f"{log.timestamp}: {log.source_state}{log.state} by {log.by}")

Testing State Machines

Use the provided test helpers:

from apps.core.state_machine.tests.helpers import (
    assert_transition_allowed,
    assert_transition_denied,
    assert_state_log_created,
    transition_and_save
)

def test_moderator_can_approve(self):
    submission = self._create_submission()
    
    # Assert transition is allowed
    assert_transition_allowed(submission, 'transition_to_approved', self.moderator)
    
    # Execute and verify
    transition_and_save(submission, 'transition_to_approved', self.moderator)
    
    # Verify log was created
    assert_state_log_created(submission, 'APPROVED', self.moderator)

Best Practices

  1. Use wrapper methods for complex transitions that need additional logic
  2. Define clear metadata in RichChoice for each state
  3. Test all transition paths including invalid ones
  4. Use CompositeGuard for complex permission requirements
  5. Log transitions for audit trails
  6. Handle callbacks atomically with the transition