tessl/pypi-django-fsm

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

—

Pending

Overview

Eval results

Files

Dynamic State Classes

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

Classes for dynamic state resolution allowing transition targets to be determined at runtime based on method return values or custom functions.

Capabilities

RETURN_VALUE

Uses the method's return value as the target state, enabling transitions where the destination state is determined by business logic at runtime.

class RETURN_VALUE(State):
    def __init__(self, *allowed_states):
        """
        Dynamic state that uses method return value as target state.
        
        Parameters:
        - *allowed_states: Optional tuple of allowed return values.
                          If provided, return value must be in this list.
        """
    
    def get_state(self, model, transition, result, args=[], kwargs={}):
        """
        Get target state from method return value.
        
        Parameters:
        - model: Model instance
        - transition: Transition object
        - result: Return value from transition method
        - args: Method arguments
        - kwargs: Method keyword arguments
        
        Returns:
        str: Target state
        
        Raises:
        InvalidResultState: If result not in allowed_states
        """

Usage example:

from django_fsm import FSMField, transition, RETURN_VALUE

class Task(models.Model):
    state = FSMField(default='new')
    priority = models.IntegerField(default=1)
    
    @transition(field=state, source='new', target=RETURN_VALUE())
    def categorize(self):
        """Determine state based on priority."""
        if self.priority >= 8:
            return 'urgent'
        elif self.priority >= 5:
            return 'normal'
        else:
            return 'low'

# Usage
task = Task.objects.create(priority=9)
task.categorize()
print(task.state)  # 'urgent'

RETURN_VALUE with Allowed States

Restrict the possible return values to a predefined set:

class Document(models.Model):
    state = FSMField(default='draft')
    content_score = models.FloatField(default=0.0)
    
    @transition(
        field=state, 
        source='review', 
        target=RETURN_VALUE('approved', 'rejected', 'needs_revision')
    )
    def review_content(self):
        """Review content and return appropriate state."""
        if self.content_score >= 0.8:
            return 'approved'
        elif self.content_score >= 0.5:
            return 'needs_revision'
        else:
            return 'rejected'

# This will raise InvalidResultState if method returns invalid state
try:
    doc = Document.objects.create(content_score=0.9)
    doc.review_content()  # Returns 'approved' - valid
except InvalidResultState as e:
    print(f"Invalid state returned: {e}")

GET_STATE

Uses a custom function to determine the target state, providing maximum flexibility for complex state resolution logic.

class GET_STATE(State):
    def __init__(self, func, states=None):
        """
        Dynamic state that uses custom function to determine target state.
        
        Parameters:
        - func: Function that takes (model, *args, **kwargs) and returns state
        - states: Optional tuple of allowed states for validation
        """
    
    def get_state(self, model, transition, result, args=[], kwargs={}):
        """
        Get target state using custom function.
        
        Parameters:
        - model: Model instance
        - transition: Transition object  
        - result: Return value from transition method (ignored)
        - args: Method arguments
        - kwargs: Method keyword arguments
        
        Returns:
        str: Target state from custom function
        
        Raises:
        InvalidResultState: If result not in allowed states
        """

Usage examples:

from django_fsm import GET_STATE

def determine_approval_state(order, *args, **kwargs):
    """Custom function to determine approval state."""
    if order.amount > 10000:
        return 'executive_approval'
    elif order.amount > 1000:
        return 'manager_approval'
    else:
        return 'auto_approved'

class Order(models.Model):
    state = FSMField(default='pending')
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    
    @transition(
        field=state, 
        source='pending', 
        target=GET_STATE(determine_approval_state)
    )
    def submit_for_approval(self):
        """Submit order for appropriate approval level."""
        # Business logic here
        self.submitted_at = timezone.now()

# Usage
order = Order.objects.create(amount=Decimal('15000.00'))
order.submit_for_approval()
print(order.state)  # 'executive_approval'

GET_STATE with Validation

Restrict possible states using the states parameter:

def calculate_risk_level(loan, *args, **kwargs):
    """Calculate risk-based state."""
    risk_score = loan.calculate_risk_score()
    if risk_score > 0.8:
        return 'high_risk'
    elif risk_score > 0.5:
        return 'medium_risk'
    else:
        return 'low_risk'

class LoanApplication(models.Model):
    state = FSMField(default='submitted')
    credit_score = models.IntegerField()
    income = models.DecimalField(max_digits=10, decimal_places=2)
    
    @transition(
        field=state,
        source='submitted',
        target=GET_STATE(
            calculate_risk_level, 
            states=('low_risk', 'medium_risk', 'high_risk')
        )
    )
    def assess_risk(self):
        pass
    
    def calculate_risk_score(self):
        # Complex risk calculation logic
        if self.credit_score < 600:
            return 0.9
        elif self.credit_score < 700:
            return 0.6
        else:
            return 0.3

Advanced Dynamic State Patterns

Context-Aware State Resolution

Use method arguments and context to determine states:

def determine_priority_state(task, urgency_level, department, *args, **kwargs):
    """Determine state based on method arguments and model data."""
    if urgency_level == 'critical':
        return 'immediate'
    elif department == 'security' and task.security_sensitive:
        return 'security_review'
    elif task.estimated_hours > 40:
        return 'project_planning'
    else:
        return 'standard_queue'

class Task(models.Model):
    state = FSMField(default='created')
    estimated_hours = models.IntegerField(default=1)
    security_sensitive = models.BooleanField(default=False)
    
    @transition(
        field=state,
        source='created',
        target=GET_STATE(determine_priority_state)
    )
    def prioritize(self, urgency_level, department):
        """Prioritize task based on urgency and department."""
        self.prioritized_at = timezone.now()

# Usage with arguments
task = Task.objects.create(estimated_hours=50)
task.prioritize('normal', 'security')
print(task.state)  # 'project_planning' (due to estimated_hours > 40)

Multi-Factor State Resolution

Combine multiple factors for complex state determination:

def complex_approval_state(expense, *args, **kwargs):
    """Multi-factor state determination."""
    factors = {
        'amount': expense.amount,
        'category': expense.category,
        'requestor_level': expense.requestor.level,
        'department_budget': expense.department.remaining_budget
    }
    
    # Executive approval needed
    if factors['amount'] > 50000:
        return 'executive_approval'
    
    # Department head approval
    elif factors['amount'] > 10000 or factors['category'] == 'travel':
        return 'department_head_approval'
    
    # Manager approval for mid-level amounts
    elif factors['amount'] > 1000:
        return 'manager_approval'
    
    # Auto-approve for senior staff small expenses
    elif factors['requestor_level'] >= 3 and factors['amount'] <= 500:
        return 'auto_approved'
    
    # Default to manager approval
    else:
        return 'manager_approval'

class ExpenseRequest(models.Model):
    state = FSMField(default='submitted')
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    category = models.CharField(max_length=50)
    requestor = models.ForeignKey(User, on_delete=models.CASCADE)
    department = models.ForeignKey(Department, on_delete=models.CASCADE)
    
    @transition(
        field=state,
        source='submitted',
        target=GET_STATE(complex_approval_state)
    )
    def route_approval(self):
        pass

Time-Based State Resolution

Determine states based on time factors:

from datetime import datetime, timedelta

def time_based_expiry_state(subscription, *args, **kwargs):
    """Determine state based on subscription timing."""
    now = timezone.now()
    expires_at = subscription.expires_at
    
    if expires_at <= now:
        return 'expired'
    elif expires_at <= now + timedelta(days=7):
        return 'expiring_soon'
    elif expires_at <= now + timedelta(days=30):
        return 'renewal_period'
    else:
        return 'active'

class Subscription(models.Model):
    state = FSMField(default='pending')
    expires_at = models.DateTimeField()
    
    @transition(
        field=state,
        source='pending',
        target=GET_STATE(time_based_expiry_state)
    )
    def activate(self):
        # Set expiration date
        self.expires_at = timezone.now() + timedelta(days=365)

External API State Resolution

Determine states based on external service responses:

def payment_processor_state(payment, *args, **kwargs):
    """Determine state based on external payment processor."""
    try:
        response = payment_gateway.check_status(payment.transaction_id)
        
        status_map = {
            'completed': 'paid',
            'pending': 'processing',
            'failed': 'failed',
            'refunded': 'refunded',
            'cancelled': 'cancelled'
        }
        
        return status_map.get(response.status, 'unknown')
        
    except PaymentGatewayError:
        return 'gateway_error'

class Payment(models.Model):
    state = FSMField(default='initialized')
    transaction_id = models.CharField(max_length=100)
    
    @transition(
        field=state,
        source='initialized',
        target=GET_STATE(
            payment_processor_state,
            states=('paid', 'processing', 'failed', 'refunded', 'cancelled', 'gateway_error', 'unknown')
        )
    )
    def sync_with_gateway(self):
        pass

Error Handling with Dynamic States

Invalid State Handling

Handle cases where dynamic state resolution fails:

from django_fsm import InvalidResultState

def risky_state_calculation(model, *args, **kwargs):
    """State calculation that might return invalid states."""
    external_status = get_external_status(model.external_id)
    # This might return values not in allowed_states
    return external_status

class ExternalSync(models.Model):
    state = FSMField(default='syncing')
    external_id = models.CharField(max_length=100)
    
    @transition(
        field=state,
        source='syncing',
        target=GET_STATE(
            risky_state_calculation,
            states=('completed', 'failed', 'partial')
        )
    )
    def sync_status(self):
        try:
            # Normal transition logic
            pass
        except InvalidResultState as e:
            # Handle invalid state by setting default
            self.state = 'failed'
            self.error_message = f"Invalid external status: {e}"
            raise

Fallback State Logic

Implement fallback logic when dynamic resolution fails:

def safe_state_calculation(model, *args, **kwargs):
    """State calculation with built-in fallback."""
    try:
        # Primary state calculation
        calculated_state = complex_calculation(model)
        
        # Validate against allowed states
        allowed = ('approved', 'rejected', 'pending')
        if calculated_state in allowed:
            return calculated_state
        else:
            # Fallback to safe default
            return 'pending'
            
    except Exception:
        # Error fallback
        return 'error'

Testing Dynamic States

Test dynamic state behavior thoroughly:

from django.test import TestCase
from django_fsm import InvalidResultState

class DynamicStateTests(TestCase):
    def test_return_value_state_resolution(self):
        """Test RETURN_VALUE state resolution."""
        task = Task.objects.create(priority=8)
        task.categorize()
        self.assertEqual(task.state, 'urgent')
    
    def test_get_state_function_resolution(self):
        """Test GET_STATE with custom function.""" 
        order = Order.objects.create(amount=Decimal('15000'))
        order.submit_for_approval()
        self.assertEqual(order.state, 'executive_approval')
    
    def test_invalid_state_raises_exception(self):
        """Test that invalid states raise InvalidResultState."""
        with patch('myapp.models.risky_state_calculation') as mock_calc:
            mock_calc.return_value = 'invalid_state'
            
            sync = ExternalSync.objects.create()
            with self.assertRaises(InvalidResultState):
                sync.sync_status()

Install with Tessl CLI