tessl/pypi-wagtail

A Django content management system with a user-friendly interface and powerful features for building websites and applications.

Overview

Eval results

Files

System Integration

Name: tessl/pypi-wagtail
Author: tessl

Signals, hooks, and utilities for customizing Wagtail behavior, integrating with external systems, and extending functionality. These tools provide powerful customization capabilities for advanced Wagtail implementations.

Capabilities

Hooks System

Registration and execution system for extending Wagtail's functionality at key integration points.

def register(hook_name, fn=None, order=0):
    """
    Register a function to be called at a specific hook point.
    
    Parameters:
        hook_name (str): Name of the hook to register for
        fn (callable): Function to register (optional if used as decorator)
        order (int): Order for hook execution (lower numbers run first)
    
    Usage:
        @hooks.register('construct_main_menu')
        def add_menu_item(request, menu_items):
            menu_items.append(MenuItem('Custom', '/custom/', icon_name='cog'))
    """

def get_hooks(hook_name):
    """
    Get all registered functions for a specific hook.
    
    Parameters:
        hook_name (str): Name of the hook
        
    Returns:
        list: List of registered hook functions
    """

# Available hook names
HOOK_NAMES = [
    'construct_main_menu',           # Modify admin main menu
    'construct_page_listing_buttons', # Add page listing action buttons
    'construct_page_action_menu',    # Add page action menu items
    'before_serve_page',            # Modify page serving process
    'after_create_page',            # Action after page creation
    'after_edit_page',              # Action after page editing
    'after_delete_page',            # Action after page deletion
    'construct_document_chooser_queryset',  # Filter document chooser
    'construct_image_chooser_queryset',     # Filter image chooser
    'register_admin_urls',          # Register custom admin URLs
    'construct_settings_menu',      # Modify settings menu
    'register_admin_viewset',       # Register admin viewsets
    'construct_homepage_panels',    # Add homepage dashboard panels
    'insert_global_admin_css',      # Add global admin CSS
    'insert_global_admin_js',       # Add global admin JavaScript
    'construct_whitelister_element_rules',  # Modify rich text whitelist
    'construct_document_embed_handler',     # Custom document embedding
    'construct_image_embed_handler',        # Custom image embedding
    'before_bulk_action',           # Before bulk operations
    'after_bulk_action',            # After bulk operations
]

Signals System

Django signals for responding to Wagtail events and lifecycle changes.

# Page lifecycle signals
page_published = Signal()
"""
Sent when a page is published.

Arguments:
    sender (Model): Page model class
    instance (Page): Page instance that was published
    revision (Revision): Revision that was published
"""

page_unpublished = Signal()
"""
Sent when a page is unpublished.

Arguments:
    sender (Model): Page model class
    instance (Page): Page instance that was unpublished
"""

page_slug_changed = Signal()
"""
Sent when a page slug changes.

Arguments:
    sender (Model): Page model class
    instance (Page): Page instance with changed slug
    old_slug (str): Previous slug value
"""

pre_page_move = Signal()
"""
Sent before a page is moved.

Arguments:
    sender (Model): Page model class
    instance (Page): Page being moved
    target (Page): New parent page
    pos (str): Position relative to target
"""

post_page_move = Signal()
"""
Sent after a page is moved.

Arguments:
    sender (Model): Page model class
    instance (Page): Page that was moved
    target (Page): New parent page
    pos (str): Position relative to target
"""

# Workflow signals
workflow_submitted = Signal()
"""Sent when a workflow is submitted for approval."""

workflow_approved = Signal()
"""Sent when a workflow is approved."""

workflow_rejected = Signal()
"""Sent when a workflow is rejected."""

workflow_cancelled = Signal()
"""Sent when a workflow is cancelled."""

task_submitted = Signal()
"""Sent when an individual task is submitted."""

task_approved = Signal()
"""Sent when an individual task is approved."""

task_rejected = Signal()
"""Sent when an individual task is rejected."""

task_cancelled = Signal()
"""Sent when an individual task is cancelled."""

# Generic model signals
published = Signal()
"""Sent when any model with publishing capability is published."""

unpublished = Signal()
"""Sent when any model with publishing capability is unpublished."""

copy_for_translation_done = Signal()
"""Sent when content is copied for translation."""

Management Commands

Command-line utilities for maintenance, deployment, and administration tasks.

def publish_scheduled_pages():
    """
    Management command: Publish pages scheduled for publication.
    
    Usage: python manage.py publish_scheduled_pages
    
    Finds pages with scheduled publication times and publishes them.
    Should be run regularly via cron job.
    """

def fixtree():
    """
    Management command: Fix tree structure corruption.
    
    Usage: python manage.py fixtree
    
    Repairs corruption in the page tree structure (MP_Node hierarchy).
    """

def move_pages():
    """
    Management command: Bulk move pages to different parents.
    
    Usage: python manage.py move_pages
    
    Interactive command for moving multiple pages at once.
    """

def purge_revisions():
    """
    Management command: Clean up old page revisions.
    
    Usage: python manage.py purge_revisions [--days=30]
    
    Removes old revisions to free up database space.
    """

def set_url_paths():
    """
    Management command: Rebuild URL path cache for all pages.
    
    Usage: python manage.py set_url_paths
    
    Regenerates url_path field for all pages in the tree.
    """

def update_index():
    """
    Management command: Update search index.
    
    Usage: python manage.py update_index [app.Model]
    
    Rebuilds search index for all or specific models.
    """

def clear_index():
    """
    Management command: Clear search index.
    
    Usage: python manage.py clear_index
    
    Removes all entries from the search index.
    """

Permission System

Advanced permission checking and policy management for fine-grained access control.

class BasePermissionPolicy:
    """
    Base class for permission policies.
    
    Provides framework for checking user permissions on models.
    """
    model: Model
    
    def user_has_permission(self, user, action):
        """Check if user has permission for an action."""
    
    def user_has_permission_for_instance(self, user, action, instance):
        """Check if user has permission for action on specific instance."""
    
    def instances_user_has_permission_for(self, user, action):
        """Get queryset of instances user has permission for."""

page_permission_policy = BasePermissionPolicy()
"""Permission policy for page operations."""

collection_permission_policy = BasePermissionPolicy()
"""Permission policy for collection operations."""

task_permission_policy = BasePermissionPolicy()
"""Permission policy for workflow task operations."""

workflow_permission_policy = BasePermissionPolicy()
"""Permission policy for workflow operations."""

class UserPagePermissionsProxy:
    """
    Proxy object for checking user permissions on pages.
    
    Provides convenient interface for permission checking.
    """
    def __init__(self, user):
        """Initialize proxy for specific user."""
    
    def for_page(self, page):
        """Get page-specific permission checker."""
    
    def can_edit_pages(self):
        """Check if user can edit any pages."""
    
    def can_publish_pages(self):
        """Check if user can publish any pages."""
    
    def explorable_pages(self):
        """Get pages user can explore."""
    
    def editable_pages(self):
        """Get pages user can edit."""
    
    def publishable_pages(self):
        """Get pages user can publish."""

class PagePermissionTester:
    """
    Tests permissions for a specific user and page combination.
    """
    def __init__(self, user_perms, page):
        """
        Initialize permission tester.
        
        Parameters:
            user_perms (UserPagePermissionsProxy): User permissions proxy
            page (Page): Page to test permissions for  
        """
    
    def can_edit(self):
        """Check if user can edit this page."""
    
    def can_delete(self):
        """Check if user can delete this page."""
    
    def can_publish(self):
        """Check if user can publish this page."""
    
    def can_unpublish(self):
        """Check if user can unpublish this page."""
    
    def can_move(self):
        """Check if user can move this page."""
    
    def can_copy(self):
        """Check if user can copy this page."""
    
    def can_lock(self):
        """Check if user can lock this page."""
    
    def can_unlock(self):
        """Check if user can unlock this page."""

Utility Functions

Helper functions for common Wagtail operations and integrations.

def get_version():
    """
    Get formatted Wagtail version string.
    
    Returns:
        str: Version string (e.g., "4.1.0")
    """

def get_semver_version():
    """
    Get semver-compatible Wagtail version.
    
    Returns:
        str: Semantic version string
    """

def get_base_cache_key():
    """
    Generate base cache key for Wagtail content.
    
    Returns:
        str: Base cache key
    """

def get_site_for_hostname(hostname, port=80):
    """
    Get Site object for hostname and port.
    
    Parameters:
        hostname (str): Site hostname
        port (int): Port number
        
    Returns:
        Site: Matching site or None
    """

def get_page_models():
    """
    Get all registered page model classes.
    
    Returns:
        list: List of page model classes
    """

def get_streamfield_names(page_class):
    """
    Get names of StreamField fields on a page class.
    
    Parameters:
        page_class (Model): Page model class
        
    Returns:
        list: List of StreamField field names
    """

Usage Examples

Using Hooks for Customization

from wagtail import hooks
from wagtail.admin.menu import MenuItem
from django.urls import path, reverse
from django.http import HttpResponse

@hooks.register('construct_main_menu')
def add_custom_menu_item(request, menu_items):
    """Add custom menu item to admin interface."""
    menu_items.append(
        MenuItem(
            'Reports',
            reverse('custom_reports'),
            icon_name='doc-full-inverse',
            order=300
        )
    )

@hooks.register('register_admin_urls')
def register_custom_admin_urls():
    """Register custom admin URLs."""
    return [
        path('reports/', custom_reports_view, name='custom_reports'),
        path('analytics/', analytics_view, name='analytics'),
    ]

def custom_reports_view(request):
    """Custom reports view."""
    return HttpResponse('<h1>Custom Reports</h1>')

@hooks.register('construct_page_listing_buttons')
def page_listing_buttons(page, page_perms, is_parent=False):
    """Add custom buttons to page listing."""
    if page_perms.can_edit():
        yield {
            'url': f'/admin/custom-action/{page.id}/',
            'label': 'Custom Action',
            'classname': 'button-small',
        }

@hooks.register('before_serve_page')
def check_access_permissions(page, request, serve_args, serve_kwargs):
    """Check custom access permissions before serving page."""
    if not user_has_custom_access(request.user, page):
        raise PermissionDenied("Custom access required")

@hooks.register('after_edit_page')
def notify_on_page_edit(request, page):
    """Send notification when page is edited."""
    send_notification(f"Page '{page.title}' was edited by {request.user}")

@hooks.register('construct_image_chooser_queryset')
def filter_images_by_collection(images, request):
    """Filter image chooser by user's allowed collections."""
    if not request.user.is_superuser:
        allowed_collections = get_user_collections(request.user)
        images = images.filter(collection__in=allowed_collections)
    return images

Signal Handlers for Integration

from wagtail.signals import page_published, page_unpublished, workflow_approved
from django.dispatch import receiver
from django.core.cache import cache
import logging

logger = logging.getLogger(__name__)

@receiver(page_published)
def handle_page_published(sender, instance, revision, **kwargs):
    """Handle page publication for external integration."""
    logger.info(f"Page published: {instance.title}")
    
    # Clear relevant caches
    cache.delete_many([
        f'page_content_{instance.id}',
        'homepage_content',
        'navigation_menu'
    ])
    
    # Send to external systems
    send_to_cdn(instance)
    update_search_index(instance)
    notify_subscribers(instance)

@receiver(page_unpublished)
def handle_page_unpublished(sender, instance, **kwargs):
    """Handle page unpublication."""
    logger.info(f"Page unpublished: {instance.title}")
    
    # Remove from external systems
    remove_from_cdn(instance)
    remove_from_search_index(instance)

@receiver(workflow_approved)
def handle_workflow_approved(sender, instance, user, **kwargs):
    """Handle workflow approval."""
    workflow_state = instance
    page = workflow_state.page
    
    logger.info(f"Workflow approved for page: {page.title}")
    
    # Auto-publish if configured
    if should_auto_publish(page):
        revision = page.get_latest_revision()
        revision.publish()
    
    # Send notifications
    notify_stakeholders(page, 'approved', user)

def send_to_cdn(page):
    """Send page content to CDN."""
    # Implementation for CDN integration
    pass

def update_search_index(page):
    """Update external search index."""
    # Implementation for search service
    pass

def notify_subscribers(page):
    """Notify content subscribers."""
    # Implementation for notifications
    pass

Custom Management Commands

# management/commands/sync_content.py
from django.core.management.base import BaseCommand
from wagtail.models import Page
import logging

class Command(BaseCommand):
    """Custom management command for content synchronization."""
    help = 'Synchronize content with external systems'
    
    def add_arguments(self, parser):
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='Show what would be done without making changes'
        )
        parser.add_argument(
            '--page-type',
            type=str,
            help='Sync only specific page type'
        )
    
    def handle(self, *args, **options):
        dry_run = options['dry_run']
        page_type = options.get('page_type')
        
        self.stdout.write('Starting content sync...')
        
        # Get pages to sync
        pages = Page.objects.live()
        if page_type:
            pages = pages.filter(content_type__model=page_type)
        
        for page in pages:
            if dry_run:
                self.stdout.write(f'Would sync: {page.title}')
            else:
                try:
                    sync_page_to_external_system(page)
                    self.stdout.write(
                        self.style.SUCCESS(f'Synced: {page.title}')
                    )
                except Exception as e:
                    self.stdout.write(
                        self.style.ERROR(f'Failed to sync {page.title}: {e}')
                    )
        
        self.stdout.write(self.style.SUCCESS('Content sync completed'))

def sync_page_to_external_system(page):
    """Sync individual page to external system."""
    # Implementation for external sync
    pass

# Usage: python manage.py sync_content --dry-run --page-type=blog_page

Advanced Permission Integration

from wagtail.permissions import page_permission_policy
from wagtail.models import UserPagePermissionsProxy
from django.contrib.auth.models import User, Group

def setup_custom_permissions():
    """Set up custom permission structure."""
    
    # Create specialized groups
    content_editors = Group.objects.get_or_create(name='Content Editors')[0]
    blog_editors = Group.objects.get_or_create(name='Blog Editors')[0]
    managers = Group.objects.get_or_create(name='Content Managers')[0]
    
    # Create users with specific roles
    editor = User.objects.create_user('editor', 'editor@example.com')
    editor.groups.add(content_editors)
    
    blog_editor = User.objects.create_user('blog_editor', 'blog@example.com')
    blog_editor.groups.add(blog_editors)
    
    manager = User.objects.create_user('manager', 'manager@example.com')
    manager.groups.add(managers)

def check_user_permissions(user, page):
    """Check what a user can do with a specific page."""
    user_perms = UserPagePermissionsProxy(user)
    page_perms = user_perms.for_page(page)
    
    permissions = {
        'can_edit': page_perms.can_edit(),
        'can_delete': page_perms.can_delete(),
        'can_publish': page_perms.can_publish(),
        'can_move': page_perms.can_move(),
        'can_copy': page_perms.can_copy(),
        'can_lock': page_perms.can_lock(),
    }
    
    return permissions

def get_user_pages(user, action='edit'):
    """Get pages user has permission for specific action."""
    user_perms = UserPagePermissionsProxy(user)
    
    if action == 'edit':
        return user_perms.editable_pages()
    elif action == 'publish':
        return user_perms.publishable_pages()
    elif action == 'explore':
        return user_perms.explorable_pages()
    else:
        return Page.objects.none()

# Custom permission view
from django.shortcuts import render
from django.contrib.auth.decorators import login_required

@login_required
def user_dashboard(request):
    """Dashboard showing user's permissions and available pages."""
    user = request.user
    user_perms = UserPagePermissionsProxy(user)
    
    context = {
        'can_edit_pages': user_perms.can_edit_pages(),
        'can_publish_pages': user_perms.can_publish_pages(),
        'editable_pages': user_perms.editable_pages()[:10],
        'publishable_pages': user_perms.publishable_pages()[:10],
    }
    
    return render(request, 'admin/user_dashboard.html', context)

Cache Integration

from django.core.cache import cache
from wagtail.utils.cache import get_base_cache_key
from wagtail.signals import page_published, page_unpublished
from django.dispatch import receiver

def get_page_cache_key(page, user=None):
    """Generate cache key for page content."""
    base_key = get_base_cache_key()
    user_key = f"user_{user.id}" if user else "anonymous"
    return f"{base_key}_page_{page.id}_{user_key}"

def cache_page_content(page, content, user=None, timeout=3600):
    """Cache page content with automatic invalidation."""
    cache_key = get_page_cache_key(page, user)
    cache.set(cache_key, content, timeout)

def get_cached_page_content(page, user=None):
    """Get cached page content."""
    cache_key = get_page_cache_key(page, user)
    return cache.get(cache_key)

@receiver(page_published)
@receiver(page_unpublished)
def invalidate_page_cache(sender, instance, **kwargs):
    """Invalidate page cache when content changes."""
    # Clear specific page cache
    cache.delete_many([
        get_page_cache_key(instance),
        get_page_cache_key(instance, user=None),
    ])
    
    # Clear related caches
    cache.delete_many([
        'navigation_menu',
        'homepage_content',
        f'page_children_{instance.get_parent().id}',
    ])

# Usage in views
def cached_page_view(request, page):
    """Page view with caching."""
    cached_content = get_cached_page_content(page, request.user)
    
    if cached_content is None:
        # Generate content
        content = render_page_content(page, request)
        cache_page_content(page, content, request.user)
        return content
    else:
        return cached_content

Install with Tessl CLI