Welcome to transitions-gui Discussions!

[aleneum]

👋 Welcome!

We’re using Discussions as a place to connect with other members of our community. We hope that you:

• Ask questions you’re wondering about.
• Share ideas.
• Engage with other community members.
• Welcome others and are open-minded. Remember that this is a community we
  build together 💪.

To get started, comment below with an introduction of yourself and tell us about what you do with this community.

[Dvzkabou]

I would like to migrate from transitions Machine to transitions-gui Webmachine. Our transitions machine is working beautifully in our flask api. Any tutorials or suggestions?
from transitions_gui import WebMachine as Machine
from transitions.extensions.states import Timeout, Tags, add_state_features
from loggingDetail import logger
from sqlModel import tblApplication, tblApplicationRental, tblLead, getDBSession, tblMinorWorkFlowStatus, tblRentalInstallDocs
import pytz
import datetime as ddt

@add_state_features(Timeout, Tags)
class EnhancedStateMachine(Machine):
pass

class ApplicationLeadStateMachine(object):
def init(self, leadID=None, appID=None):
states = [
{'name': 'new', 'timeout': 60, 'on_timeout': 'timeout_handler'},
'create', 'saved', 'siteVisiteDateRequested', 'siteVisiteDateConfirmed', 'svc', 'dropOff', 'assignSupplier',
'acceptSupplier', 'approveRejection', 'reassignSupplier', 'prepareQuote', 'validatePackage',
'installationRentalPrice', 'noticeToProceed', 'rentalQuote', 'affordabilityAssessment',
'ssegCompleted', 'cocValidation', 'loadPurchaseOrder', 'setDeliveryDate', 'assignInstaller',
'acceptInstaller', 'installationDate', 'installation', 'systemCommissioning',
'approveCOC', 'uploadInvoice', 'installationSignoff', 'financialSignoff', 'supplierRejected', 'installerRejected',
'ssegRequired'
]

    self.leadID = leadID
    self.appID = appID
    self.machine = EnhancedStateMachine(model=self, states=states, initial='new', after_state_change='update_database')

    # Define transitions
    self.machine.add_transition('create', 'new', 'create')
    self.machine.add_transition('save', '*', 'saved')
    self.machine.add_transition('siteVisiteDateRequested', '*', 'siteVisiteDateRequested')
    self.machine.add_transition('siteVisiteDateConfirmed', '*', 'siteVisiteDateConfirmed')
    self.machine.add_transition('assignInstaller', '*', 'assignInstaller')
    self.machine.add_transition('acceptInstaller', '*', 'acceptInstaller')
    self.machine.add_transition('installerRejected', '*', 'installerRejected')
    self.machine.add_transition('assignSupplier', '*', 'assignSupplier')
    self.machine.add_transition('acceptSupplier', '*', 'acceptSupplier')
    self.machine.add_transition('supplierRejected', '*', 'supplierRejected')
    self.machine.add_transition('dropOff', '*', 'dropOff')
    self.machine.add_transition('svc', '*', 'svc')
    self.machine.add_transition('prepareQuote', '*', 'prepareQuote')
    self.machine.add_transition('rentalQuote', '*', 'rentalQuote')
    self.machine.add_transition('validatePackage', '*', 'validatePackage')
    self.machine.add_transition('noticeToProceed', '*', 'noticeToProceed')
    self.machine.add_transition('ssegCompleted', '*', 'ssegCompleted')
    self.machine.add_transition('approveCOC', '*', 'approveCOC', before='reviewCOC', after='reviewComplete')
    self.machine.add_transition('ssegRequired', '*', 'ssegRequired')
    self.machine.add_transition('loadPurchaseOrder', '*', 'loadPurchaseOrder')
    self.machine.add_transition('setDeliveryDate', '*', 'setDeliveryDate')

    self.leadLevel1Status = 1
    self.leadLevel2Status = 0
    self.leadLevel3Status = 0
    self.currentRoute = 'Lead'
    self.appLevel1Status = 1
    self.appLevel2Status = 0
    self.appLevel3Status = 0

def reviewCOC(self):
    session = getDBSession()
    tempDoc = session.query(tblRentalInstallDocs)\
                    .filter(tblRentalInstallDocs.RecordID == self.appID)\
                    .filter(tblRentalInstallDocs.RecordType == 1)\
                    .first()
    if tempDoc:
        tempDoc.DocStatusID = 2
        session.commit()
    session.close()

def reviewComplete(self):
    session = getDBSession()
    tempAR = session.query(tblApplicationRental).filter(tblApplicationRental.ApplicationID == self.appID).first()
    if tempAR:
        ssegRequired = tempAR.RequireSSEG
        if ssegRequired == 1:
            self.ssegRequired()

def pre_assign_supplier(self):
    self.appLevel2Status = 2
    self.leadLevel2Status = 2

def pre_reject_supplier(self):
    self.appLevel2Status = 2
    self.leadLevel2Status = 2

def determine_level2_state(self):
    current_state = self.state
    level2_state = current_state.split('_')[1] if '_' in current_state else 'assigned'
    return level2_state

def move_to_state(self, state_suffix):
    try:
        level2_state = self.determine_level2_state()
        target_state = f'active_{level2_state}_{state_suffix}'
        status_code = self.getLevel3StatusCode(state_suffix)
        
        if hasattr(self, target_state):  # Check if the target state is valid
            self.to_state(target_state)
            if self.currentRoute == 'Application':
                self.appLevel3Status = status_code
                self.updateApplication()
            else:
                self.leadLevel3Status = status_code
                self.updateLead()
            logger.info(f"Successfully transitioned to {target_state} with status code {status_code}.")
        else:
            logger.error(f"Attempted invalid transition to {target_state} from {self.state}.")
    except Exception as e:
        logger.error(f"Error during state transition to {state_suffix}: {e}")

def getLevel3StatusCode(self, state_suffix):
    try:
        session = getDBSession()
        minorStatus = session.query(tblMinorWorkFlowStatus).filter(tblMinorWorkFlowStatus.StateDesc == state_suffix).first()
        if minorStatus:
            statusCode = minorStatus.Id
        else:
            statusCode = 99
        return statusCode
    except Exception as e:
        logger.error(f"Error updating Lead ID {self.leadID} App ID {self.appID}: {str(e)}")
        return 99
    finally:
        session.close()

def refresh_state(self):
    session = getDBSession()  
    try:
        if self.leadID:
            lead = session.query(tblLead).filter(tblLead.Id == self.leadID).one_or_none()
            if lead:
                self.leadLevel1Status = lead.LeadStatusID
                self.leadLevel2Status = lead.LeadSubStatusID
                self.leadLevel3Status = lead.MinorWorkFlowStep
                leadCurrentApp = lead.CurrentAppID
                if self.leadLevel2Status == 12 and leadCurrentApp > 0:
                    self.currentRoute = 'Application'
                    if self.appID is None:
                        self.appID = leadCurrentApp
                else:
                    self.currentRoute = 'Lead'
                    currentState = lead.CurrentState
                    installerID = lead.InstallerID
                    if installerID:
                        self.installerID = installerID
                    supplierID = lead.SupplierGroupID
                    if supplierID:
                        self.supplierID = supplierID    
                    logger.info(f'Attempt to set state to: {currentState}')
                    if currentState:
                        try:
                            self.state = currentState
                            logger.info(f"Successfully transitioned to {currentState}.")
                        except AttributeError:
                            logger.error(f"No transition method found for '{currentState}'")
                        except Exception as e:
                            logger.error(f"Error during state transition to {currentState}: {e}")
                
        if self.appID:
            application = session.query(tblApplication).filter(tblApplication.Id == self.appID).one_or_none()
            if application:
                self.currentRoute = 'Application'
                self.appLevel1Status = application.WorkflowStep
                self.appLevel2Status = application.SubWorkFlowStep
                self.appLevel3Status = application.MinorWorkFlowStep
                currentState = application.CurrentState
                applicationRental = session.query(tblApplicationRental).filter(tblApplicationRental.ApplicationID == self.appID).first()
                if applicationRental:
                    installerID = applicationRental.InstallerID
                    if installerID:
                        self.installerID = installerID
                    supplierID = applicationRental.SupplierGroupID
                    if supplierID:
                        self.supplierID = supplierID 
        logger.info(f'Current State from DB: {currentState} Current State: {self.state}')
    except Exception as e:
        logger.error(f"Error refreshing state machine for Lead ID {self.leadID}, App ID {self.appID}: {str(e)}")
    finally:
        session.close()

def updateLead(self):
    session = getDBSession()  
    try:
        lead = session.query(tblLead).filter_by(Id=self.leadID).first()
        if lead:
            lead.LeadStatusID = self.leadLevel1Status
            lead.LeadSubStatusID = self.leadLevel2Status
            lead.MinorWorkFlowStep = self.leadLevel3Status
            lead.CurrentState = self.state
            lead.DateModified = self.getCurrentTime()
            session.commit()
    except Exception as e:
        logger.error(f"Error updating Lead ID {self.leadID}: {str(e)}")
    finally:
        session.close()
      
def updateApplication(self):
    session = getDBSession()  
    try:
        application = session.query(tblApplication).filter_by(Id=self.appID).first()
        if application:
            application.WorkflowStep = self.appLevel1Status
            application.SubWorkFlowStep = self.appLevel2Status
            application.MinorWorkFlowStep = self.appLevel3Status
            application.CurrentState = self.state
            application.DateModified = self.getCurrentTime()
            session.commit()
    except Exception as e:
        logger.error(f"Error updating App ID {self.appID}: {str(e)}")
    finally:
        session.close()  
        
def update_database(self):
    logger.info(f'Current State: {self.state} for {self.currentRoute}')
    status_code = self.getLevel3StatusCode(self.state)
    if self.currentRoute == 'Application':
        self.appLevel3Status = status_code
        self.updateApplication()
    else:
        self.leadLevel3Status = status_code
        self.updateLead()
        
def getCurrentTime(self):
    sa_time = pytz.timezone('Africa/Johannesburg')
    current_time = ddt.datetime.now(sa_time)
    formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
    return formatted_time

def timeout_handler(self):
    logger.warning(f"State {self.state} timed out")
    # Handle timeout accordingly

[aleneum]

Hello @Dvzkabou,

it should be enough to switch from transitions import Machine to from transitions_gui import WebMachine as Machine. This will start a tornado web server on port 8080. If that port is already taken, you just need to pass your desired port:

from transitions_gui import WebMachine as Machine

m = Machine(..., port=9999)

FYI: A new version of transitions-gui has been released just recently.