0
# Workflows and Publishing
1

2
Content approval workflows with task-based moderation, publishing controls, and comprehensive revision management. Wagtail's workflow system enables collaborative content creation with customizable approval processes.
3

4
## Capabilities
5

6
### Workflow Management
7

8
Core workflow system for managing content approval processes.
9

10
```python { .api }
11
class Workflow(models.Model):
12
    """
13
    Defines a content approval workflow with multiple tasks.
14
    
15
    Properties:
16
        name (str): Workflow name
17
        active (bool): Whether workflow is active and can be started
18
    """
19
    name: str
20
    active: bool
21
    
22
    def start(self, page, user):
23
        """
24
        Start this workflow for a page.
25
        
26
        Parameters:
27
            page (Page): Page to start workflow for
28
            user (User): User starting the workflow
29
            
30
        Returns:
31
            WorkflowState: The started workflow state
32
        """
33
    
34
    def deactivate(self):
35
        """Deactivate this workflow, preventing new starts."""
36
    
37
    def all_pages(self):
38
        """Get all pages this workflow applies to."""
39
    
40
    def get_tasks(self):
41
        """Get all tasks in this workflow in order."""
42

43
class WorkflowState(models.Model):
44
    """
45
    Represents a workflow instance for a specific page.
46
    
47
    Properties:
48
        page (Page): Page this workflow state applies to
49
        workflow (Workflow): The workflow being executed
50
        status (str): Current status ('in_progress', 'approved', 'cancelled', 'rejected')
51
        created_at (datetime): When workflow was started
52
        requested_by (User): User who requested the workflow
53
        current_task_state (TaskState): Currently active task
54
    """
55
    page: Page
56
    workflow: Workflow
57
    status: str
58
    created_at: datetime
59
    requested_by: User
60
    current_task_state: TaskState
61
    
62
    def cancel(self, user=None):
63
        """
64
        Cancel this workflow.
65
        
66
        Parameters:
67
            user (User): User cancelling the workflow
68
        """
69
    
70
    def resume(self, user=None):
71
        """
72
        Resume a paused workflow.
73
        
74
        Parameters:
75
            user (User): User resuming the workflow
76
        """
77
    
78
    def finish(self, user=None):
79
        """
80
        Mark workflow as complete.
81
        
82
        Parameters:
83
            user (User): User finishing the workflow
84
        """
85
    
86
    def copy_approved_task_states_to_revision(self, revision):
87
        """Copy approved task states to a revision for publishing."""
88
    
89
    def get_next_task(self):
90
        """Get the next task to be executed in this workflow."""
91
    
92
    def all_tasks_with_states(self):
93
        """Get all tasks in workflow with their current states."""
94
```
95

96
### Task System
97

98
Individual task components that make up workflow steps.
99

100
```python { .api }
101
class Task(models.Model):
102
    """
103
    Base class for workflow tasks.
104
    
105
    Properties:
106
        name (str): Task name
107
        active (bool): Whether task is active
108
    """
109
    name: str
110
    active: bool
111
    
112
    def start(self, workflow_state, user=None):
113
        """
114
        Start this task within a workflow.
115
        
116
        Parameters:
117
            workflow_state (WorkflowState): The workflow state
118
            user (User): User starting the task
119
            
120
        Returns:
121
            TaskState: The started task state
122
        """
123
    
124
    def on_action(self, task_state, user, action_name, **kwargs):
125
        """
126
        Handle task action (approve, reject, etc.).
127
        
128
        Parameters:
129
            task_state (TaskState): Current task state
130
            user (User): User performing the action
131
            action_name (str): Action being performed
132
            **kwargs: Additional action parameters
133
            
134
        Returns:
135
            tuple: (next_task_state, workflow_complete)
136
        """
137
    
138
    def user_can_access_editor(self, page, user):
139
        """
140
        Check if user can access the page editor during this task.
141
        
142
        Parameters:
143
            page (Page): Page being edited
144
            user (User): User to check permissions for
145
            
146
        Returns:
147
            bool: Whether user can access editor
148
        """
149
    
150
    def get_actions(self, page, user):
151
        """
152
        Get available actions for user on this task.
153
        
154
        Parameters:
155
            page (Page): Page the task applies to
156
            user (User): User to check actions for
157
            
158
        Returns:
159
            list: Available action names
160
        """
161

162
class GroupApprovalTask(Task):
163
    """
164
    Task that requires approval from members of specific groups.
165
    
166
    Properties:
167
        groups (QuerySet): Groups whose members can approve this task
168
    """
169
    groups: QuerySet
170
    
171
    def user_can_approve(self, user):
172
        """Check if user is in one of the approval groups."""
173
    
174
    def get_actors(self):
175
        """Get all users who can act on this task."""
176

177
class TaskState(models.Model):
178
    """
179
    Represents the state of a task within a workflow instance.
180
    
181
    Properties:
182
        workflow_state (WorkflowState): Parent workflow state
183
        task (Task): The task this state represents
184
        status (str): Current status ('in_progress', 'approved', 'rejected', 'cancelled', 'skipped')
185
        started_at (datetime): When task was started
186
        finished_at (datetime): When task was completed
187
        finished_by (User): User who completed the task
188
        comment (str): Comment provided with task completion
189
    """
190
    workflow_state: WorkflowState
191
    task: Task
192
    status: str
193
    started_at: datetime
194
    finished_at: datetime
195
    finished_by: User
196
    comment: str
197
    
198
    def approve(self, user=None, comment=''):
199
        """
200
        Approve this task.
201
        
202
        Parameters:
203
            user (User): User approving the task
204
            comment (str): Optional approval comment
205
        """
206
    
207
    def reject(self, user=None, comment=''):
208
        """
209
        Reject this task.
210
        
211
        Parameters:
212
            user (User): User rejecting the task
213
            comment (str): Optional rejection comment
214
        """
215
    
216
    def cancel(self, user=None):
217
        """
218
        Cancel this task.
219
        
220
        Parameters:
221
            user (User): User cancelling the task
222
        """
223
    
224
    def copy_to_revision(self, revision):
225
        """Copy this task state to a revision for publishing."""
226
```
227

228
### Publishing System
229

230
Models and utilities for managing content publishing and revisions.
231

232
```python { .api }
233
class PublishingPanel:
234
    """
235
    Admin panel for publishing controls and workflow status.
236
    
237
    Displays workflow status, publishing history, and available actions.
238
    """
239
    def __init__(self, **kwargs):
240
        """Initialize publishing panel."""
241

242
class ScheduledPublishing:
243
    """
244
    Functionality for scheduling page publication at specific times.
245
    """
246
    @staticmethod
247
    def schedule_page_publication(page, go_live_at, user=None):
248
        """
249
        Schedule a page for publication.
250
        
251
        Parameters:
252
            page (Page): Page to schedule
253
            go_live_at (datetime): When to publish the page
254
            user (User): User scheduling the publication
255
        """
256

257
def publish_scheduled_pages():
258
    """
259
    Management command function to publish pages scheduled for publication.
260
    
261
    Should be run regularly via cron job or similar scheduling system.
262
    """
263

264
class WorkflowContentObject:
265
    """
266
    Mixin for models that can participate in workflows.
267
    
268
    Provides workflow-related methods for any content model.
269
    """
270
    def get_current_workflow_state(self):
271
        """Get the current workflow state for this object."""
272
    
273
    def has_workflow_in_progress(self):
274
        """Check if this object has a workflow in progress."""
275
```
276

277
### Workflow Permissions
278

279
Permission classes and utilities for workflow access control.
280

281
```python { .api }
282
class TaskPermissionTester:
283
    """
284
    Utility for testing task permissions for users.
285
    """
286
    def __init__(self, user, task):
287
        """
288
        Initialize permission tester.
289
        
290
        Parameters:
291
            user (User): User to test permissions for
292
            task (Task): Task to test permissions on
293
        """
294
    
295
    def can_approve(self):
296
        """Check if user can approve this task."""
297
    
298
    def can_reject(self):
299
        """Check if user can reject this task."""
300
    
301
    def can_cancel(self):
302
        """Check if user can cancel this task."""
303

304
class WorkflowPermissionTester:
305
    """
306
    Utility for testing workflow permissions for users.
307
    """
308
    def __init__(self, user, workflow):
309
        """
310
        Initialize workflow permission tester.
311
        
312
        Parameters:
313
            user (User): User to test permissions for
314
            workflow (Workflow): Workflow to test permissions on
315
        """
316
    
317
    def can_start(self):
318
        """Check if user can start this workflow."""
319
    
320
    def can_cancel(self):
321
        """Check if user can cancel workflow instances."""
322
```
323

324
## Usage Examples
325

326
### Creating Custom Workflows
327

328
```python
329
from wagtail.models import Workflow, Task, GroupApprovalTask
330
from django.contrib.auth.models import Group
331

332
# Create user groups for different roles
333
editors_group = Group.objects.create(name='Editors')
334
managers_group = Group.objects.create(name='Managers') 
335
legal_group = Group.objects.create(name='Legal Team')
336

337
# Create tasks
338
content_review_task = GroupApprovalTask.objects.create(
339
    name='Content Review',
340
    active=True
341
)
342
content_review_task.groups.add(editors_group)
343

344
legal_review_task = GroupApprovalTask.objects.create(
345
    name='Legal Review', 
346
    active=True
347
)
348
legal_review_task.groups.add(legal_group)
349

350
manager_approval_task = GroupApprovalTask.objects.create(
351
    name='Manager Approval',
352
    active=True
353
)
354
manager_approval_task.groups.add(managers_group)
355

356
# Create workflow
357
blog_workflow = Workflow.objects.create(
358
    name='Blog Post Approval',
359
    active=True
360
)
361

362
# Add tasks to workflow in order
363
blog_workflow.workflow_tasks.create(task=content_review_task, sort_order=0)
364
blog_workflow.workflow_tasks.create(task=legal_review_task, sort_order=1)  
365
blog_workflow.workflow_tasks.create(task=manager_approval_task, sort_order=2)
366

367
# Assign workflow to page types
368
from wagtail.models import WorkflowPage
369

370
WorkflowPage.objects.create(
371
    workflow=blog_workflow,
372
    page=BlogPage.get_content_type()
373
)
374
```
375

376
### Starting and Managing Workflows
377

378
```python
379
from wagtail.models import Page, Workflow
380

381
# Get a page and workflow
382
page = BlogPage.objects.get(slug='my-blog-post')
383
workflow = Workflow.objects.get(name='Blog Post Approval')
384

385
# Start workflow
386
user = request.user
387
workflow_state = workflow.start(page, user)
388

389
print(f"Started workflow: {workflow_state.status}")
390
print(f"Current task: {workflow_state.current_task_state.task.name}")
391

392
# Check workflow progress
393
if workflow_state.current_task_state:
394
    current_task = workflow_state.current_task_state.task
395
    print(f"Waiting for: {current_task.name}")
396
    
397
    # Get users who can approve current task
398
    if hasattr(current_task, 'groups'):
399
        approvers = []
400
        for group in current_task.groups.all():
401
            approvers.extend(group.user_set.all())
402
        print(f"Can be approved by: {[u.username for u in approvers]}")
403

404
# Check if workflow is complete
405
if workflow_state.status == 'approved':
406
    print("Workflow approved - ready to publish")
407
elif workflow_state.status == 'rejected':
408
    print("Workflow rejected - needs revision")
409
```
410

411
### Task Actions and Approvals
412

413
```python
414
from wagtail.models import TaskState
415

416
# Get current task state
417
page = BlogPage.objects.get(slug='my-blog-post')
418
workflow_state = page.current_workflow_state
419
task_state = workflow_state.current_task_state
420

421
# Approve task
422
if request.user in task_state.task.get_actors():
423
    task_state.approve(
424
        user=request.user,
425
        comment="Content looks good, approved for next stage"
426
    )
427
    print("Task approved")
428

429
# Reject task
430
def reject_task(task_state, user, reason):
431
    task_state.reject(
432
        user=user,
433
        comment=f"Rejected: {reason}"
434
    )
435
    # Workflow will return to previous state or end
436

437
# Cancel entire workflow
438
def cancel_workflow(workflow_state, user, reason):
439
    workflow_state.cancel(user=user)
440
    # Add comment via separate mechanism if needed
441
    
442
# Resume cancelled workflow
443
def resume_workflow(workflow_state, user):
444
    if workflow_state.status == 'cancelled':
445
        workflow_state.resume(user=user)
446
```
447

448
### Custom Task Types
449

450
```python
451
from wagtail.models import Task, TaskState
452
from django.contrib.auth.models import User
453

454
class CustomApprovalTask(Task):
455
    """Custom task requiring approval from specific users."""
456
    
457
    required_approvers = models.ManyToManyField(User, blank=True)
458
    minimum_approvals = models.PositiveIntegerField(default=1)
459
    
460
    def start(self, workflow_state, user=None):
461
        """Start task and notify required approvers."""
462
        task_state = super().start(workflow_state, user)
463
        
464
        # Send notifications to required approvers
465
        for approver in self.required_approvers.all():
466
            send_approval_notification(approver, task_state)
467
        
468
        return task_state
469
    
470
    def on_action(self, task_state, user, action_name, **kwargs):
471
        """Handle custom approval logic."""
472
        if action_name == 'approve':
473
            # Track individual approvals
474
            approval, created = TaskApproval.objects.get_or_create(
475
                task_state=task_state,
476
                user=user,
477
                defaults={'approved': True}
478
            )
479
            
480
            # Check if enough approvals received
481
            approval_count = TaskApproval.objects.filter(
482
                task_state=task_state,
483
                approved=True
484
            ).count()
485
            
486
            if approval_count >= self.minimum_approvals:
487
                return task_state.approve(user=user), False
488
            else:
489
                return task_state, False  # Continue waiting for more approvals
490
                
491
        return super().on_action(task_state, user, action_name, **kwargs)
492
    
493
    def get_actors(self):
494
        """Get users who can act on this task."""
495
        return self.required_approvers.all()
496

497
class TaskApproval(models.Model):
498
    """Track individual approvals for custom tasks."""
499
    task_state = models.ForeignKey(TaskState, on_delete=models.CASCADE)
500
    user = models.ForeignKey(User, on_delete=models.CASCADE)
501
    approved = models.BooleanField(default=False)
502
    created_at = models.DateTimeField(auto_now_add=True)
503
```
504

505
### Workflow Views and Templates
506

507
```python
508
from django.shortcuts import render, get_object_or_404, redirect
509
from django.contrib.auth.decorators import login_required
510
from wagtail.models import Page, TaskState
511

512
@login_required
513
def workflow_dashboard(request):
514
    """Dashboard showing user's workflow tasks."""
515
    user = request.user
516
    
517
    # Get tasks assigned to user
518
    pending_tasks = TaskState.objects.filter(
519
        status='in_progress',
520
        task__groups__user=user
521
    ).select_related('workflow_state__page', 'task')
522
    
523
    # Get pages user has submitted for workflow
524
    submitted_pages = Page.objects.filter(
525
        workflowstate__requested_by=user,
526
        workflowstate__status='in_progress'
527
    )
528
    
529
    return render(request, 'workflows/dashboard.html', {
530
        'pending_tasks': pending_tasks,
531
        'submitted_pages': submitted_pages,
532
    })
533

534
@login_required  
535
def approve_task(request, task_state_id):
536
    """Approve a workflow task."""
537
    task_state = get_object_or_404(TaskState, id=task_state_id)
538
    
539
    # Check permissions
540
    if request.user not in task_state.task.get_actors():
541
        return redirect('workflow_dashboard')
542
    
543
    if request.method == 'POST':
544
        comment = request.POST.get('comment', '')
545
        task_state.approve(user=request.user, comment=comment)
546
        return redirect('workflow_dashboard')
547
    
548
    return render(request, 'workflows/approve_task.html', {
549
        'task_state': task_state,
550
        'page': task_state.workflow_state.page,
551
    })
552

553
def workflow_history(request, page_id):
554
    """Show workflow history for a page."""
555
    page = get_object_or_404(Page, id=page_id)
556
    
557
    # Get all workflow states for this page
558
    workflow_states = page.workflowstate_set.all().order_by('-created_at')
559
    
560
    history = []
561
    for state in workflow_states:
562
        task_states = state.task_states.all().order_by('started_at')
563
        history.append({
564
            'workflow_state': state,
565
            'task_states': task_states,
566
        })
567
    
568
    return render(request, 'workflows/history.html', {
569
        'page': page,
570
        'history': history,
571
    })
572
```
573

574
### Workflow Signals and Hooks
575

576
```python
577
from wagtail import hooks
578
from wagtail.signals import (
579
    workflow_submitted, workflow_approved, workflow_rejected, workflow_cancelled,
580
    task_submitted, task_approved, task_rejected, task_cancelled
581
)
582
from django.dispatch import receiver
583

584
@receiver(workflow_submitted)
585
def on_workflow_submitted(sender, instance, user, **kwargs):
586
    """Handle workflow submission."""
587
    workflow_state = instance
588
    page = workflow_state.page
589
    
590
    print(f"Workflow submitted for page: {page.title}")
591
    
592
    # Send notification to workflow participants
593
    notify_workflow_participants(workflow_state)
594

595
@receiver(task_approved)
596
def on_task_approved(sender, instance, user, **kwargs):
597
    """Handle task approval."""
598
    task_state = instance
599
    page = task_state.workflow_state.page
600
    
601
    print(f"Task '{task_state.task.name}' approved for page: {page.title}")
602
    
603
    # Log approval action
604
    log_workflow_action(task_state, 'approved', user)
605

606
@receiver(workflow_approved) 
607
def on_workflow_approved(sender, instance, user, **kwargs):
608
    """Handle workflow completion."""
609
    workflow_state = instance
610
    page = workflow_state.page
611
    
612
    print(f"Workflow approved for page: {page.title}")
613
    
614
    # Auto-publish if configured
615
    if should_auto_publish(page):
616
        page.save_revision().publish()
617

618
@hooks.register('construct_page_action_menu')
619
def add_workflow_actions(menu_items, request, context):
620
    """Add workflow actions to page action menu."""
621
    page = context['page']
622
    user = request.user
623
    
624
    # Add submit for workflow action
625
    if not page.has_workflow_in_progress():
626
        available_workflows = get_workflows_for_page(page)
627
        for workflow in available_workflows:
628
            menu_items.append({
629
                'url': f'/admin/pages/{page.id}/workflow/{workflow.id}/start/',
630
                'label': f'Submit to {workflow.name}',
631
                'attrs': {'class': 'workflow-submit'},
632
            })
633

634
def notify_workflow_participants(workflow_state):
635
    """Send notifications to workflow participants."""
636
    current_task = workflow_state.current_task_state
637
    if current_task:
638
        actors = current_task.task.get_actors()
639
        for actor in actors:
640
            send_task_notification(actor, current_task)
641

642
def send_task_notification(user, task_state):
643
    """Send notification about pending task."""
644
    # Implementation would send email, push notification, etc.
645
    pass
646

647
def log_workflow_action(task_state, action, user):
648
    """Log workflow actions for audit trail."""
649
    WorkflowLog.objects.create(
650
        task_state=task_state,
651
        action=action,
652
        user=user,
653
        timestamp=timezone.now()
654
    )
655
```
656

657
### Scheduled Publishing
658

659
```python
660
from django.utils import timezone
661
from datetime import timedelta
662
from wagtail.models import Revision
663

664
def schedule_page_publication(page, publish_datetime, user):
665
    """Schedule a page for future publication."""
666
    # Create revision for scheduled publication
667
    revision = page.save_revision(
668
        user=user,
669
        submitted_for_moderation=False,
670
        approved_go_live_at=publish_datetime
671
    )
672
    
673
    print(f"Page '{page.title}' scheduled for publication at {publish_datetime}")
674
    return revision
675

676
def publish_scheduled_pages():
677
    """Management command function to publish scheduled pages."""
678
    now = timezone.now()
679
    
680
    # Find revisions scheduled for publication  
681
    scheduled_revisions = Revision.objects.filter(
682
        approved_go_live_at__lte=now,
683
        approved_go_live_at__isnull=False
684
    ).exclude(
685
        content_object__live=True,
686
        content_object__has_unpublished_changes=False
687
    )
688
    
689
    for revision in scheduled_revisions:
690
        try:
691
            revision.publish()
692
            print(f"Published scheduled page: {revision.content_object.title}")
693
        except Exception as e:
694
            print(f"Failed to publish {revision.content_object.title}: {e}")
695

696
# Example usage in management command
697
from django.core.management.base import BaseCommand
698

699
class Command(BaseCommand):
700
    """Management command for publishing scheduled pages."""
701
    
702
    def handle(self, *args, **options):
703
        publish_scheduled_pages()
704
        self.stdout.write("Finished publishing scheduled pages")
705

706
# Schedule via admin interface
707
def admin_schedule_publication(request, page):
708
    """Admin view for scheduling publication."""
709
    if request.method == 'POST':
710
        publish_at = request.POST.get('publish_at')
711
        if publish_at:
712
            publish_datetime = timezone.datetime.fromisoformat(publish_at)
713
            schedule_page_publication(page, publish_datetime, request.user)
714
            return redirect('wagtailadmin_pages:edit', page.id)
715
    
716
    return render(request, 'admin/schedule_publication.html', {
717
        'page': page,
718
    })
719
```