pytransitions example

pytransitions.py

from transitions import Machine


class Payments(models.Model):
    STATES = {
        'started': 'Started',
        'captured': 'Captured',
        'completed': 'Completed',
    }

    state = models.CharField(default='started', choices=STATES.items(), max_length=16)

    def __init__(self, *args, **kwargs):
        super(Payments, self).__init__(*args, **kwargs)
        
        # Initialize the state machine
        self.machine = Machine(
            model=self,  # the object on which the library attaches trigger functions
            states=self.STATES.keys(),  # states in the state machine
            initial='started',  # initial state of the machine
            after_state_change='save',  # method to call after state change, this will call self.save()
        )

        # Lets add transitions to our state machine, we can now call self.capture() to perform
        # the transition and capture a payment
        self.machine.add_transition(
            trigger='capture', 
            source='started', 
            destination='captured',
            before='contact_payment_gateway',
        )
        # We can now call self.complete() to perform the transition and complete a payment
        self.machine.add_transition(
            trigger='complete', 
            source='captured', 
            destination='completed',
            after='notify_user',
        )

pytransitions_retry.py

class Payments(models.Model):
    STATES = {
        'started': 'Started',
        'captured': 'Captured',
        'completed': 'Completed',
        'incomplete': 'Incomplete',  # new incomplete state for us
    }

    def __init__(self, *args, **kwargs):
        super(Payments, self).__init__(*args, **kwargs)
        
        # Initialize the state machine
        self.machine = Machine(
            model=self,
            states=self.STATES.keys(),
            initial='started',
            after_state_change='save',
        )

        self.machine.add_transition(
            trigger='capture', 
            source='started', 
            destination='captured',
            # we check only enter captured state if the payment is captured
            # otherwise we go to incomplete, as you can see in the next
            # transition
            conditions=['has_captured_payment',],
        )

        self.machine.add_transition(
            trigger='capture', 
            source='started', 
            destination='incomplete',
        )

        self.machine.add_transition(
            trigger='retry', 
            source='incomplete', 
            destination='completed',
            # similarly, we only go from incomplete to complete if the payment
            # is captured, otherwise we stay as incomplete
            conditions=['has_captured_payment',],
        )

        self.machine.add_transition(
            trigger='retry', 
            source='incomplete', 
            destination='incomplete',
        )

        self.machine.add_transition(
            trigger='complete', 
            source='captured', 
            destination='completed',
        )
        
        # we notify the user whenever we enter “complete” state, this
        # this calls self.notify_user()
        self.machine.on_enter_complete('notify_user')

    def has_captured_payment(self):
        try:
            self.contact_payment_gateway()
        except PaymentGatewayException:
            return False
        return True

solution1.py

class Payment(models.Model):
    is_started = models.BooleanField(default=True)
    is_captured = models.BooleanField(default=False)
    is_completed = models.BooleanField(default=False)

    def can_capture(self):
        return self.is_started and not self.is_captured and not self.is_completed

    def can_complete(self):
        return self.is_captured

    def contact_payment_gateway(self):
        """ does payment gateway magic """

    def notify_user(self):
        """ notifies the user that their payment is complete """

    def capture(self):
        if self.can_capture():
            self.contact_payment_gateway()
            self.is_captured = True
            self.save()
        else:
            raise ValueError()

    def complete(self):
        if self.can_complete():
            self.is_completed = True
            self.save()
            self.notify_user()
        else:
            raise ValueError()

solution2.py

class Payment(models.Model):
    STATE_CHOICES = (
        ('started', 'Started'),
        ('captured', 'Captured'),
        ('completed', 'Completed'),
    )

    state = models.CharField(choices=STATE_CHOICES, default='started', max_length=16)

    def can_capture(self):
        return self.state == 'started'

    def can_complete(self):
        return self.state == 'captured'

    def capture(self):
        if self.can_capture():
            self.contact_payment_gateway()
            self.state = 'captured'
            self.save()
        else:
            raise ValueError()

    def complete(self):
        if self.can_complete():
            self.state = 'completed'
            self.save()
            self.notify_user()
        else:
            raise ValueError()

solution2_retry.py

class Payment(models.Model):
    STATE_CHOICES = (
        ('started', 'Started'),
        ('captured', 'Captured'),
        ('completed', 'Completed'),
        # we introduce a new "incomplete" state
        ('incomplete', 'Incomplete'),
    )

    def can_complete(self):
        # We can complete both captured and incomplete payments
        return self.state in ('captured', 'incomplete')

    def can_retry(self):
        # We can only retry incomplete payments
        return self.state == 'incomplete'

    def capture(self):
        if self.can_capture():
            try:
                self.contact_payment_gateway()
            except PaymentGatewayException:
                # We can mark a payment as incomplete if we couldn't 
                # capture it from the payment gateway
                self.state = 'incomplete'
                self.save()
                return

            self.state = 'captured'
            self.save()
        else:
            raise ValueError()

    def retry(self):
        # We can call retry() to complete the payment if we it is incomplete
        if self.can_retry():
            self.complete()