Created
January 27, 2018 19:50
-
-
Save dhruvbaldawa/dc1088724b358187ff448f7fc5dcfd55 to your computer and use it in GitHub Desktop.
pytransitions example
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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', | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment