🔍 Code Extractor

class ApprovalCycle_v1

Maturity: 28

Model representing a approval cycle for a document version.

File:
/tf/active/vicechatdev/CDocs/models/approval.py
Lines:
220 - 1221
Complexity:
moderate

Purpose

Model representing a approval cycle for a document version.

Source Code

class ApprovalCycle(BaseModel):
    """Model representing a approval cycle for a document version."""
    
    def __init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None):
        """
        Initialize a approval cycle.
        
        Args:
            data: Dictionary of approval cycle properties
            uid: Approval cycle UID to load from database (if data not provided)
        """
        if data is None and uid is not None:
            # Fetch approval cycle data from database
            data = db.get_node_by_uid(uid)
            
        super().__init__(data or {})
        self._comments_cache = None
        self._approvers_cache = None
    
    @classmethod
    def create(cls, document_version_uid: str, 
              approvers: List[Union[DocUser, str]],
              due_date: Optional[datetime] = None,
              instructions: str = '',
              properties: Optional[Dict[str, Any]] = None) -> Optional['ApprovalCycle']:
        """
        Create a new approval cycle.
        
        Args:
            document_version_uid: UID of the document version to approval
            approvers: List of users or UIDs to assign as approvers
            due_date: Date when approval should be completed
            instructions: Instructions for approvers
            properties: Additional properties for the approval cycle
            
        Returns:
            New ApprovalCycle instance or None if creation failed
        """
        try:
            # Generate due date if not provided
            if not due_date:
                due_date = datetime.now() + timedelta(days=settings.DEFAULT_APPROVAL_DAYS)
                
            # Prepare properties
            props = properties or {}
            props.update({
                'status': 'PENDING',
                'startDate': datetime.now(),
                'dueDate': due_date,
                'instructions': instructions
            })
            
            # Create node in database
            approval_data = db.create_node_with_relationship(
                NodeLabels.APPROVAL_CYCLE,
                props,
                document_version_uid,
                RelTypes.FOR_APPROVAL
            )
            
            if not approval_data:
                logger.error(f"Failed to create approval cycle for document version {document_version_uid}")
                return None
                
            # Create the approval cycle instance
            approval_cycle = cls(approval_data)
            
            # Add approvers
            #approver_uids = []
            #for approver in approvers:
                #approver_uid = approver.uid if isinstance(approver, DocUser) else approver
                #approver_uids.append(approver_uid)
                #cls.add_approver(approver)
                # db.create_relationship(
                #     approval_cycle.uid,
                #     approver_uid,
                #     RelTypes.APPROVALED_BY
                # )
            
            # Update document version status
            from CDocs.models.document import DocumentVersion
            version = DocumentVersion(uid=document_version_uid)
            if version:
                # Update version status
                version.status = 'IN_APPROVAL'
                
                # Also update document status
                document = version.document
                if document:
                    document.status = 'IN_APPROVAL'
            
            return approval_cycle
            
        except Exception as e:
            logger.error(f"Error creating approval cycle: {e}")
            return None
    
    @classmethod
    def get_approvals_for_document(cls, document_uid: str) -> List['ApprovalCycle']:
        """
        Get all approval cycles for a document.
        
        Args:
            document_uid: UID of the document
            
        Returns:
            List of ApprovalCycle instances
        """
        try:
            # Query for approval cycles related to the document
            result = db.run_query(
                """
                MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
                MATCH (v)<-[:FOR_APPROVAL]-(r:ApprovalCycle)
                RETURN r.UID as uid
                ORDER BY r.startDate DESC
                """,
                {"doc_uid": document_uid}
            )
            
            # Create ApprovalCycle instances
            approvals = [cls(uid=record['uid']) for record in result if 'uid' in record]
            return approvals
            
        except Exception as e:
            logger.error(f"Error getting approvals for document {document_uid}: {e}")
            return []

    @property
    def status(self) -> str:
        """Get approval cycle status."""
        return self._data.get('status', '')
    
    @status.setter
    def status(self, value: str) -> None:
        """Set approval cycle status."""
        if value not in settings.APPROVAL_STATUSES:
            logger.warning(f"Invalid approval status: {value}")
            return
            
        old_status = self.status
        
        self._data['status'] = value
        update_data = {'status': value}
        
        # Set completion date if we're transitioning to a terminal status
        if value in ['COMPLETED', 'REJECTED'] and old_status not in ['COMPLETED', 'REJECTED']:
            self._data['completionDate'] = datetime.now()
            update_data['completionDate'] = self._data['completionDate']
            
        db.update_node(self.uid, update_data)

    # Add this to the ApprovalCycle class after the required_approval_percentage property
    @property
    def decision(self) -> Optional[str]:
        """
        Get the overall decision for this approval cycle.
        
        Returns:
            Decision string (APPROVED, REJECTED, etc.) or None if not set
        """
        return self._data.get('decision')
        
    @decision.setter
    def decision(self, value: str) -> None:
        """
        Set the overall decision for this approval cycle.
        
        Args:
            value: Decision value (APPROVED, REJECTED, etc.)
        """
        if value and value not in settings.APPROVAL_DECISIONS:
            logger.warning(f"Invalid approval decision: {value}. Using allowed values from settings.APPROVAL_DECISIONS")
            return
            
        self._data['decision'] = value
        db.update_node(self.uid, {'decision': value})

    # Add this property to the ApprovalCycle class
    @property
    def initiated_by_uid(self) -> Optional[str]:
        """
        Get the UID of the user who initiated this approval cycle.
        
        Returns:
            User UID string or None if not set
        """
        return self._data.get('initiated_by_uid')
        
    @initiated_by_uid.setter
    def initiated_by_uid(self, uid: str) -> None:
        """
        Set the UID of the user who initiated this approval cycle.
        
        Args:
            uid: User UID
        """
        if uid:
            self._data['initiated_by_uid'] = uid
            db.update_node(self.uid, {'initiated_by_uid': uid})

    @property
    def required_approval_percentage(self) -> int:
        """
        Get the required approval percentage for this approval cycle.
        
        Returns:
            Required approval percentage (defaults to 100 if not set)
        """
        return self._data.get('required_approval_percentage', 100)
    
    @required_approval_percentage.setter
    def required_approval_percentage(self, value: int) -> None:
        """
        Set the required approval percentage.
        
        Args:
            value: Percentage value (1-100)
        """
        if not isinstance(value, int) or value < 1 or value > 100:
            logger.warning(f"Invalid approval percentage: {value}. Must be an integer between 1 and 100.")
            return
            
        self._data['required_approval_percentage'] = value
        db.update_node(self.uid, {'required_approval_percentage': value})

    def can_approval(self, approver_uid: str) -> bool:
        """
        Check if a user can approval right now (important for sequential approvals).
        
        Args:
            approver_uid: UID of the approver
            
        Returns:
            Boolean indicating if user can approval now
        """
        try:
            # Get the assignment
            assignment = self.get_approver_assignment(approver_uid)
            if not assignment:
                return False
                
            # If approval is not sequential, anyone can approval
            if not self.sequential:
                return True
                
            # In sequential mode, only active or completed approvers can approval
            return assignment.status in ['ACTIVE', 'COMPLETED']
            
        except Exception as e:
            logger.error(f"Error checking if user can approval: {e}")
            return False

    # Add this method to the ApprovalCycle class
    def save(self) -> bool:
        """Save changes to database."""
        try:
            # If node doesn't exist, create it
            if not db.node_exists(self.uid):
                created = db.create_node_with_uid(
                    NodeLabels.APPROVAL_CYCLE,
                    self._data,
                    self.uid
                )
                
                if created:
                    # If document version relationship needs to be established
                    document_version_uid = self._data.get('document_version_uid')
                    if document_version_uid:
                        # Create relationship between approval cycle and document version
                        db.create_relationship(
                            document_version_uid,
                            self.uid,
                            RelTypes.FOR_APPROVAL
                        )
                    
                    # Add relationships to approvers if needed
                    approver_uids = self._data.get('approver_uids', [])
                    for approver_uid in approver_uids:
                        db.create_relationship(
                            self.uid, 
                            approver_uid,
                            RelTypes.APPROVED_BY
                        )
                        
                return created
            
            # Otherwise update existing node
            return db.update_node(self.uid, self._data)
        except Exception as e:
            logger.error(f"Error saving approval cycle: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return False
    
    @property
    def start_date(self) -> Optional[datetime]:
        """Get when approval cycle started."""
        return self._data.get('startDate')
    
    @property
    def due_date(self) -> Optional[datetime]:
        """Get when approval is due."""
        return self._data.get('dueDate')
    
    @due_date.setter
    def due_date(self, date: datetime) -> None:
        """Set due date."""
        self._data['dueDate'] = date
        db.update_node(self.uid, {'dueDate': date})
    
    @property
    def completion_date(self) -> Optional[datetime]:
        """Get when approval was completed."""
        return self._data.get('completionDate')
    
    @completion_date.setter
    def completion_date(self, date: datetime) -> None:
        """Set completion date."""
        self._data['completionDate'] = date
        db.update_node(self.uid, {'completionDate': date})
    
    @property
    def instructions(self) -> str:
        """Get instructions for approvers."""
        return self._data.get('instructions', '')
    
    @property
    def is_completed(self) -> bool:
        """Whether approval cycle is completed."""
        return self.status in ['COMPLETED', 'REJECTED']
    
    @property
    def is_overdue(self) -> bool:
        """Whether approval is overdue."""
        return (
            self.due_date is not None 
            and datetime.now() > self.due_date 
            and not self.is_completed
        )
    
    @property
    def document_version_uid(self) -> Optional[str]:
        """Get the UID of the document version being approvaled."""
        result = db.run_query(
            """
            MATCH (r:ApprovalCycle {UID: $uid})-[:FOR_APPROVAL]->(v:DocumentVersion)
            RETURN v.UID as version_uid
            """,
            {"uid": self.uid}
        )
        
        if result and 'version_uid' in result[0]:
            return result[0]['version_uid']
            
        return None
    
    @property
    def document_uid(self) -> Optional[str]:
        """Get the UID of the controlled document  being approvaled."""
        result = db.run_query(
            """
            MATCH (r:ApprovalCycle {UID: $uid})-[:FOR_APPROVAL]->(v:DocumentVersion)<-[:HAS_VERSION]-(d:ControlledDocument)
            RETURN d.UID as document_uid
            """,
            {"uid": self.uid}
        )
        
        if result and 'document_uid' in result[0]:
            return result[0]['document_uid']
            
        return None
    
    @property
    def document_version_number(self) -> Optional[str]:
        """Get the version number being approvaled."""
        result = db.run_query(
            """
            MATCH (r:ApprovalCycle {UID: $uid})-[:FOR_APPROVAL]->(v:DocumentVersion)
            RETURN v.version_number as document_version_number
            """,
            {"uid": self.uid}
        )
        
        if result and 'document_version_number' in result[0]:
            return result[0]['document_version_number']
            
        return None
    
    @property
    def document_version(self) -> Optional[Any]:
        """Get the document version being approvaled."""
        from CDocs.models.document import DocumentVersion
        
        version_uid = self.document_version_uid
        if version_uid:
            return DocumentVersion(uid=version_uid)
            
        return None
    
    @property
    def document(self) -> Optional[Any]:
        """Get the controlled document  being approvaled."""
        from CDocs.models.document import ControlledDocument
        
        document_uid = self.document_uid
        if document_uid:
            return ControlledDocument(uid=document_uid)
            
        return None
    
    @property
    def approvers(self) -> List[DocUser]:
        """Get users assigned as approvers."""
        if self._approvers_cache is not None:
            return self._approvers_cache
            
        result = db.run_query(
            """
            MATCH (r:ApprovalCycle {UID: $uid})-[:APPROVED_BY]->(u:User)
            RETURN u
            """,
            {"uid": self.uid}
        )
        
        approvers = [DocUser(record['u']) for record in result if 'u' in record]
        self._approvers_cache = approvers
        return approvers
    
    @property
    def approver_uids(self) -> List[str]:
        """Get UIDs of users assigned as approvers."""
        return [approver.uid for approver in self.approvers]
    
        # In /tf/active/CDocs/models/approval.py
    
    def add_approver(self, approver: Union[DocUser, str], instructions: Optional[str] = None) -> bool:
        """
        Add a approver to the approval cycle.
        
        Args:
            approver: User or UID to add as approver
            instructions: Optional approver-specific instructions
            
        Returns:
            Boolean indicating success
        """
        try:
            approver_uid = approver.uid if isinstance(approver, DocUser) else approver
            
            # Check if already a approver
            if approver_uid in self.approver_uids:
                return True
                
            # Add approver relationship
            success = db.create_relationship(
                self.uid,
                approver_uid,
                RelTypes.APPROVED_BY
            )
            
            # Create a ApproverAssignment node to track this approver's progress
            if success:
                # Get user info 
                user_info = db.run_query(
                    """
                    MATCH (u:User {UID: $uid})
                    RETURN u.Name as name
                    """,
                    {"uid": approver_uid}
                )
                
                # Create assignment data
                assignment_uid = str(uuid.uuid4())
                assignment_data = {
                    'UID': assignment_uid,
                    'approver_uid': approver_uid,
                    'approver_name': user_info[0]['name'] if user_info else "Unknown",
                    'approval_cycle_uid': self.uid,
                    'status': 'PENDING',
                    'assigned_date': datetime.now()
                }
                
                # Add instructions if provided
                if instructions:
                    assignment_data['instructions'] = instructions
                
                # Create the node
                db.create_node(NodeLabels.APPROVER_ASSIGNMENT, assignment_data)
                
                # Create the ASSIGNMENT relationship between ApprovalCycle and ApproverAssignment
                db.create_relationship(
                    self.uid, 
                    assignment_uid,
                    RelTypes.ASSIGNMENT
                )
            
            # Clear cache
            self._approvers_cache = None
            
            return success
                
        except Exception as e:
            logger.error(f"Error adding approver: {e}")
            return False
    
    def remove_approver(self, approver: Union[DocUser, str]) -> bool:
        """
        Remove a approver from the approval cycle.
        
        Args:
            approver: User or UID to remove
            
        Returns:
            Boolean indicating success
        """
        try:
            approver_uid = approver.uid if isinstance(approver, DocUser) else approver
            
            # Remove relationship
            rel_result = db.run_query(
                """
                MATCH (r:ApprovalCycle {UID: $approval_uid})-[rel:APPROVED_BY]->(u:User {UID: $user_uid})
                DELETE rel
                RETURN count(rel) AS deleted
                """,
                {"approval_uid": self.uid, "user_uid": approver_uid}
            )
            
            success = rel_result and rel_result[0].get('deleted', 0) > 0
            
            # Also remove any associated ApproverAssignment
            if success:
                db.run_query(
                    """
                    MATCH (a:ApproverAssignment)
                    WHERE a.approval_cycle_uid = $approval_uid AND a.approver_uid = $approver_uid
                    DELETE a
                    """,
                    {"approval_uid": self.uid, "approver_uid": approver_uid}
                )
            
            # Clear cache
            self._approvers_cache = None
            
            return success
                
        except Exception as e:
            logger.error(f"Error removing approver: {e}")
            return False
    
    @property
    def comments(self) -> List[ApprovalComment]:
        """Get all comments for this approval cycle."""
        if self._comments_cache is not None:
            return self._comments_cache
            
        result = db.run_query(
            """
            MATCH (c:ApprovalComment)-[:COMMENTED_ON]->(r:ApprovalCycle {UID: $uid})
            RETURN c
            ORDER BY c.timestamp DESC
            """,
            {"uid": self.uid}
        )
        
        comments = [ApprovalComment(record['c']) for record in result if 'c' in record]
        self._comments_cache = comments
        return comments
    
    def add_comment(self, commenter: Union[DocUser, str], 
                   text: str,
                   requires_resolution: bool = False) -> Optional[ApprovalComment]:
        """
        Add a comment to the approval cycle.
        
        Args:
            commenter: User making the comment or their UID
            text: Comment text
            requires_resolution: Whether this comment requires resolution
            
        Returns:
            New ApprovalComment instance or None if creation failed
        """
        comment = ApprovalComment.create(
            self.uid, 
            commenter, 
            text, 
            requires_resolution
        )
        
        if comment:
            # Update status if needed
            if self.status == 'PENDING':
                self.status = 'IN_PROGRESS'
                
            # Clear cache
            self._comments_cache = None
            
        return comment
    
    def get_unresolved_comments(self) -> List[ApprovalComment]:
        """
        Get all unresolved comments that require resolution.
        
        Returns:
            List of unresolved ApprovalComment instances
        """
        # First get all comments using the updated method which handles relationships correctly
        all_comments = self.comments
        
        # Then filter for unresolved ones
        return [
            comment for comment in all_comments 
            if comment.requires_resolution and not comment.is_resolved
        ]
    
    def complete_approval(self, approved: bool = True) -> bool:
        """
        Complete the approval cycle.
        
        Args:
            approved: Whether the document was approved in the approval
            
        Returns:
            Boolean indicating success
        """
        try:
            # Check if we have unresolved comments
            unresolved = self.get_unresolved_comments()
            if unresolved and approved:
                logger.warning(f"Cannot complete approval with {len(unresolved)} unresolved comments")
                return False
                
            # Update status
            self.status = 'COMPLETED' if approved else 'REJECTED'
            
            # Update document version status if needed
            document_version = self.document_version
            if document_version:
                if approved:
                    # Move to approval if document type requires it
                    doc_type = document_version.document.doc_type
                    doc_type_details = settings.get_document_type(doc_type)
                    
                    if doc_type_details and doc_type_details.get('required_approvals'):
                        document_version.status = 'IN_APPROVAL'
                        if document_version.document:
                            document_version.document.status = 'IN_APPROVAL'
                    else:
                        # No approval needed, mark as approved
                        document_version.status = 'APPROVED'
                        if document_version.document:
                            document_version.document.status = 'APPROVED'
                else:
                    # Approval rejected, revert to draft
                    document_version.status = 'DRAFT'
                    if document_version.document:
                        document_version.document.status = 'DRAFT'
            
            return True
            
        except Exception as e:
            logger.error(f"Error completing approval: {e}")
            return False
    
    def get_approval_status(self, approver: Union[DocUser, str]) -> Dict[str, Any]:
        """
        Get approval status for a specific approver.
        
        Args:
            approver: Approver to check
            
        Returns:
            Dictionary with approval status information
        """
        approver_uid = approver.uid if isinstance(approver, DocUser) else approver
        
        # Check if user is a approver
        if approver_uid not in self.approver_uids:
            return {'status': 'NOT_ASSIGNED'}
            
        # Check if approver has commented
        result = db.run_query(
            """
            MATCH (r:ApprovalCycle {UID: $approval_uid})-[:COMMENTED_ON]->(c:ApprovalComment)
            WHERE c.commenterUID = $approver_uid
            RETURN count(c) as comment_count
            """,
            {"approval_uid": self.uid, "approver_uid": approver_uid}
        )
        
        comment_count = result[0]['comment_count'] if result else 0
        
        if comment_count > 0:
            return {
                'status': 'COMMENTED',
                'comment_count': comment_count
            }
        else:
            return {'status': 'PENDING'}
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary representation."""
        result = super().to_dict()
        
        # Add status details
        status_details = settings.APPROVAL_STATUSES.get(self.status, {})
        if status_details:
            result['statusName'] = status_details.get('name', self.status)
            result['statusColor'] = status_details.get('color', '#000000')
            result['statusIcon'] = status_details.get('icon', 'clock')
            
        # Add approver information
        result['approvers'] = [
            {
                'UID': approver.uid,
                'name': approver.name,
                'email': approver.email
            }
            for approver in self.approvers
        ]
        
        # Add comment counts
        all_comments = self.comments
        result['commentCount'] = len(all_comments)
        result['unresolvedCount'] = len(self.get_unresolved_comments())
        
        # Add document version info
        version = self.document_version
        if version:
            result['documentVersion'] = {
                'UID': version.uid,
                'versionNumber': version.version_number,
                'status': version.status
            }
            document = version.document
            if document:
                result['document'] = {
                    'UID': document.uid,
                    'docNumber': document.doc_number,
                    'title': document.title
                }
        
        return result

    def get_approver_assignments(self) -> List['ApproverAssignment']:
        """
        Get all approver assignments for this approval cycle.
        
        Returns:
            List of ApproverAssignment instances
        """
        try:
            # First try to get assignments using the ASSIGNMENT relationship
            result = db.run_query(
                """
                MATCH (r:ApprovalCycle {UID: $approval_uid})-[:ASSIGNMENT]->(a:ApproverAssignment)
                RETURN a
                """,
                {"approval_uid": self.uid}
            )
            
            assignments = []
            
            # If we found assignments via relationships, use those
            if result and len(result) > 0:
                assignments = [ApproverAssignment(record['a']) for record in result if 'a' in record]
                
                # If we found some assignments but not for all approvers, we need to check for legacy ones
                if len(assignments) < len(self.approver_uids):
                    # Get all approver UIDs that already have an assignment
                    assigned_approver_uids = [a.approver_uid for a in assignments]
                    
                    # Look for approvers without assignments
                    for approver_uid in self.approver_uids:
                        if approver_uid not in assigned_approver_uids:
                            # Create assignment for this approver
                            assignment = self.get_approver_assignment(approver_uid)
                            if assignment:
                                assignments.append(assignment)
            else:
                # Try the old method using APPROVED_BY + property lookup
                result = db.run_query(
                    """
                    MATCH (r:ApprovalCycle {UID: $approval_uid})-[:APPROVED_BY]->(u:User)
                    OPTIONAL MATCH (a:ApproverAssignment)
                    WHERE a.approval_cycle_uid = $approval_uid AND a.approver_uid = u.UID
                    RETURN u, a
                    """,
                    {"approval_uid": self.uid}
                )
                
                for record in result:
                    user = record.get('u', {})
                    assignment_data = record.get('a', {})
                    
                    # If no existing assignment record, create one with default values
                    if not assignment_data:
                        # Create a new ApproverAssignment
                        assignment_uid = str(uuid.uuid4())
                        assignment_data = {
                            'UID': assignment_uid,
                            'approver_uid': user.get('UID'),
                            'approver_name': user.get('Name'),
                            'approval_cycle_uid': self.uid,
                            'status': 'PENDING',
                            'assigned_date': datetime.now()
                        }
                        
                        # Create the node in the database
                        from CDocs.db.schema_manager import NodeLabels
                        db.create_node(NodeLabels.APPROVER_ASSIGNMENT, assignment_data)
                        
                        # Create the relationship
                        db.create_relationship(
                            self.uid,
                            assignment_uid,
                            RelTypes.ASSIGNMENT
                        )
                    else:
                        # Create relationship if it doesn't exist already (backward compatibility)
                        db.create_relationship(
                            self.uid,
                            assignment_data.get('UID'),
                            RelTypes.ASSIGNMENT
                        )
                    
                    # Create and return the assignment object
                    assignments.append(ApproverAssignment(assignment_data))
            
            return assignments
                
        except Exception as e:
            logger.error(f"Error getting approver assignments: {e}")
            return []

    def get_approver_assignment(self, approver_uid: str) -> Optional['ApproverAssignment']:
        """
        Get assignment for a specific approver.
        
        Args:
            approver_uid: UID of the approver
            
        Returns:
            ApproverAssignment instance or None if not found
        """
        try:
            # First check if this user is a approver (using the APPROVED_BY relationship)
            approver_check = db.run_query(
                """
                MATCH (r:ApprovalCycle {UID: $approval_uid})-[:APPROVED_BY]->(u:User {UID: $approver_uid})
                RETURN count(u) > 0 as is_approver
                """,
                {"approval_uid": self.uid, "approver_uid": approver_uid}
            )
            
            # Exit early if not a approver
            if not approver_check or not approver_check[0].get('is_approver', False):
                logger.warning(f"User {approver_uid} is not a approver for cycle {self.uid}")
                return None
                
            # Try to find existing assignment node using ASSIGNMENT relationship
            result = db.run_query(
                """
                MATCH (r:ApprovalCycle {UID: $approval_uid})-[:ASSIGNMENT]->(a:ApproverAssignment)
                WHERE a.approver_uid = $approver_uid
                RETURN a
                """,
                {"approval_uid": self.uid, "approver_uid": approver_uid}
            )
            
            # If assignment exists via relationship, return it
            if result and 'a' in result[0]:
                return ApproverAssignment(result[0]['a'])
            
            # Also check by property for backward compatibility
            result = db.run_query(
                """
                MATCH (a:ApproverAssignment)
                WHERE a.approval_cycle_uid = $approval_uid
                AND a.approver_uid = $approver_uid
                RETURN a
                """,
                {"approval_uid": self.uid, "approver_uid": approver_uid}
            )
            
            # If assignment exists via property, return it
            if result and 'a' in result[0]:
                # Create the relationship that was missing
                db.create_relationship(
                    self.uid,
                    result[0]['a']['UID'],
                    RelTypes.ASSIGNMENT
                )
                return ApproverAssignment(result[0]['a'])
            
            # Otherwise, create a new assignment to represent this approver    
            # Get user info
            user_info = db.run_query(
                """
                MATCH (u:User {UID: $uid})
                RETURN u.Name as name
                """,
                {"uid": approver_uid}
            )
            
            # Create new assignment
            assignment_uid = str(uuid.uuid4())
            assignment_data = {
                'UID': assignment_uid,
                'approver_uid': approver_uid,
                'approver_name': user_info[0]['name'] if user_info else "Unknown",
                'approval_cycle_uid': self.uid,
                'status': 'PENDING',
                'assigned_date': datetime.now()
            }
            
            # Create the node in the database
            from CDocs.db.schema_manager import NodeLabels
            db.create_node(NodeLabels.APPROVER_ASSIGNMENT, assignment_data)
            
            # Create the ASSIGNMENT relationship
            db.create_relationship(
                self.uid,
                assignment_uid,
                RelTypes.ASSIGNMENT
            )
            
            # Return the new assignment
            return ApproverAssignment(assignment_data)
                
        except Exception as e:
            logger.error(f"Error getting approver assignment: {e}")
            return None

    def is_approver(self, approver_uid: str) -> bool:
        """
        Check if a user is a approver for this cycle.
        
        Args:
            approver_uid: UID of the user to check
            
        Returns:
            Boolean indicating if user is a approver
        """
        try:
            # Use the APPROVED_BY relationship directly
            result = db.run_query(
                """
                MATCH (r:ApprovalCycle {UID: $approval_uid})-[:APPROVED_BY]->(u:User {UID: $approver_uid})
                RETURN count(u) > 0 as is_approver
                """,
                {"approval_uid": self.uid, "approver_uid": approver_uid}
            )
            
            return result and result[0].get('is_approver', False)
                
        except Exception as e:
            logger.error(f"Error checking if user is approver: {e}")
            return False

    def get_next_approver(self, current_sequence: int) -> Optional['ApproverAssignment']:
        """
        Get the next approver in sequence for sequential approvals.
        
        Args:
            current_sequence: Current sequence number
            
        Returns:
            Next ApproverAssignment in sequence or None
        """
        try:
            if not self.sequential:
                return None
                
            result = db.run_query(
                """
                MATCH (a:ApproverAssignment)
                WHERE a.approval_cycle_uid = $approval_uid
                AND a.sequence_order > $sequence
                AND a.status = 'PENDING'
                RETURN a
                ORDER BY a.sequence_order
                LIMIT 1
                """,
                {"approval_uid": self.uid, "sequence": current_sequence}
            )
            
            if result and 'a' in result[0]:
                return ApproverAssignment(result[0]['a'])
                
            return None
            
        except Exception as e:
            logger.error(f"Error getting next approver: {e}")
            return None

    @property
    def sequential(self) -> bool:
        """Whether this approval cycle uses sequential approval."""
        return self._data.get('sequential', False)

Parameters

Name Type Default Kind
bases BaseModel -

Parameter Details

bases: Parameter of type BaseModel

Return Value

Returns unspecified type

Class Interface

Methods

__init__(self, data, uid)

Purpose: Initialize a approval cycle. Args: data: Dictionary of approval cycle properties uid: Approval cycle UID to load from database (if data not provided)

Parameters:

  • data: Type: Optional[Dict[str, Any]]
  • uid: Type: Optional[str]

Returns: None

create(cls, document_version_uid, approvers, due_date, instructions, properties) -> Optional['ApprovalCycle']

Purpose: Create a new approval cycle. Args: document_version_uid: UID of the document version to approval approvers: List of users or UIDs to assign as approvers due_date: Date when approval should be completed instructions: Instructions for approvers properties: Additional properties for the approval cycle Returns: New ApprovalCycle instance or None if creation failed

Parameters:

  • cls: Parameter
  • document_version_uid: Type: str
  • approvers: Type: List[Union[DocUser, str]]
  • due_date: Type: Optional[datetime]
  • instructions: Type: str
  • properties: Type: Optional[Dict[str, Any]]

Returns: Returns Optional['ApprovalCycle']

get_approvals_for_document(cls, document_uid) -> List['ApprovalCycle']

Purpose: Get all approval cycles for a document. Args: document_uid: UID of the document Returns: List of ApprovalCycle instances

Parameters:

  • cls: Parameter
  • document_uid: Type: str

Returns: Returns List['ApprovalCycle']

status(self) -> str property

Purpose: Get approval cycle status.

Returns: Returns str

status(self, value) -> None

Purpose: Set approval cycle status.

Parameters:

  • value: Type: str

Returns: Returns None

decision(self) -> Optional[str] property

Purpose: Get the overall decision for this approval cycle. Returns: Decision string (APPROVED, REJECTED, etc.) or None if not set

Returns: Returns Optional[str]

decision(self, value) -> None

Purpose: Set the overall decision for this approval cycle. Args: value: Decision value (APPROVED, REJECTED, etc.)

Parameters:

  • value: Type: str

Returns: Returns None

initiated_by_uid(self) -> Optional[str] property

Purpose: Get the UID of the user who initiated this approval cycle. Returns: User UID string or None if not set

Returns: Returns Optional[str]

initiated_by_uid(self, uid) -> None

Purpose: Set the UID of the user who initiated this approval cycle. Args: uid: User UID

Parameters:

  • uid: Type: str

Returns: Returns None

required_approval_percentage(self) -> int property

Purpose: Get the required approval percentage for this approval cycle. Returns: Required approval percentage (defaults to 100 if not set)

Returns: Returns int

required_approval_percentage(self, value) -> None

Purpose: Set the required approval percentage. Args: value: Percentage value (1-100)

Parameters:

  • value: Type: int

Returns: Returns None

can_approval(self, approver_uid) -> bool

Purpose: Check if a user can approval right now (important for sequential approvals). Args: approver_uid: UID of the approver Returns: Boolean indicating if user can approval now

Parameters:

  • approver_uid: Type: str

Returns: Returns bool

save(self) -> bool

Purpose: Save changes to database.

Returns: Returns bool

start_date(self) -> Optional[datetime] property

Purpose: Get when approval cycle started.

Returns: Returns Optional[datetime]

due_date(self) -> Optional[datetime] property

Purpose: Get when approval is due.

Returns: Returns Optional[datetime]

due_date(self, date) -> None

Purpose: Set due date.

Parameters:

  • date: Type: datetime

Returns: Returns None

completion_date(self) -> Optional[datetime] property

Purpose: Get when approval was completed.

Returns: Returns Optional[datetime]

completion_date(self, date) -> None

Purpose: Set completion date.

Parameters:

  • date: Type: datetime

Returns: Returns None

instructions(self) -> str property

Purpose: Get instructions for approvers.

Returns: Returns str

is_completed(self) -> bool property

Purpose: Whether approval cycle is completed.

Returns: Returns bool

is_overdue(self) -> bool property

Purpose: Whether approval is overdue.

Returns: Returns bool

document_version_uid(self) -> Optional[str] property

Purpose: Get the UID of the document version being approvaled.

Returns: Returns Optional[str]

document_uid(self) -> Optional[str] property

Purpose: Get the UID of the controlled document being approvaled.

Returns: Returns Optional[str]

document_version_number(self) -> Optional[str] property

Purpose: Get the version number being approvaled.

Returns: Returns Optional[str]

document_version(self) -> Optional[Any] property

Purpose: Get the document version being approvaled.

Returns: Returns Optional[Any]

document(self) -> Optional[Any] property

Purpose: Get the controlled document being approvaled.

Returns: Returns Optional[Any]

approvers(self) -> List[DocUser] property

Purpose: Get users assigned as approvers.

Returns: Returns List[DocUser]

approver_uids(self) -> List[str] property

Purpose: Get UIDs of users assigned as approvers.

Returns: Returns List[str]

add_approver(self, approver, instructions) -> bool

Purpose: Add a approver to the approval cycle. Args: approver: User or UID to add as approver instructions: Optional approver-specific instructions Returns: Boolean indicating success

Parameters:

  • approver: Type: Union[DocUser, str]
  • instructions: Type: Optional[str]

Returns: Returns bool

remove_approver(self, approver) -> bool

Purpose: Remove a approver from the approval cycle. Args: approver: User or UID to remove Returns: Boolean indicating success

Parameters:

  • approver: Type: Union[DocUser, str]

Returns: Returns bool

comments(self) -> List[ApprovalComment] property

Purpose: Get all comments for this approval cycle.

Returns: Returns List[ApprovalComment]

add_comment(self, commenter, text, requires_resolution) -> Optional[ApprovalComment]

Purpose: Add a comment to the approval cycle. Args: commenter: User making the comment or their UID text: Comment text requires_resolution: Whether this comment requires resolution Returns: New ApprovalComment instance or None if creation failed

Parameters:

  • commenter: Type: Union[DocUser, str]
  • text: Type: str
  • requires_resolution: Type: bool

Returns: Returns Optional[ApprovalComment]

get_unresolved_comments(self) -> List[ApprovalComment]

Purpose: Get all unresolved comments that require resolution. Returns: List of unresolved ApprovalComment instances

Returns: Returns List[ApprovalComment]

complete_approval(self, approved) -> bool

Purpose: Complete the approval cycle. Args: approved: Whether the document was approved in the approval Returns: Boolean indicating success

Parameters:

  • approved: Type: bool

Returns: Returns bool

get_approval_status(self, approver) -> Dict[str, Any]

Purpose: Get approval status for a specific approver. Args: approver: Approver to check Returns: Dictionary with approval status information

Parameters:

  • approver: Type: Union[DocUser, str]

Returns: Returns Dict[str, Any]

to_dict(self) -> Dict[str, Any]

Purpose: Convert to dictionary representation.

Returns: Returns Dict[str, Any]

get_approver_assignments(self) -> List['ApproverAssignment']

Purpose: Get all approver assignments for this approval cycle. Returns: List of ApproverAssignment instances

Returns: Returns List['ApproverAssignment']

get_approver_assignment(self, approver_uid) -> Optional['ApproverAssignment']

Purpose: Get assignment for a specific approver. Args: approver_uid: UID of the approver Returns: ApproverAssignment instance or None if not found

Parameters:

  • approver_uid: Type: str

Returns: Returns Optional['ApproverAssignment']

is_approver(self, approver_uid) -> bool

Purpose: Check if a user is a approver for this cycle. Args: approver_uid: UID of the user to check Returns: Boolean indicating if user is a approver

Parameters:

  • approver_uid: Type: str

Returns: Returns bool

get_next_approver(self, current_sequence) -> Optional['ApproverAssignment']

Purpose: Get the next approver in sequence for sequential approvals. Args: current_sequence: Current sequence number Returns: Next ApproverAssignment in sequence or None

Parameters:

  • current_sequence: Type: int

Returns: Returns Optional['ApproverAssignment']

sequential(self) -> bool property

Purpose: Whether this approval cycle uses sequential approval.

Returns: Returns bool

Required Imports

import logging
import uuid
from typing import Dict
from typing import List
from typing import Any

Usage Example

# Example usage:
# result = ApprovalCycle(bases)

Tags

class approvalcycle

Similar Components

AI-powered semantic similarity - components with related functionality:

  • class ApprovalCycle 97.5% similar

    Model representing an approval cycle for a document version.

    From: /tf/active/vicechatdev/CDocs/models/approval_bis.py

  • class ReviewCycle 78.4% similar

    Model representing a review cycle for a document version.

    From: /tf/active/vicechatdev/CDocs/models/review.py

  • function create_approval_cycle_v1 74.0% similar

    Creates a new approval cycle for a controlled document, managing approver assignments, permissions, notifications, and audit logging.

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

  • class ApproverAssignment_v1 69.2% similar

    Model class representing an approver assignment within an approval cycle, managing the relationship between an approver and their approval task including status, decisions, and timeline tracking.

    From: /tf/active/vicechatdev/CDocs/models/approval.py

  • class ApprovalComment_v1 67.1% similar

    A model class representing a comment made during a document approval cycle, with support for resolution tracking and database persistence.

    From: /tf/active/vicechatdev/CDocs/models/approval.py
← Back to Browse