tessl/pypi-django-fsm

Django friendly finite state machine support for models through FSMField and the @transition decorator.

—

Pending

Overview

Eval results

Files

Signal System and Events

Name: tessl/pypi-django-fsm
Author: tessl

Django signals for hooking into state transition lifecycle events, enabling custom logic before and after transitions.

Capabilities

pre_transition Signal

Fired before state transition execution, allowing you to implement validation, logging, or preparatory actions.

# From django_fsm.signals
pre_transition = Signal()

Signal Arguments:

sender: Model class
instance: Model instance undergoing transition
name: Name of the transition method
field: FSM field instance
source: Current state before transition
target: Target state after transition
method_args: Arguments passed to transition method
method_kwargs: Keyword arguments passed to transition method

Usage example:

from django_fsm.signals import pre_transition
from django.dispatch import receiver
import logging

@receiver(pre_transition)
def log_state_transition(sender, instance, name, source, target, **kwargs):
    """Log all state transitions before they occur."""
    logging.info(
        f"About to transition {sender.__name__} {instance.pk} "
        f"from {source} to {target} via {name}"
    )

# For specific models only
@receiver(pre_transition, sender=Order)
def validate_order_transition(sender, instance, name, source, target, **kwargs):
    """Validate order transitions before they happen."""
    if name == 'ship' and not instance.payment_confirmed:
        raise ValueError("Cannot ship unconfirmed payment")

post_transition Signal

Fired after successful state transition, enabling cleanup, notifications, or cascade operations.

# From django_fsm.signals  
post_transition = Signal()

Signal Arguments:

sender: Model class
instance: Model instance that transitioned
name: Name of the transition method
field: FSM field instance
source: Previous state before transition
target: New state after transition
method_args: Arguments passed to transition method
method_kwargs: Keyword arguments passed to transition method
exception: Exception instance (only present if transition failed with on_error state)

Usage example:

from django_fsm.signals import post_transition
from django.dispatch import receiver

@receiver(post_transition)
def handle_state_change(sender, instance, name, source, target, **kwargs):
    """Handle any state change across all models."""
    print(f"State changed: {sender.__name__} {instance.pk} -> {target}")

@receiver(post_transition, sender=Order)
def order_state_changed(sender, instance, name, source, target, **kwargs):
    """Handle order-specific state changes."""
    if target == 'shipped':
        # Send shipping notification
        send_shipping_notification(instance)
    elif target == 'cancelled':
        # Process refund
        process_refund(instance)

Signal Implementation Patterns

Audit Trail Implementation

Create comprehensive audit logs using signals:

from django_fsm.signals import pre_transition, post_transition
from django.contrib.auth import get_user_model
import json

class StateTransitionAudit(models.Model):
    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField()
    content_object = GenericForeignKey('content_type', 'object_id')
    
    transition_name = models.CharField(max_length=100)
    source_state = models.CharField(max_length=100)
    target_state = models.CharField(max_length=100)
    user = models.ForeignKey(get_user_model(), null=True, on_delete=models.SET_NULL)
    timestamp = models.DateTimeField(auto_now_add=True)
    metadata = models.JSONField(default=dict)

@receiver(post_transition)
def create_audit_record(sender, instance, name, source, target, **kwargs):
    """Create audit record for every state transition."""
    StateTransitionAudit.objects.create(
        content_object=instance,
        transition_name=name,
        source_state=source,
        target_state=target,
        user=getattr(instance, '_current_user', None),
        metadata={
            'method_args': kwargs.get('method_args', []),
            'method_kwargs': kwargs.get('method_kwargs', {}),
            'has_exception': 'exception' in kwargs
        }
    )

Notification System

Implement notifications based on state changes:

from django_fsm.signals import post_transition
from django.core.mail import send_mail

@receiver(post_transition, sender=Order)
def send_order_notifications(sender, instance, name, source, target, **kwargs):
    """Send notifications based on order state changes."""
    
    notification_map = {
        'confirmed': {
            'subject': 'Order Confirmed',
            'template': 'order_confirmed.html',
            'recipients': [instance.customer.email]
        },
        'shipped': {
            'subject': 'Order Shipped', 
            'template': 'order_shipped.html',
            'recipients': [instance.customer.email]
        },
        'delivered': {
            'subject': 'Order Delivered',
            'template': 'order_delivered.html', 
            'recipients': [instance.customer.email, instance.sales_rep.email]
        }
    }
    
    if target in notification_map:
        config = notification_map[target]
        send_notification(instance, config)

def send_notification(order, config):
    """Send email notification using template."""
    from django.template.loader import render_to_string
    
    message = render_to_string(config['template'], {'order': order})
    send_mail(
        subject=config['subject'],
        message=message,
        from_email='noreply@example.com',
        recipient_list=config['recipients']
    )

Cascade State Changes

Trigger related object state changes:

@receiver(post_transition, sender=Order)
def cascade_order_state_changes(sender, instance, name, source, target, **kwargs):
    """Cascade state changes to related objects."""
    
    if target == 'cancelled':
        # Cancel all order items
        for item in instance.orderitem_set.all():
            if hasattr(item, 'cancel') and can_proceed(item.cancel):
                item.cancel()
                item.save()
        
        # Release reserved inventory
        for item in instance.orderitem_set.all():
            item.product.release_inventory(item.quantity)
    
    elif target == 'shipped':
        # Update inventory for shipped items
        for item in instance.orderitem_set.all():
            item.product.reduce_inventory(item.quantity)

Error Handling in Signals

Handle exceptions in signal receivers:

from django_fsm.signals import post_transition
import logging

logger = logging.getLogger('fsm_signals')

@receiver(post_transition)
def safe_notification_handler(sender, instance, name, source, target, **kwargs):
    """Send notifications with error handling."""
    try:
        if target == 'published':
            send_publication_notifications(instance)
    except Exception as e:
        logger.error(
            f"Failed to send notification for {sender.__name__} {instance.pk}: {e}",
            exc_info=True
        )
        # Don't re-raise - notification failure shouldn't break the transition

Conditional Signal Processing

Process signals based on specific conditions:

@receiver(post_transition)
def conditional_processing(sender, instance, name, source, target, **kwargs):
    """Process signals based on conditions."""
    
    # Only process during business hours
    from datetime import datetime
    current_hour = datetime.now().hour
    if not (9 <= current_hour <= 17):
        return
    
    # Only process specific transitions
    if name not in ['publish', 'approve', 'reject']:
        return
    
    # Only for specific models
    if sender not in [Article, Document, BlogPost]:
        return
    
    # Process the signal
    handle_business_hours_transition(instance, name, target)

Signal-Based Metrics

Collect metrics using signals:

from django.core.cache import cache
from django_fsm.signals import post_transition

@receiver(post_transition)
def collect_transition_metrics(sender, instance, name, source, target, **kwargs):
    """Collect metrics about state transitions."""
    
    # Count transitions by type
    cache_key = f"transition_count:{sender.__name__}:{name}"
    cache.set(cache_key, cache.get(cache_key, 0) + 1, timeout=3600)
    
    # Track state distribution
    state_key = f"state_count:{sender.__name__}:{target}"
    cache.set(state_key, cache.get(state_key, 0) + 1, timeout=3600)
    
    # Track transition timing
    timing_key = f"transition_time:{sender.__name__}:{name}"
    transition_time = timezone.now()
    cache.set(timing_key, transition_time.isoformat(), timeout=3600)

Testing Signal Handlers

Test signal-based functionality:

from django.test import TestCase
from django_fsm.signals import post_transition
from unittest.mock import patch, Mock

class SignalTests(TestCase):
    def test_notification_sent_on_state_change(self):
        """Test that notifications are sent when order is shipped."""
        
        with patch('myapp.signals.send_mail') as mock_send_mail:
            order = Order.objects.create(state='confirmed')
            order.ship()
            order.save()
            
            # Verify notification was sent
            mock_send_mail.assert_called_once()
            args, kwargs = mock_send_mail.call_args
            self.assertIn('Order Shipped', args[0])  # subject
            self.assertIn(order.customer.email, kwargs['recipient_list'])
    
    def test_signal_handler_exception_handling(self):
        """Test that signal handler exceptions don't break transitions."""
        
        with patch('myapp.signals.send_publication_notifications') as mock_notify:
            mock_notify.side_effect = Exception("Notification failed")
            
            article = Article.objects.create(state='draft')
            article.publish()
            article.save()
            
            # Transition should succeed despite notification failure
            self.assertEqual(article.state, 'published')

Performance Considerations

Optimize signal handlers for performance:

from django_fsm.signals import post_transition
from django.core.cache import cache

@receiver(post_transition)
def optimized_signal_handler(sender, instance, name, source, target, **kwargs):
    """Optimized signal handler with caching and batching."""
    
    # Use caching to avoid repeated database queries
    cache_key = f"config:{sender.__name__}:{name}"
    config = cache.get(cache_key)
    if config is None:
        config = load_transition_config(sender, name)
        cache.set(cache_key, config, timeout=300)
    
    # Batch operations when possible
    if target == 'processed':
        # Collect IDs for batch processing
        batch_key = f"batch_process:{sender.__name__}"
        ids = cache.get(batch_key, [])
        ids.append(instance.pk)
        cache.set(batch_key, ids, timeout=60)
        
        # Process in batches of 10
        if len(ids) >= 10:
            process_batch(sender, ids)
            cache.delete(batch_key)

Install with Tessl CLI