🔍 Code Extractor

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

Purpose

WorkflowControllerBase serves as an abstract base class for implementing workflow management systems (reviews, approvals, etc.) in a document management system. It defines a comprehensive interface for creating and managing workflow cycles, assignments, participants, comments, and notifications. Subclasses must implement all abstract methods to provide concrete functionality for specific workflow types. The class handles the complete lifecycle of workflows including creation, participant management, status updates, completion, cancellation, and notifications.

Source Code

class WorkflowControllerBase:
    """Base controller for workflow processes like reviews and approvals."""
    
    def __init__(self):
        """Initialize the base workflow controller."""
        self.workflow_type = None  # Should be set by subclasses e.g., 'REVIEW' or 'APPROVAL'
    
    def create_cycle(self, document_version_uid: str, 
                    user_uids: List[str],
                    due_date: Optional[datetime] = None,
                    instructions: str = '',
                    sequential: bool = False,
                    required_approval_percentage: int = 100,
                    initiated_by_uid: Optional[str] = None,
                    notify_users: bool = True) -> Optional[Dict[str, Any]]:
        """
        Create a new workflow cycle.
        
        Args:
            document_version_uid: UID of the document version
            user_uids: List of user UIDs to assign
            due_date: Optional due date (default: based on settings)
            instructions: Instructions for assignees
            sequential: Whether workflow should be sequential
            required_approval_percentage: Percentage of approvals required
            initiated_by_uid: UID of the user who initiated the workflow
            notify_users: Whether to send notifications to assigned users
            
        Returns:
            Dictionary with the created cycle information or None if failed
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement create_cycle")
    
    def get_cycle_by_uid(self, cycle_uid: str) -> Optional[Dict[str, Any]]:
        """
        Get workflow cycle information by UID.
        
        Args:
            cycle_uid: UID of the workflow cycle
            
        Returns:
            Dictionary with cycle information or None if not found
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_cycle_by_uid")
    
    def get_cycles_for_document(self, document_uid: str) -> List[Dict[str, Any]]:
        """
        Get all workflow cycles for a document.
        
        Args:
            document_uid: UID of the document
            
        Returns:
            List of dictionaries with cycle information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_cycles_for_document")
    
    def get_cycles_for_document_version(self, version_uid: str) -> List[Dict[str, Any]]:
        """
        Get all workflow cycles for a document version.
        
        Args:
            version_uid: UID of the document version
            
        Returns:
            List of dictionaries with cycle information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_cycles_for_document_version")
    
    def get_current_cycle_for_document(self, document_uid: str) -> Optional[Dict[str, Any]]:
        """
        Get the current active workflow cycle for a document.
        
        Args:
            document_uid: UID of the document
            
        Returns:
            Dictionary with cycle information or None if no active cycle
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_current_cycle_for_document")
    
    def get_assignments_for_cycle(self, cycle_uid: str) -> List[Dict[str, Any]]:
        """
        Get all assignments for a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            
        Returns:
            List of dictionaries with assignment information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_assignments_for_cycle")
    
    def get_assignment_by_uid(self, assignment_uid: str) -> Optional[Dict[str, Any]]:
        """
        Get assignment information by UID.
        
        Args:
            assignment_uid: UID of the assignment
            
        Returns:
            Dictionary with assignment information or None if not found
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_assignment_by_uid")
    
    def get_assignment_for_user(self, cycle_uid: str, user_uid: str) -> Optional[Dict[str, Any]]:
        """
        Get assignment for a specific user in a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            user_uid: UID of the user
            
        Returns:
            Dictionary with assignment information or None if not found
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_assignment_for_user")
    
    def get_assignments_for_user(self, user_uid: str, 
                                status_filter: Optional[List[str]] = None,
                                limit: int = 100, 
                                offset: int = 0) -> Tuple[List[Dict[str, Any]], int]:
        """
        Get all assignments for a user.
        
        Args:
            user_uid: UID of the user
            status_filter: Optional list of status values to filter by
            limit: Maximum number of results to return
            offset: Number of results to skip
            
        Returns:
            Tuple of (list of assignment dictionaries, total count)
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_assignments_for_user")
    
    def update_cycle_status(self, cycle_uid: str, status: str) -> bool:
        """
        Update the status of a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            status: New status value
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement update_cycle_status")
    
    def complete_cycle(self, cycle_uid: str, decision: str, comment: Optional[str] = None) -> bool:
        """
        Complete a workflow cycle with a decision.
        
        Args:
            cycle_uid: UID of the workflow cycle
            decision: Decision value (e.g., 'APPROVED' or 'REJECTED')
            comment: Optional comment about the decision
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement complete_cycle")
    
    def cancel_cycle(self, cycle_uid: str, reason: Optional[str] = None) -> bool:
        """
        Cancel a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            reason: Optional reason for cancellation
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement cancel_cycle")
    
    def add_participant(self, cycle_uid: str, user_uid: str, sequence_order: int = 0) -> bool:
        """
        Add a participant to a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            user_uid: UID of the user to add
            sequence_order: Order in the sequence (for sequential workflows)
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement add_participant")
    
    def remove_participant(self, cycle_uid: str, user_uid: str, reason: Optional[str] = None) -> bool:
        """
        Remove a participant from a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            user_uid: UID of the user to remove
            reason: Optional reason for removal
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement remove_participant")
    
    def update_assignment_status(self, assignment_uid: str, status: str) -> bool:
        """
        Update the status of an assignment.
        
        Args:
            assignment_uid: UID of the assignment
            status: New status value
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement update_assignment_status")
    
    def complete_assignment(self, assignment_uid: str, 
                           decision: str, 
                           comments: Optional[str] = None) -> bool:
        """
        Complete an assignment with a decision.
        
        Args:
            assignment_uid: UID of the assignment
            decision: Decision value (e.g., 'APPROVED' or 'REJECTED')
            comments: Optional comments about the decision
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement complete_assignment")
    
    def add_comment(self, cycle_uid: str, 
                   user_uid: str, 
                   text: str,
                   requires_resolution: bool = False,
                   comment_type: str = 'GENERAL',
                   parent_comment_uid: Optional[str] = None,
                   location: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
        """
        Add a comment to a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            user_uid: UID of the user making the comment
            text: Comment text
            requires_resolution: Whether the comment requires resolution
            comment_type: Type of comment
            parent_comment_uid: UID of parent comment (for replies)
            location: Location information (e.g., page number, coordinates)
            
        Returns:
            Dictionary with the created comment information or None if failed
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement add_comment")
    
    def get_comments(self, cycle_uid: str) -> List[Dict[str, Any]]:
        """
        Get all comments for a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            
        Returns:
            List of dictionaries with comment information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_comments")
    
    def resolve_comment(self, comment_uid: str, resolution_text: str) -> bool:
        """
        Resolve a comment.
        
        Args:
            comment_uid: UID of the comment
            resolution_text: Resolution text
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement resolve_comment")
    
    def get_comment_by_uid(self, comment_uid: str) -> Optional[Dict[str, Any]]:
        """
        Get comment information by UID.
        
        Args:
            comment_uid: UID of the comment
            
        Returns:
            Dictionary with comment information or None if not found
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_comment_by_uid")
    
    def update_due_date(self, cycle_uid: str, due_date: datetime) -> bool:
        """
        Update the due date for a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            due_date: New due date
            
        Returns:
            Boolean indicating success
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement update_due_date")
    
    def get_overdue_assignments(self) -> List[Dict[str, Any]]:
        """
        Get all overdue assignments.
        
        Returns:
            List of dictionaries with overdue assignment information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_overdue_assignments")
    
    def notify_participant(self, assignment_uid: str, message: Optional[str] = None) -> bool:
        """
        Send a notification to a workflow participant.
        
        Args:
            assignment_uid: UID of the assignment
            message: Optional custom message
            
        Returns:
            Boolean indicating success
        """
        try:
            # Get assignment details
            assignment = self.get_assignment_by_uid(assignment_uid)
            if not assignment:
                logger.error(f"Assignment {assignment_uid} not found for notification")
                return False
                
            user_uid = assignment.get('user_uid')
            if not user_uid:
                logger.error(f"No user UID found for assignment {assignment_uid}")
                return False
                
            # Get user details
            user = DocUser.get_by_uid(user_uid)
            if not user:
                logger.error(f"User {user_uid} not found for notification")
                return False
                
            # Get cycle details
            cycle_uid = assignment.get('cycle_uid')
            if not cycle_uid:
                logger.error(f"No cycle UID found for assignment {assignment_uid}")
                return False
                
            cycle = self.get_cycle_by_uid(cycle_uid)
            if not cycle:
                logger.error(f"Cycle {cycle_uid} not found for notification")
                return False
                
            # Get document details
            document_version_uid = cycle.get('document_version_uid')
            if not document_version_uid:
                logger.error(f"No document version UID found for cycle {cycle_uid}")
                return False
                
            from CDocs.models.document import DocumentVersion
            doc_version = DocumentVersion(uid=document_version_uid)
            if not doc_version:
                logger.error(f"Document version {document_version_uid} not found")
                return False
                
            document = doc_version.document
            if not document:
                logger.error(f"Document not found for version {document_version_uid}")
                return False
                
            # Prepare notification details
            title = f"{self.workflow_type} Request: {document.title}"
            if not message:
                message = f"You have been assigned to {self.workflow_type.lower()} document {document.doc_number}: {document.title}"
                
            # Send notification
            notification_data = {
                'recipient_uid': user_uid,
                'title': title,
                'message': message,
                'link_url': f"/document/{document.uid}/version/{doc_version.uid}/{self.workflow_type.lower()}",
                'notification_type': f"{self.workflow_type}_REQUEST",
                'source_uid': cycle_uid,
                'priority': 'MEDIUM'
            }
            
            return send_notification(notification_data)
            
        except Exception as e:
            logger.error(f"Error sending notification to participant: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return False

    def notify_all_participants(self, cycle_uid: str, message: Optional[str] = None) -> bool:
        """
        Send notifications to all participants in a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            message: Optional custom message
            
        Returns:
            Boolean indicating overall success
        """
        try:
            # Get all assignments for this cycle
            assignments = self.get_assignments_for_cycle(cycle_uid)
            if not assignments:
                logger.warning(f"No assignments found for cycle {cycle_uid}")
                return False
                
            # Send notification to each participant
            success_count = 0
            for assignment in assignments:
                if self.notify_participant(assignment['UID'], message):
                    success_count += 1
                    
            # Return true if at least one notification was sent successfully
            return success_count > 0
            
        except Exception as e:
            logger.error(f"Error notifying all participants: {e}")
            return False
    
    def get_workflow_statistics(self) -> Dict[str, Any]:
        """
        Get statistics about workflow cycles.
        
        Returns:
            Dictionary with statistics information
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement get_workflow_statistics")
    
    def search_cycles(self, query: Dict[str, Any], 
                     limit: int = 100, 
                     offset: int = 0) -> Tuple[List[Dict[str, Any]], int]:
        """
        Search for workflow cycles based on criteria.
        
        Args:
            query: Dictionary with search criteria
            limit: Maximum number of results to return
            offset: Number of results to skip
            
        Returns:
            Tuple of (list of cycle dictionaries, total count)
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement search_cycles")
    
    def is_user_participant(self, cycle_uid: str, user_uid: str) -> bool:
        """
        Check if a user is a participant in a workflow cycle.
        
        Args:
            cycle_uid: UID of the workflow cycle
            user_uid: UID of the user to check
            
        Returns:
            Boolean indicating if user is a participant
        """
        # This is an abstract method that should be implemented by subclasses
        raise NotImplementedError("Subclasses must implement is_user_participant")

Parameters

Parameter Details

__init__: No parameters required. The constructor initializes the workflow_type attribute to None, which should be set by subclasses to identify the specific workflow type (e.g., 'REVIEW' or 'APPROVAL').

Return Value

Instantiation returns a WorkflowControllerBase object. Methods return various types: cycle/assignment/comment creation methods return Optional[Dict[str, Any]] with the created object data or None on failure; query methods return List[Dict[str, Any]] for collections or Optional[Dict[str, Any]] for single items; status update methods return bool indicating success; search methods return Tuple[List[Dict[str, Any]], int] containing results and total count; notification methods return bool indicating success.

Class Interface

Methods

Attributes

Dependencies

• logging
• uuid
• typing
• datetime
• CDocs
• traceback

Required Imports

import logging
import uuid
from typing import Dict, List, Any, Optional, Union, Tuple
from datetime import datetime, timedelta
from CDocs import db
from CDocs.config import settings
from CDocs.db.schema_manager import NodeLabels, RelTypes
from CDocs.models.user_extensions import DocUser
from CDocs.utils.notifications import send_notification
import traceback

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs.models.document import DocumentVersion

Usage Example

# This is an abstract base class - must be subclassed
class ReviewController(WorkflowControllerBase):
    def __init__(self):
        super().__init__()
        self.workflow_type = 'REVIEW'
    
    def create_cycle(self, document_version_uid, user_uids, **kwargs):
        # Implement cycle creation logic
        return {'UID': 'cycle-123', 'status': 'ACTIVE'}
    
    # Implement all other abstract methods...

# Usage:
controller = ReviewController()

# Create a review cycle
cycle = controller.create_cycle(
    document_version_uid='doc-v1-uid',
    user_uids=['user1-uid', 'user2-uid'],
    due_date=datetime.now() + timedelta(days=7),
    instructions='Please review and provide feedback',
    sequential=False,
    required_approval_percentage=100,
    initiated_by_uid='manager-uid',
    notify_users=True
)

# Get assignments for the cycle
assignments = controller.get_assignments_for_cycle(cycle['UID'])

# Complete an assignment
controller.complete_assignment(
    assignment_uid=assignments[0]['UID'],
    decision='APPROVED',
    comments='Looks good to me'
)

# Add a comment
controller.add_comment(
    cycle_uid=cycle['UID'],
    user_uid='user1-uid',
    text='Please clarify section 3',
    requires_resolution=True,
    comment_type='QUESTION'
)

# Notify all participants
controller.notify_all_participants(
    cycle_uid=cycle['UID'],
    message='Reminder: Review due in 2 days'
)

Best Practices

• This is an abstract base class - never instantiate directly, always create a subclass that implements all abstract methods
• Subclasses must set self.workflow_type in their __init__ method to identify the workflow type (e.g., 'REVIEW', 'APPROVAL')
• All methods that raise NotImplementedError must be overridden in subclasses with concrete implementations
• The notify_participant and notify_all_participants methods are fully implemented and can be used as-is or overridden
• Error handling is built into notification methods with comprehensive logging - ensure logger is configured
• Methods follow a consistent pattern: cycle management, assignment management, comment management, and notification methods
• Use the get_assignment_by_uid method before calling notify_participant to ensure assignment exists
• Sequential workflows should use the sequence_order parameter in add_participant method
• Status updates should be validated against allowed status values in subclass implementations
• The required_approval_percentage parameter allows for partial approval workflows (e.g., 75% approval required)
• Comments support threading via parent_comment_uid and can include location data for document annotations
• Always check return values: None indicates failure for Optional returns, False indicates failure for bool returns
• Use pagination parameters (limit, offset) in search and list methods to handle large result sets
• The notify_users parameter in create_cycle allows control over automatic notifications at cycle creation

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class ReviewControllerBase 88.4% 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 ApprovalControllerBase 84.8% 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

• class WorkflowCycleBase 81.4% similar

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

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

• class AssignmentBase 74.0% 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 WorkflowPanelBase 73.6% similar

  Base class for workflow panels (review and approval) that provides common functionality and UI components.

  From: /tf/active/vicechatdev/CDocs single class/ui/workflow_panel_base.py