class ApprovalCycle
Model representing an approval cycle for a document version.
/tf/active/vicechatdev/CDocs/models/approval_bis.py
220 - 804
moderate
Purpose
Model representing an approval cycle for a document version.
Source Code
class ApprovalCycle(BaseModel):
"""Model representing an approval cycle for a document version."""
def __init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None):
"""
Initialize an 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 approve
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:
# Prepare properties
props = properties or {}
# Calculate due date if not provided
if not due_date:
default_days = settings.get('DEFAULT_APPROVAL_DAYS', 7)
due_date = datetime.now() + timedelta(days=default_days)
# Set standard properties
props.update({
'status': 'PENDING',
'createdAt': datetime.now(),
'dueDate': due_date,
'instructions': instructions,
'requiredApprovalPercentage': props.get('requiredApprovalPercentage', 100),
'approvalType': props.get('approvalType', 'STANDARD'),
'sequential': props.get('sequential', False)
})
# Create approval cycle node
approval_uid = str(uuid.uuid4())
props['UID'] = approval_uid
# Create node in database
success = db.create_node(NodeLabels.APPROVAL_CYCLE, props)
if not success:
logger.error("Failed to create approval cycle")
return None
# Create relationship: Approval Cycle -> FOR_APPROVAL -> Document Version
rel_success = db.create_relationship(
approval_uid,
document_version_uid,
RelTypes.FOR_APPROVAL
)
if not rel_success:
logger.error(f"Failed to create relationship between approval cycle and document version {document_version_uid}")
# Clean up the orphaned approval cycle node
db.delete_node(approval_uid)
return None
# Create approval cycle instance
approval_cycle = cls(props)
# Add approvers
sequence_order = 1
for approver in approvers:
if isinstance(approver, str):
approver_uid = approver
else:
approver_uid = approver.uid
# Add approver with sequential ordering if needed
approval_cycle.add_approver(
approver_uid,
sequence_order=sequence_order if props.get('sequential', False) else None
)
sequence_order += 1
# Update approval cycle status to IN_APPROVAL
approval_cycle.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."""
try:
result = db.run_query(
"""
MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
<-[:FOR_APPROVAL]-(a:ApprovalCycle)
RETURN a
ORDER BY a.createdAt DESC
""",
{"doc_uid": document_uid}
)
approval_cycles = []
for record in result:
if 'a' in record and record['a']:
approval_cycles.append(cls(record['a']))
return approval_cycles
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', 'UNKNOWN')
@status.setter
def status(self, new_status: str) -> None:
"""Set approval cycle status."""
old_status = self.status
if old_status == new_status:
return
self._data['status'] = new_status
self._data['statusUpdatedAt'] = datetime.now()
# Set completion date if status is terminal
if new_status in ['APPROVED', 'REJECTED', 'CANCELLED']:
self._data['completionDate'] = self._data['statusUpdatedAt']
# Update in database
db.update_node(self.uid, {
'status': new_status,
'statusUpdatedAt': self._data['statusUpdatedAt'],
'completionDate': self._data.get('completionDate')
})
@property
def decision(self) -> Optional[str]:
"""Get approval cycle decision."""
return self._data.get('decision')
@decision.setter
def decision(self, value: str) -> None:
"""Set approval cycle decision."""
self._data['decision'] = value
self._data['decisionDate'] = datetime.now()
# Update in database
db.update_node(self.uid, {
'decision': value,
'decisionDate': self._data['decisionDate']
})
@property
def initiated_by_uid(self) -> Optional[str]:
"""Get UID of user who initiated the approval cycle."""
return self._data.get('initiatedByUID')
@initiated_by_uid.setter
def initiated_by_uid(self, uid: str) -> None:
"""Set UID of user who initiated the approval cycle."""
self._data['initiatedByUID'] = uid
# Update in database
db.update_node(self.uid, {'initiatedByUID': uid})
@property
def required_approval_percentage(self) -> int:
"""Get required percentage of approvers that must approve."""
return self._data.get('requiredApprovalPercentage', 100)
@required_approval_percentage.setter
def required_approval_percentage(self, value: int) -> None:
"""Set required percentage of approvers that must approve."""
value = max(0, min(100, value)) # Ensure value is between 0 and 100
self._data['requiredApprovalPercentage'] = value
# Update in database
db.update_node(self.uid, {'requiredApprovalPercentage': value})
def can_approve(self, approver_uid: str) -> bool:
"""Check if a user can approve in the current state."""
# Get approver assignment
assignment = self.get_approver_assignment(approver_uid)
if not assignment:
return False
# If not sequential, any assigned approver can approve if status is IN_APPROVAL
if not self._data.get('sequential', False):
return self.status == 'IN_APPROVAL'
# For sequential approval, check if this approver is next in sequence
if self.status != 'IN_APPROVAL':
return False
# Find current sequence
current_sequence = 1 # Default to first sequence
# Get all assignments and find completed ones
assignments = self.get_approver_assignments()
for a in assignments:
if a.status == 'COMPLETED' and a.sequence_order >= current_sequence:
current_sequence = a.sequence_order + 1
# Check if this approver's turn
return assignment.sequence_order == current_sequence
def save(self) -> bool:
"""Save changes to database."""
try:
# If node doesn't exist, create it
if not db.node_exists(self.uid):
return db.create_node_with_uid(
NodeLabels.APPROVAL_CYCLE,
self._data,
self.uid
)
# Otherwise update existing node
return db.update_node(self.uid, self._data)
except Exception as e:
logger.error(f"Error saving approval cycle: {e}")
return False
@property
def document_version_uid(self) -> Optional[str]:
"""Get UID of document version associated with this approval cycle."""
# First check if we already have it in properties
if self._data.get('document_version_uid'):
return self._data.get('document_version_uid')
# Otherwise query the database
result = db.run_query(
"""
MATCH (a:ApprovalCycle {UID: $uid})-[:FOR_APPROVAL]->(v:DocumentVersion)
RETURN v.UID as version_uid
""",
{"uid": self.uid}
)
if result and 'version_uid' in result[0]:
# Cache it for future use
self._data['document_version_uid'] = result[0]['version_uid']
return result[0]['version_uid']
return None
@property
def document_uid(self) -> Optional[str]:
"""Get UID of document associated with this approval cycle."""
# First check if we already have it in properties
if self._data.get('document_uid'):
return self._data.get('document_uid')
# Otherwise query the database
result = db.run_query(
"""
MATCH (a:ApprovalCycle {UID: $uid})-[:FOR_APPROVAL]->(v:DocumentVersion)-[:VERSION_OF]->(d:Document)
RETURN d.UID as document_uid
""",
{"uid": self.uid}
)
if result and 'document_uid' in result[0]:
# Cache it for future use
self._data['document_uid'] = result[0]['document_uid']
return result[0]['document_uid']
return None
@property
def due_date(self) -> Optional[datetime]:
"""Get date when approval should be completed."""
return self._data.get('dueDate')
@due_date.setter
def due_date(self, date: datetime) -> None:
"""Set date when approval should be completed."""
self._data['dueDate'] = date
# Update in database
db.update_node(self.uid, {'dueDate': date})
@property
def completion_date(self) -> Optional[datetime]:
"""Get date when approval was completed."""
return self._data.get('completionDate')
@completion_date.setter
def completion_date(self, date: datetime) -> None:
"""Set date when approval was completed."""
self._data['completionDate'] = date
# Update in database
db.update_node(self.uid, {'completionDate': date})
@property
def created_at(self) -> Optional[datetime]:
"""Get date when approval cycle was created."""
return self._data.get('createdAt')
@property
def is_complete(self) -> bool:
"""Check if approval cycle is complete."""
return self.status in ['APPROVED', 'REJECTED', 'CANCELLED']
@property
def is_active(self) -> bool:
"""Check if approval cycle is active."""
return self.status in ['PENDING', 'IN_APPROVAL']
@property
def is_overdue(self) -> bool:
"""Check if approval cycle is overdue."""
if not self.due_date or self.is_complete:
return False
return datetime.now() > self.due_date
@property
def instructions(self) -> str:
"""Get instructions for approvers."""
return self._data.get('instructions', '')
@property
def approver_count(self) -> int:
"""Get number of approvers."""
if not hasattr(self, '_approver_count'):
result = db.run_query(
"""
MATCH (a:ApprovalCycle {UID: $uid})<-[:ASSIGNMENT]-(aa:ApproverAssignment)
WHERE aa.removalDate IS NULL
RETURN count(aa) as count
""",
{"uid": self.uid}
)
self._approver_count = result[0]['count'] if result else 0
return self._approver_count
def add_approver(self, approver: Union[DocUser, str], instructions: Optional[str] = None) -> bool:
"""Add an approver to the approval cycle."""
try:
approver_uid = approver.uid if isinstance(approver, DocUser) else approver
# Check if approver is already assigned
if self.is_approver(approver_uid):
logger.warning(f"Approver {approver_uid} is already assigned to approval cycle {self.uid}")
return False
# Create approver assignment
result = ApproverAssignment.create(
self.uid,
approver_uid,
instructions=instructions
)
# Reset cache
self._approvers_cache = None
if hasattr(self, '_approver_count'):
delattr(self, '_approver_count')
return result is not None
except Exception as e:
logger.error(f"Error adding approver {approver} to approval cycle {self.uid}: {e}")
return False
def remove_approver(self, approver: Union[DocUser, str]) -> bool:
"""Remove an approver from the approval cycle."""
try:
approver_uid = approver.uid if isinstance(approver, DocUser) else approver
# Get approver assignment
assignment = self.get_approver_assignment(approver_uid)
if not assignment:
logger.warning(f"Approver {approver_uid} is not assigned to approval cycle {self.uid}")
return False
# Mark as removed
assignment.removal_date = datetime.now()
# Reset cache
self._approvers_cache = None
if hasattr(self, '_approver_count'):
delattr(self, '_approver_count')
return True
except Exception as e:
logger.error(f"Error removing approver {approver} from approval cycle {self.uid}: {e}")
return False
@property
def comments(self) -> List[ApprovalComment]:
"""Get comments for this approval cycle."""
if self._comments_cache is None:
try:
result = db.run_query(
"""
MATCH (c:ApprovalComment)-[:COMMENTED_ON]->(a:ApprovalCycle {UID: $uid})
RETURN c
ORDER BY c.timestamp DESC
""",
{"uid": self.uid}
)
self._comments_cache = []
for record in result:
if 'c' in record and record['c']:
self._comments_cache.append(ApprovalComment(record['c']))
except Exception as e:
logger.error(f"Error getting comments for approval cycle {self.uid}: {e}")
self._comments_cache = []
return self._comments_cache
def add_comment(self, commenter: Union[DocUser, str],
text: str,
requires_resolution: bool = False) -> Optional[ApprovalComment]:
"""Add a comment to the approval cycle."""
comment = ApprovalComment.create(
self.uid,
commenter,
text,
requires_resolution
)
# Reset comments cache
self._comments_cache = None
return comment
def get_unresolved_comments(self) -> List[ApprovalComment]:
"""Get comments that require resolution."""
return [c for c in self.comments if c.requires_resolution and not c.is_resolved]
def complete_approval(self, approved: bool = True) -> bool:
"""Complete the approval cycle with decision."""
try:
# Check if there are unresolved comments that require resolution
if self.get_unresolved_comments():
logger.warning(f"Cannot complete approval cycle {self.uid} with unresolved comments")
return False
# Update status based on decision
self.status = 'APPROVED' if approved else 'REJECTED'
self.decision = 'APPROVED' if approved else 'REJECTED'
self.completion_date = datetime.now()
return True
except Exception as e:
logger.error(f"Error completing approval cycle {self.uid}: {e}")
return False
def get_approval_status(self, approver: Union[DocUser, str]) -> Dict[str, Any]:
"""Get approval status for a specific approver."""
approver_uid = approver.uid if isinstance(approver, DocUser) else approver
# Get approver assignment
assignment = self.get_approver_assignment(approver_uid)
if not assignment:
return {
'isApprover': False,
'status': 'NOT_ASSIGNED',
'canApprove': False
}
return {
'isApprover': True,
'status': assignment.status,
'canApprove': self.can_approve(approver_uid),
'decision': assignment.decision,
'decisionDate': assignment.decision_date,
'comments': assignment.decision_comments,
}
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary representation."""
result = super().to_dict()
# Add document information
document_version_uid = self.document_version_uid
if document_version_uid:
result['documentVersionUID'] = document_version_uid
document_uid = self.document_uid
if document_uid:
result['documentUID'] = document_uid
# Add approver assignments
assignments = self.get_approver_assignments()
result['approverAssignments'] = [a.to_dict() for a in assignments]
result['approverCount'] = len(assignments)
# Add completion metrics
result['isComplete'] = self.is_complete
result['isActive'] = self.is_active
result['isOverdue'] = self.is_overdue
return result
def get_approver_assignments(self) -> List['ApproverAssignment']:
"""Get all approver assignments for this cycle."""
if self._approvers_cache is None:
try:
result = db.run_query(
"""
MATCH (a:ApprovalCycle {UID: $uid})<-[:ASSIGNMENT]-(aa:ApproverAssignment)
WHERE aa.removalDate IS NULL
RETURN aa
ORDER BY aa.sequenceOrder ASC
""",
{"uid": self.uid}
)
self._approvers_cache = []
for record in result:
if 'aa' in record and record['aa']:
self._approvers_cache.append(ApproverAssignment(record['aa']))
except Exception as e:
logger.error(f"Error getting approver assignments for cycle {self.uid}: {e}")
self._approvers_cache = []
return self._approvers_cache
def get_approver_assignment(self, approver_uid: str) -> Optional['ApproverAssignment']:
"""Get assignment for a specific approver."""
for assignment in self.get_approver_assignments():
if assignment.approver_uid == approver_uid:
return assignment
return None
def is_approver(self, approver_uid: str) -> bool:
"""Check if a user is an approver for this cycle."""
return self.get_approver_assignment(approver_uid) is not None
def get_next_approver(self, current_sequence: int) -> Optional['ApproverAssignment']:
"""Get next approver in sequence."""
if not self._data.get('sequential', False):
return None
for assignment in self.get_approver_assignments():
if assignment.sequence_order > current_sequence and assignment.status == 'PENDING':
return assignment
return None
@property
def approval_type(self) -> str:
"""Get approval cycle type."""
return self._data.get('approvalType', 'STANDARD')
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 an 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 approve 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: Parameterdocument_version_uid: Type: strapprovers: Type: List[Union[DocUser, str]]due_date: Type: Optional[datetime]instructions: Type: strproperties: 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.
Parameters:
cls: Parameterdocument_uid: Type: str
Returns: Returns List['ApprovalCycle']
status(self) -> str
property
Purpose: Get approval cycle status.
Returns: Returns str
status(self, new_status) -> None
Purpose: Set approval cycle status.
Parameters:
new_status: Type: str
Returns: Returns None
decision(self) -> Optional[str]
property
Purpose: Get approval cycle decision.
Returns: Returns Optional[str]
decision(self, value) -> None
Purpose: Set approval cycle decision.
Parameters:
value: Type: str
Returns: Returns None
initiated_by_uid(self) -> Optional[str]
property
Purpose: Get UID of user who initiated the approval cycle.
Returns: Returns Optional[str]
initiated_by_uid(self, uid) -> None
Purpose: Set UID of user who initiated the approval cycle.
Parameters:
uid: Type: str
Returns: Returns None
required_approval_percentage(self) -> int
property
Purpose: Get required percentage of approvers that must approve.
Returns: Returns int
required_approval_percentage(self, value) -> None
Purpose: Set required percentage of approvers that must approve.
Parameters:
value: Type: int
Returns: Returns None
can_approve(self, approver_uid) -> bool
Purpose: Check if a user can approve in the current state.
Parameters:
approver_uid: Type: str
Returns: Returns bool
save(self) -> bool
Purpose: Save changes to database.
Returns: Returns bool
document_version_uid(self) -> Optional[str]
property
Purpose: Get UID of document version associated with this approval cycle.
Returns: Returns Optional[str]
document_uid(self) -> Optional[str]
property
Purpose: Get UID of document associated with this approval cycle.
Returns: Returns Optional[str]
due_date(self) -> Optional[datetime]
property
Purpose: Get date when approval should be completed.
Returns: Returns Optional[datetime]
due_date(self, date) -> None
Purpose: Set date when approval should be completed.
Parameters:
date: Type: datetime
Returns: Returns None
completion_date(self) -> Optional[datetime]
property
Purpose: Get date when approval was completed.
Returns: Returns Optional[datetime]
completion_date(self, date) -> None
Purpose: Set date when approval was completed.
Parameters:
date: Type: datetime
Returns: Returns None
created_at(self) -> Optional[datetime]
property
Purpose: Get date when approval cycle was created.
Returns: Returns Optional[datetime]
is_complete(self) -> bool
property
Purpose: Check if approval cycle is complete.
Returns: Returns bool
is_active(self) -> bool
property
Purpose: Check if approval cycle is active.
Returns: Returns bool
is_overdue(self) -> bool
property
Purpose: Check if approval cycle is overdue.
Returns: Returns bool
instructions(self) -> str
property
Purpose: Get instructions for approvers.
Returns: Returns str
approver_count(self) -> int
property
Purpose: Get number of approvers.
Returns: Returns int
add_approver(self, approver, instructions) -> bool
Purpose: Add an approver to the approval cycle.
Parameters:
approver: Type: Union[DocUser, str]instructions: Type: Optional[str]
Returns: Returns bool
remove_approver(self, approver) -> bool
Purpose: Remove an approver from the approval cycle.
Parameters:
approver: Type: Union[DocUser, str]
Returns: Returns bool
comments(self) -> List[ApprovalComment]
property
Purpose: Get 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.
Parameters:
commenter: Type: Union[DocUser, str]text: Type: strrequires_resolution: Type: bool
Returns: Returns Optional[ApprovalComment]
get_unresolved_comments(self) -> List[ApprovalComment]
Purpose: Get comments that require resolution.
Returns: Returns List[ApprovalComment]
complete_approval(self, approved) -> bool
Purpose: Complete the approval cycle with decision.
Parameters:
approved: Type: bool
Returns: Returns bool
get_approval_status(self, approver) -> Dict[str, Any]
Purpose: Get approval status for a specific approver.
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 cycle.
Returns: Returns List['ApproverAssignment']
get_approver_assignment(self, approver_uid) -> Optional['ApproverAssignment']
Purpose: Get assignment for a specific approver.
Parameters:
approver_uid: Type: str
Returns: Returns Optional['ApproverAssignment']
is_approver(self, approver_uid) -> bool
Purpose: Check if a user is an approver for this cycle.
Parameters:
approver_uid: Type: str
Returns: Returns bool
get_next_approver(self, current_sequence) -> Optional['ApproverAssignment']
Purpose: Get next approver in sequence.
Parameters:
current_sequence: Type: int
Returns: Returns Optional['ApproverAssignment']
approval_type(self) -> str
property
Purpose: Get approval cycle type.
Returns: Returns str
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
Similar Components
AI-powered semantic similarity - components with related functionality:
-
class ApprovalCycle_v1 97.5% similar
-
class ReviewCycle 80.6% similar
-
function create_approval_cycle_v1 75.4% similar
-
class ApproverAssignment_v1 67.7% similar
-
class ApproverAssignment 67.2% similar