🔍 Code Extractor

Base class for workflow cycles (ReviewCycle and ApprovalCycle) that manages workflow lifecycle, status tracking, and common properties for document review and approval processes.

Purpose

WorkflowCycleBase provides a foundation for implementing workflow cycles in a document management system. It handles common workflow operations including initialization with unique identifiers, status management (pending, in progress, completed, etc.), timestamp tracking (creation, start, completion), and lifecycle methods (start, cancel, complete). The class uses an internal dictionary (_data) to store all state and tracks modifications for persistence. It's designed to be subclassed by ReviewCycle and ApprovalCycle which implement specific workflow behaviors.

Source Code

class WorkflowCycleBase:
    """Base class for workflow cycles (ReviewCycle and ApprovalCycle)."""
    
    def __init__(self, data=None):
        """Initialize with optional data dictionary."""
        self._data = data or {}
        self._modified = False
        
        # Set default values if not provided
        if 'UID' not in self._data:
            self._data['UID'] = str(uuid.uuid4())
            self._modified = True
            
        if 'created_at' not in self._data:
            self._data['created_at'] = datetime.now().isoformat()
            self._modified = True
            
        if 'status' not in self._data:
            self._data['status'] = WorkflowStatus.PENDING
            self._modified = True
    
    @property
    def uid(self) -> str:
        """Get the UID."""
        return self._data.get('UID', '')
    
    @property
    def title(self) -> str:
        """Get the title."""
        return self._data.get('title', '')
    
    @title.setter
    def title(self, value: str) -> None:
        """Set the title."""
        self._data['title'] = value
        self._modified = True
    
    @property
    def description(self) -> str:
        """Get the description."""
        return self._data.get('description', '')
    
    @description.setter
    def description(self, value: str) -> None:
        """Set the description."""
        self._data['description'] = value
        self._modified = True
    
    @property
    def status(self) -> str:
        """Get the status."""
        return self._data.get('status', WorkflowStatus.DRAFT)
    
    @status.setter
    def status(self, value: str) -> None:
        """Set the status."""
        self._data['status'] = value
        self._modified = True
    
    @property
    def created_at(self) -> str:
        """Get the creation timestamp."""
        return self._data.get('created_at', '')
    
    @property
    def updated_at(self) -> str:
        """Get the update timestamp."""
        return self._data.get('updated_at', self.created_at)
    
    @property
    def started_at(self) -> Optional[str]:
        """Get the start timestamp."""
        return self._data.get('started_at')
    
    @started_at.setter
    def started_at(self, value: Optional[str]) -> None:
        """Set the start timestamp."""
        self._data['started_at'] = value
        self._modified = True
    
    @property
    def completed_at(self) -> Optional[str]:
        """Get the completion timestamp."""
        return self._data.get('completed_at')
    
    @completed_at.setter
    def completed_at(self, value: Optional[str]) -> None:
        """Set the completion timestamp."""
        self._data['completed_at'] = value
        self._modified = True
    
    @property
    def due_date(self) -> Optional[str]:
        """Get the due date."""
        return self._data.get('due_date')
    
    @due_date.setter
    def due_date(self, value: Optional[str]) -> None:
        """Set the due date."""
        self._data['due_date'] = value
        self._modified = True
    
    @property
    def metadata(self) -> Dict[str, Any]:
        """Get the metadata."""
        return self._data.get('metadata', {})
    
    @property
    def workflow_type(self) -> str:
        """Get the workflow type."""
        return self._data.get('workflow_type', '')
        
    @workflow_type.setter
    def workflow_type(self, value: str) -> None:
        """Set the workflow type."""
        self._data['workflow_type'] = value
        self._modified = True
    
    @property
    def sequential(self) -> bool:
        """Whether this workflow cycle uses sequential processing."""
        return self._data.get('sequential', False)
    
    @property
    def decision(self) -> Optional[str]:
        """Get the final decision for this workflow cycle."""
        return self._data.get('decision')
    
    @decision.setter
    def decision(self, value: Optional[str]) -> None:
        """
        Set the decision for this workflow cycle.
        For ReviewCycle, valid values are in settings.REVIEW_DECISIONS
        For ApprovalCycle, valid values are in settings.APPROVAL_DECISIONS
        """
        valid_decisions = []
        if isinstance(self, type) and hasattr(self, "__name__"):
            if self.__name__ == "ReviewCycle":
                valid_decisions = settings.REVIEW_DECISIONS
            elif self.__name__ == "ApprovalCycle":
                valid_decisions = settings.APPROVAL_DECISIONS
        
        if value and valid_decisions and value not in valid_decisions:
            logger.warning(f"Invalid decision value: {value}")
            return
        
        self._data['decision'] = value
        self._modified = True
    
    @sequential.setter
    def sequential(self, value: bool) -> None:
        """Set whether this workflow uses sequential processing."""
        self._data['sequential'] = value
        self._modified = True
    
    @property
    def current_sequence(self) -> int:
        """Get the current sequence number for sequential workflows."""
        return self._data.get('current_sequence', 0)
    
    @current_sequence.setter
    def current_sequence(self, value: int) -> None:
        """Set the current sequence number."""
        self._data['current_sequence'] = value
        self._modified = True

    @workflow_type.setter
    def workflow_type(self, value: str) -> None:
        """Set the workflow type."""
        self._data['workflow_type'] = value
        self._modified = True
    
    @metadata.setter
    def metadata(self, value: Dict[str, Any]) -> None:
        """Set the metadata."""
        self._data['metadata'] = value
        self._modified = True
    
    def start_cycle(self) -> bool:
        """Start the workflow cycle."""
        if self.status != WorkflowStatus.PENDING:
            return False
        
        self.status = WorkflowStatus.IN_PROGRESS
        self.started_at = datetime.now().isoformat()
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        return self.save()
    
    def cancel_cycle(self) -> bool:
        """Cancel the workflow cycle."""
        if self.status in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED, 
                           WorkflowStatus.REJECTED, WorkflowStatus.CANCELED]:
            return False
        
        self.status = WorkflowStatus.CANCELED
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        return self.save()
    
    def complete_cycle(self, decision: str = None) -> bool:
        """
        Complete the workflow cycle with optional decision.
        
        Args:
            decision: The final decision (approval, rejection, etc.)
            
        Returns:
            Boolean indicating success
        """
        # Check if already completed
        if self.is_complete():
            return False
            
        # Set decision if provided
        if decision:
            self.decision = decision
            
        # Update status based on decision
        if decision == "REJECTED":
            self.status = WorkflowStatus.REJECTED
        elif self.status not in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED]:
            if hasattr(self, "__class__") and hasattr(self.__class__, "__name__"):
                if self.__class__.__name__ == "ReviewCycle":
                    self.status = WorkflowStatus.COMPLETED
                elif self.__class__.__name__ == "ApprovalCycle":
                    self.status = WorkflowStatus.APPROVED
                    
        # Set completion time
        self.completed_at = datetime.now().isoformat()
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        
        return self.save()
    
    def is_overdue(self) -> bool:
        """Whether workflow cycle is overdue."""
        return (
            self.due_date is not None 
            and datetime.fromisoformat(self.due_date) < datetime.now() 
            and not self.is_complete()
        )
    
    def is_complete(self) -> bool:
        """Check if the workflow cycle is complete."""
        return self.status in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED]
    
    def update_status(self) -> None:
        """Update the cycle status based on assignment statuses."""
        # To be implemented in subclasses
        pass
    
    def get_progress(self) -> Dict[str, int]:
        """Get progress statistics for the cycle."""
        # To be implemented in subclasses
        return {
            "total": 0,
            "pending": 0,
            "in_progress": 0,
            "completed": 0,
            "rejected": 0,
            "skipped": 0,
        }
    
    def save(self) -> bool:
        """Save changes to database."""
        # To be implemented in subclasses
        raise NotImplementedError("Subclasses must implement save method")

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

data: Optional dictionary containing initial workflow cycle data. If not provided, an empty dictionary is used. The dictionary can contain keys like 'UID', 'created_at', 'status', 'title', 'description', 'workflow_type', 'sequential', 'due_date', 'metadata', etc. If 'UID', 'created_at', or 'status' are missing, they are automatically generated with default values (UUID, current timestamp, and PENDING status respectively).

Return Value

Instantiation returns a WorkflowCycleBase object (or subclass instance) with initialized state. Key methods return: start_cycle/cancel_cycle/complete_cycle return bool indicating success; is_overdue/is_complete return bool; get_progress returns Dict[str, int] with progress statistics; save raises NotImplementedError (must be implemented by subclasses).

Class Interface

Methods

Attributes

Name	Type	Description	Scope
_data	Dict[str, Any]	Internal dictionary storing all workflow cycle data including UID, timestamps, status, and custom fields	instance
_modified	bool	Flag tracking whether the workflow cycle data has been modified since last save	instance

Dependencies

• typing
• datetime
• uuid
• logging
• CDocs.db.db_operations
• CDocs.db.schema_manager
• CDocs.config.settings
• CDocs.models.user_extensions
• CDocs.models.document

Required Imports

from typing import List, Optional, Dict, Any, Union, Type
from datetime import datetime
import uuid
import logging
from CDocs.db import db_operations as db
from CDocs.db.schema_manager import NodeLabels, RelTypes
from CDocs.config import settings
from CDocs.models.user_extensions import DocUser
from CDocs.models.document import DocumentVersion

Usage Example

# Note: WorkflowCycleBase is abstract and should be subclassed
# Example with hypothetical ReviewCycle subclass:

from workflow_cycle_base import WorkflowCycleBase
from datetime import datetime

# Create a new workflow cycle
cycle = WorkflowCycleBase()
print(f"Created cycle with UID: {cycle.uid}")

# Set properties
cycle.title = "Document Review"
cycle.description = "Review technical documentation"
cycle.workflow_type = "review"
cycle.sequential = True
cycle.due_date = "2024-12-31T23:59:59"

# Start the cycle
if cycle.start_cycle():
    print(f"Cycle started at: {cycle.started_at}")
    print(f"Status: {cycle.status}")

# Check progress
progress = cycle.get_progress()
print(f"Progress: {progress}")

# Check if overdue
if cycle.is_overdue():
    print("Cycle is overdue!")

# Complete the cycle
if cycle.complete_cycle(decision="APPROVED"):
    print(f"Cycle completed at: {cycle.completed_at}")
    print(f"Final decision: {cycle.decision}")

# Or cancel if needed
# cycle.cancel_cycle()

# Access metadata
metadata = cycle.metadata
metadata['reviewer_count'] = 5
cycle.metadata = metadata

Best Practices

• Always subclass WorkflowCycleBase rather than using it directly - it's designed as an abstract base class
• Implement the save() method in subclasses to persist changes to the database
• Implement update_status() and get_progress() methods in subclasses for specific workflow logic
• Check the return value of lifecycle methods (start_cycle, cancel_cycle, complete_cycle) to ensure operations succeeded
• Use the _modified flag to track changes and optimize database writes
• Set appropriate decision values that match settings.REVIEW_DECISIONS or settings.APPROVAL_DECISIONS
• Always check is_complete() before attempting to modify a completed workflow
• Use properties rather than direct _data dictionary access to ensure modification tracking
• Handle the NotImplementedError from save() by implementing it in subclasses
• For sequential workflows, manage current_sequence appropriately to track progress
• Validate status transitions to prevent invalid state changes
• Use ISO format strings for all timestamp properties (created_at, started_at, completed_at, due_date)
• The class automatically generates UID, created_at, and status if not provided in initialization data

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class WorkflowControllerBase 81.4% similar

  Abstract base class that defines the interface for workflow controllers managing document review and approval processes.

  From: /tf/active/vicechatdev/CDocs single class/controllers/workflow_controller_base.py

• class ReviewControllerBase 75.6% similar

  Abstract base controller class for managing review workflow processes, providing core functionality for handling review cycles, comments, and completion checks.

  From: /tf/active/vicechatdev/CDocs single class/controllers/workflow_controller_base.py

• class AssignmentBase 73.7% similar

  Base class for managing assignment lifecycle in a document review/approval workflow system, tracking status, timestamps, user assignments, and decisions.

  From: /tf/active/vicechatdev/CDocs single class/models/workflow_base.py

• class ApprovalControllerBase 72.1% similar

  Abstract base controller class for managing approval workflow processes, providing a template for approval cycle operations and workflow template management.

  From: /tf/active/vicechatdev/CDocs single class/controllers/workflow_controller_base.py

• function create_review_cycle 70.5% similar

  Creates a new review cycle for a controlled document, assigning reviewers with optional sequential workflow, custom instructions, and approval requirements.

  From: /tf/active/vicechatdev/CDocs/controllers/review_controller.py

class WorkflowCycleBase:
    """Base class for workflow cycles (ReviewCycle and ApprovalCycle)."""
    
    def __init__(self, data=None):
        """Initialize with optional data dictionary."""
        self._data = data or {}
        self._modified = False
        
        # Set default values if not provided
        if 'UID' not in self._data:
            self._data['UID'] = str(uuid.uuid4())
            self._modified = True
            
        if 'created_at' not in self._data:
            self._data['created_at'] = datetime.now().isoformat()
            self._modified = True
            
        if 'status' not in self._data:
            self._data['status'] = WorkflowStatus.PENDING
            self._modified = True
    
    @property
    def uid(self) -> str:
        """Get the UID."""
        return self._data.get('UID', '')
    
    @property
    def title(self) -> str:
        """Get the title."""
        return self._data.get('title', '')
    
    @title.setter
    def title(self, value: str) -> None:
        """Set the title."""
        self._data['title'] = value
        self._modified = True
    
    @property
    def description(self) -> str:
        """Get the description."""
        return self._data.get('description', '')
    
    @description.setter
    def description(self, value: str) -> None:
        """Set the description."""
        self._data['description'] = value
        self._modified = True
    
    @property
    def status(self) -> str:
        """Get the status."""
        return self._data.get('status', WorkflowStatus.DRAFT)
    
    @status.setter
    def status(self, value: str) -> None:
        """Set the status."""
        self._data['status'] = value
        self._modified = True
    
    @property
    def created_at(self) -> str:
        """Get the creation timestamp."""
        return self._data.get('created_at', '')
    
    @property
    def updated_at(self) -> str:
        """Get the update timestamp."""
        return self._data.get('updated_at', self.created_at)
    
    @property
    def started_at(self) -> Optional[str]:
        """Get the start timestamp."""
        return self._data.get('started_at')
    
    @started_at.setter
    def started_at(self, value: Optional[str]) -> None:
        """Set the start timestamp."""
        self._data['started_at'] = value
        self._modified = True
    
    @property
    def completed_at(self) -> Optional[str]:
        """Get the completion timestamp."""
        return self._data.get('completed_at')
    
    @completed_at.setter
    def completed_at(self, value: Optional[str]) -> None:
        """Set the completion timestamp."""
        self._data['completed_at'] = value
        self._modified = True
    
    @property
    def due_date(self) -> Optional[str]:
        """Get the due date."""
        return self._data.get('due_date')
    
    @due_date.setter
    def due_date(self, value: Optional[str]) -> None:
        """Set the due date."""
        self._data['due_date'] = value
        self._modified = True
    
    @property
    def metadata(self) -> Dict[str, Any]:
        """Get the metadata."""
        return self._data.get('metadata', {})
    
    @property
    def workflow_type(self) -> str:
        """Get the workflow type."""
        return self._data.get('workflow_type', '')
        
    @workflow_type.setter
    def workflow_type(self, value: str) -> None:
        """Set the workflow type."""
        self._data['workflow_type'] = value
        self._modified = True
    
    @property
    def sequential(self) -> bool:
        """Whether this workflow cycle uses sequential processing."""
        return self._data.get('sequential', False)
    
    @property
    def decision(self) -> Optional[str]:
        """Get the final decision for this workflow cycle."""
        return self._data.get('decision')
    
    @decision.setter
    def decision(self, value: Optional[str]) -> None:
        """
        Set the decision for this workflow cycle.
        For ReviewCycle, valid values are in settings.REVIEW_DECISIONS
        For ApprovalCycle, valid values are in settings.APPROVAL_DECISIONS
        """
        valid_decisions = []
        if isinstance(self, type) and hasattr(self, "__name__"):
            if self.__name__ == "ReviewCycle":
                valid_decisions = settings.REVIEW_DECISIONS
            elif self.__name__ == "ApprovalCycle":
                valid_decisions = settings.APPROVAL_DECISIONS
        
        if value and valid_decisions and value not in valid_decisions:
            logger.warning(f"Invalid decision value: {value}")
            return
        
        self._data['decision'] = value
        self._modified = True
    
    @sequential.setter
    def sequential(self, value: bool) -> None:
        """Set whether this workflow uses sequential processing."""
        self._data['sequential'] = value
        self._modified = True
    
    @property
    def current_sequence(self) -> int:
        """Get the current sequence number for sequential workflows."""
        return self._data.get('current_sequence', 0)
    
    @current_sequence.setter
    def current_sequence(self, value: int) -> None:
        """Set the current sequence number."""
        self._data['current_sequence'] = value
        self._modified = True

    @workflow_type.setter
    def workflow_type(self, value: str) -> None:
        """Set the workflow type."""
        self._data['workflow_type'] = value
        self._modified = True
    
    @metadata.setter
    def metadata(self, value: Dict[str, Any]) -> None:
        """Set the metadata."""
        self._data['metadata'] = value
        self._modified = True
    
    def start_cycle(self) -> bool:
        """Start the workflow cycle."""
        if self.status != WorkflowStatus.PENDING:
            return False
        
        self.status = WorkflowStatus.IN_PROGRESS
        self.started_at = datetime.now().isoformat()
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        return self.save()
    
    def cancel_cycle(self) -> bool:
        """Cancel the workflow cycle."""
        if self.status in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED, 
                           WorkflowStatus.REJECTED, WorkflowStatus.CANCELED]:
            return False
        
        self.status = WorkflowStatus.CANCELED
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        return self.save()
    
    def complete_cycle(self, decision: str = None) -> bool:
        """
        Complete the workflow cycle with optional decision.
        
        Args:
            decision: The final decision (approval, rejection, etc.)
            
        Returns:
            Boolean indicating success
        """
        # Check if already completed
        if self.is_complete():
            return False
            
        # Set decision if provided
        if decision:
            self.decision = decision
            
        # Update status based on decision
        if decision == "REJECTED":
            self.status = WorkflowStatus.REJECTED
        elif self.status not in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED]:
            if hasattr(self, "__class__") and hasattr(self.__class__, "__name__"):
                if self.__class__.__name__ == "ReviewCycle":
                    self.status = WorkflowStatus.COMPLETED
                elif self.__class__.__name__ == "ApprovalCycle":
                    self.status = WorkflowStatus.APPROVED
                    
        # Set completion time
        self.completed_at = datetime.now().isoformat()
        self._data['updated_at'] = datetime.now().isoformat()
        self._modified = True
        
        return self.save()
    
    def is_overdue(self) -> bool:
        """Whether workflow cycle is overdue."""
        return (
            self.due_date is not None 
            and datetime.fromisoformat(self.due_date) < datetime.now() 
            and not self.is_complete()
        )
    
    def is_complete(self) -> bool:
        """Check if the workflow cycle is complete."""
        return self.status in [WorkflowStatus.COMPLETED, WorkflowStatus.APPROVED]
    
    def update_status(self) -> None:
        """Update the cycle status based on assignment statuses."""
        # To be implemented in subclasses
        pass
    
    def get_progress(self) -> Dict[str, int]:
        """Get progress statistics for the cycle."""
        # To be implemented in subclasses
        return {
            "total": 0,
            "pending": 0,
            "in_progress": 0,
            "completed": 0,
            "rejected": 0,
            "skipped": 0,
        }
    
    def save(self) -> bool:
        """Save changes to database."""
        # To be implemented in subclasses
        raise NotImplementedError("Subclasses must implement save method")