class ApprovalController
Maturity: 28
Controller for managing document approval processes.
File:
/tf/active/vicechatdev/CDocs single class/controllers/approval_controller.py
Lines:
38 - 1942
38 - 1942
Complexity:
moderate
moderate
Purpose
Controller for managing document approval processes.
Source Code
class ApprovalController(ApprovalControllerBase):
"""Controller for managing document approval processes."""
@log_controller_action("create_cycle")
@require_permission(["CREATE_APPROVAL", "INITIATE_APPROVAL"])
def create_cycle(self,
document_version_uid: str,
user_uids: List[str],
due_date: Optional[datetime] = None,
instructions: str = '',
sequential: bool = True,
required_approval_percentage: int = 100,
initiated_by_uid: Optional[str] = None,
notify_users: bool = True,
approval_type: str = "STANDARD") -> Optional[Dict[str, Any]]:
"""
Create a new approval cycle for a document version.
Args:
document_version_uid: UID of the document version to approve
user_uids: List of user UIDs to assign as approvers
due_date: Optional due date (default: based on settings)
instructions: Instructions for approvers
sequential: Whether approval should be sequential (default: True)
required_approval_percentage: Percentage of approvals required
initiated_by_uid: UID of the user who initiated the approval
notify_users: Whether to send notifications to approvers
approval_type: Type of approval (e.g., "STANDARD", "REGULATORY")
Returns:
Dictionary with the created approval cycle information or None if failed
"""
try:
# Validate document version exists
document_version = DocumentVersion(uid=document_version_uid)
if not document_version:
raise ResourceNotFoundError(f"Document version not found: {document_version_uid}")
# Get associated document
document = document_version.document
if not document:
raise ResourceNotFoundError("Document not found for version")
# Check if document status allows approval
allowed_statuses = ["DRAFT", "IN_REVIEW", "APPROVED"]
if document.status not in allowed_statuses:
raise BusinessRuleError(f"Cannot start approval for document with status {document.status}")
# Validate approval type
if approval_type not in settings.APPROVAL_WORKFLOW_TYPES:
raise ValidationError(f"Invalid approval type: {approval_type}")
# Validate approver list
if not user_uids or len(user_uids) == 0:
raise ValidationError("At least one approver must be specified")
# Validate approval percentage
if required_approval_percentage < 1 or required_approval_percentage > 100:
raise ValidationError("Required approval percentage must be between 1 and 100")
# Set due date if not provided
if not due_date:
# Default to settings-defined approval period (e.g., 14 days)
approval_period_days = getattr(settings, 'DEFAULT_APPROVAL_PERIOD_DAYS', 14)
due_date = datetime.now() + timedelta(days=approval_period_days)
# Create approval cycle with proper document version instance
approval_cycle = ApprovalCycle.create(
document_version_uid=document_version.uid,
approvers=user_uids,
due_date=due_date,
instructions=instructions or "",
sequential=sequential,
properties={
"approval_type": approval_type,
"required_approval_percentage": required_approval_percentage,
"initiated_by_uid": initiated_by_uid,
"initiated_by_name": DocUser(uid=initiated_by_uid).name if initiated_by_uid else None
}
)
if not approval_cycle:
raise BusinessRuleError("Failed to create approval cycle")
# Log approval cycle creation
audit_trail.log_event(
event_type="APPROVAL_STARTED",
user=DocUser(uid=initiated_by_uid) if initiated_by_uid else None,
resource_uid=approval_cycle.uid,
resource_type="ApprovalCycle",
details={
"document_uid": document.uid,
"version_uid": document_version.uid,
"approver_uids": user_uids,
"due_date": due_date.isoformat() if due_date else None,
"approval_type": approval_type,
"sequential": sequential
}
)
# Notify approvers if requested
if notify_users:
try:
self._send_approver_notifications(
approval_cycle=approval_cycle,
document=document,
approver_uids=user_uids
)
except Exception as e:
logger.error(f"Error sending approver notifications: {e}")
# Continue even if notifications fail
# Update document status
document_version.status = "IN_APPROVAL"
document.status = "IN_APPROVAL"
# Format output
result = approval_cycle.to_dict()
# Add document information for convenience
result['document'] = {
"uid": document.uid,
"doc_number": document.doc_number,
"title": document.title,
"status": document.status
}
return {
"success": True,
"approval_cycle": result,
"message": "Approval cycle created successfully"
}
except (ResourceNotFoundError, ValidationError, PermissionError, BusinessRuleError) as e:
# Re-raise known errors
raise
except Exception as e:
logger.error(f"Error creating approval cycle: {e}")
logger.error(traceback.format_exc())
raise BusinessRuleError(f"Failed to create approval cycle: {e}")
@log_controller_action("get_cycle_by_uid")
def get_cycle_by_uid(self, cycle_uid: str) -> Optional[Dict[str, Any]]:
"""
Get detailed information about an approval cycle.
Args:
cycle_uid: UID of the approval cycle
Returns:
Dictionary with approval cycle information or None if not found
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Get basic approval data
result = approval_cycle.to_dict()
# Add document version information
document_version_uid = approval_cycle.document_version_uid
if document_version_uid:
document_version = DocumentVersion(uid=document_version_uid)
if document_version:
document = document_version.document
if document:
result['document'] = {
'uid': document.uid,
'doc_number': document.doc_number,
'title': document.title,
'status': document.status
}
result['document_version'] = {
'uid': document_version.uid,
'version_number': document_version.version_number,
'status': document_version.status
}
# Add approver assignments
approver_assignments = approval_cycle.get_approver_assignments()
if approver_assignments:
result['approver_assignments'] = [
assignment.to_dict() for assignment in approver_assignments
]
# Add comments
comments = approval_cycle.comments
if comments:
result['comments'] = [comment.to_dict() for comment in comments]
return result
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving approval cycle {cycle_uid}: {e}")
logger.error(traceback.format_exc())
raise BusinessRuleError(f"Failed to retrieve approval cycle: {e}")
@log_controller_action("get_cycles_for_document")
def get_cycles_for_document(self, document_uid: str,
include_active_only: bool = False) -> List[Dict[str, Any]]:
"""
Get all approval cycles for a document.
Args:
document_uid: UID of the document
include_active_only: Whether to include only active approval cycles
Returns:
List of dictionaries with cycle information
"""
try:
# Check if document exists
document = ControlledDocument(uid=document_uid)
if not document:
raise ResourceNotFoundError(f"Document not found: {document_uid}")
# Get all approval cycles
approval_cycles = ApprovalCycle.get_approvals_for_document(document_uid)
# Filter by status if needed
if include_active_only:
approval_cycles = [ac for ac in approval_cycles
if ac.status in [WorkflowStatus.PENDING.value, WorkflowStatus.IN_PROGRESS.value]]
# Convert to dictionaries
result = [ac.to_dict() for ac in approval_cycles]
# Add document info to response
doc_info = {
"document_uid": document_uid,
"document_number": document.doc_number,
"document_title": document.title
}
# Return formatted response
return {
"success": True,
"document": doc_info,
"approval_cycles": result,
"count": len(result)
}
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving approval cycles for document {document_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve approval cycles: {e}")
@log_controller_action("get_cycles_for_document_version")
def get_cycles_for_document_version(self, version_uid: str) -> List[Dict[str, Any]]:
"""
Get all approval cycles for a document version.
Args:
version_uid: UID of the document version
Returns:
List of dictionaries with cycle information
"""
try:
# Check if document version exists
document_version = DocumentVersion(uid=version_uid)
if not document_version:
raise ResourceNotFoundError(f"Document version not found: {version_uid}")
# Query for approval cycles
query = """
MATCH (v:DocumentVersion {UID: $version_uid})
MATCH (a:Approval)-[:FOR_APPROVAL]->(v)
RETURN a.UID as uid
ORDER BY a.created_at DESC
"""
result = db.run_query(query, {"version_uid": version_uid})
# Get approval cycles
approval_cycles = []
for record in result:
if 'uid' in record:
approval_cycle = ApprovalCycle(uid=record['uid'])
if approval_cycle:
approval_cycles.append(approval_cycle.to_dict())
# Get document info
document = document_version.document
doc_info = {
"document_uid": document.uid if document else None,
"document_number": document.doc_number if document else None,
"document_title": document.title if document else None,
"version_uid": version_uid,
"version_number": document_version.version_number
}
# Return formatted response
return {
"success": True,
"document_version": doc_info,
"approval_cycles": approval_cycles,
"count": len(approval_cycles)
}
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving approval cycles for version {version_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve approval cycles: {e}")
@log_controller_action("get_current_cycle_for_document")
def get_current_cycle_for_document(self, document_uid: str) -> Optional[Dict[str, Any]]:
"""
Get the current active approval cycle for a document.
Args:
document_uid: UID of the document
Returns:
Dictionary with cycle information or None if no active cycle
"""
try:
# Query for the most recent active approval cycle
query = """
MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
MATCH (a:Approval)-[:FOR_APPROVAL]->(v)
WHERE a.status IN ['PENDING', 'IN_PROGRESS']
RETURN a.UID as uid, v.UID as version_uid, v.version_number as version_number
ORDER BY a.created_at DESC
LIMIT 1
"""
result = db.run_query(query, {"doc_uid": document_uid})
if not result or len(result) == 0:
return None
approval_uid = result[0].get('uid')
if not approval_uid:
return None
# Get full approval cycle details
approval_data = self.get_cycle_by_uid(approval_uid)
# Add version info to response
approval_data['version'] = {
'uid': result[0].get('version_uid'),
'version_number': result[0].get('version_number')
}
return approval_data
except Exception as e:
logger.error(f"Error retrieving current approval cycle for document {document_uid}: {e}")
return None
@log_controller_action("get_user_pending_approvals")
def get_user_pending_approvals(self, user_uid: str, include_completed: bool = False) -> Dict[str, Any]:
"""
Get pending approvals for a user.
Args:
user_uid: UID of the user
include_completed: Whether to include completed approvals
Returns:
Dictionary with pending approvals information
"""
try:
status_filter = ["PENDING", "IN_PROGRESS"]
if include_completed:
status_filter.extend(["COMPLETED", "REJECTED"])
assignments, total = self.get_assignments_for_user(
user_uid=user_uid,
status_filter=status_filter,
limit=100,
offset=0
)
return {"success": True, "approvals": assignments, "total": total}
except Exception as e:
logger.error(f"Error getting pending approvals for user {user_uid}: {e}")
logger.error(traceback.format_exc())
return {"success": False, "error": str(e), "approvals": [], "total": 0}
@log_controller_action("get_assignments_for_cycle")
def get_assignments_for_cycle(self, cycle_uid: str) -> List[Dict[str, Any]]:
"""
Get all approver assignments for an approval cycle.
Args:
cycle_uid: UID of the approval cycle
Returns:
List of dictionaries with assignment information
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Get approver assignments
assignments = approval_cycle.get_approver_assignments()
# Convert to dictionaries
assignment_dicts = [assignment.to_dict() for assignment in assignments]
# Add user information to each assignment
for assignment_dict in assignment_dicts:
approver_uid = assignment_dict.get('approver_uid') or assignment_dict.get('user_uid')
if approver_uid:
user = DocUser(uid=approver_uid)
if user:
assignment_dict['user'] = {
'uid': user.uid,
'name': user.name,
'email': user.email,
'department': user.department
}
return assignment_dicts
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving assignments for approval cycle {cycle_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve assignments: {e}")
@log_controller_action("get_assignment_by_uid")
def get_assignment_by_uid(self, assignment_uid: str) -> Optional[Dict[str, Any]]:
"""
Get approver assignment information by UID.
Args:
assignment_uid: UID of the assignment
Returns:
Dictionary with assignment information or None if not found
"""
try:
# Query for assignment
assignment_data = db.get_node_by_uid(assignment_uid)
if not assignment_data:
raise ResourceNotFoundError(f"Assignment not found: {assignment_uid}")
# Create ApproverAssignment instance
assignment = ApproverAssignment(assignment_data)
# Get associated approval cycle
approval_cycle = None
if assignment.approval_cycle_uid:
approval_cycle = ApprovalCycle(uid=assignment.approval_cycle_uid)
# Convert assignment to dict
result = assignment.to_dict()
# Add approval cycle information
if approval_cycle:
result['approval_cycle'] = {
'uid': approval_cycle.uid,
'status': approval_cycle.status,
'started_at': approval_cycle.started_at,
'due_date': approval_cycle.due_date
}
# Add document information if available
document_version = approval_cycle.document_version
if document_version:
document = document_version.document
if document:
result['document'] = {
'uid': document.uid,
'doc_number': document.doc_number,
'title': document.title
}
# Add user information
approver_uid = assignment.approver_uid
if approver_uid:
user = DocUser(uid=approver_uid)
if user:
result['user'] = {
'uid': user.uid,
'name': user.name,
'email': user.email
}
return result
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving assignment {assignment_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve assignment: {e}")
@log_controller_action("get_assignment_for_user")
def get_assignment_for_user(self, cycle_uid: str, user_uid: str) -> Optional[Dict[str, Any]]:
"""
Get assignment for a specific approver in an approval cycle.
Args:
cycle_uid: UID of the approval cycle
user_uid: UID of the approver
Returns:
Dictionary with assignment information or None if not found
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Check if user is an approver
if not approval_cycle.is_approver(user_uid):
return None
# Get approver assignment
assignment = approval_cycle.get_approver_assignment(user_uid)
if not assignment:
return None
# Convert to dictionary
return assignment.to_dict()
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving assignment for user {user_uid} in cycle {cycle_uid}: {e}")
return None
@log_controller_action("get_assignments_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 approval 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)
"""
try:
# Normalize status filter
if isinstance(status_filter, str):
status_filter = [status_filter]
# Construct status filter condition for query
status_condition = ""
if status_filter and len(status_filter) > 0:
status_list = ", ".join([f"'{status}'" for status in status_filter])
status_condition = f"AND a.status IN [{status_list}]"
# Build query
query = f"""
MATCH (a:Approver)
WHERE a.approver_uid = $user_uid {status_condition}
RETURN a.UID as uid
ORDER BY a.assigned_at DESC
SKIP $offset
LIMIT $limit
"""
# Execute query
result = db.run_query(query, {
"user_uid": user_uid,
"offset": offset,
"limit": limit
})
# Get assignments
assignments = []
for record in result:
if 'uid' in record:
assignment = self.get_assignment_by_uid(record['uid'])
if assignment:
assignments.append(assignment)
# Count total assignments
count_query = f"""
MATCH (a:Approver)
WHERE a.approver_uid = $user_uid {status_condition}
RETURN count(a) as count
"""
count_result = db.run_query(count_query, {"user_uid": user_uid})
total_count = count_result[0]['count'] if count_result else 0
return assignments, total_count
except Exception as e:
logger.error(f"Error retrieving approval assignments for user {user_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve approval assignments: {e}")
@log_controller_action("update_cycle_status")
@require_permission(["MANAGE_APPROVALS"])
def update_cycle_status(self, cycle_uid: str, status: str) -> bool:
"""
Update the status of an approval cycle.
Args:
cycle_uid: UID of the approval cycle
status: New status value
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Validate the status
if status not in [s.value for s in WorkflowStatus]:
raise ValidationError(f"Invalid status: {status}")
# Update status
approval_cycle.status = status
# Log status change
audit_trail.log_event(
event_type="APPROVAL_STATUS_CHANGE",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"previous_status": approval_cycle.status,
"new_status": status
}
)
return True
except (ResourceNotFoundError, ValidationError):
raise
except Exception as e:
logger.error(f"Error updating approval cycle status: {e}")
raise BusinessRuleError(f"Failed to update approval cycle status: {e}")
@log_controller_action("complete_cycle")
def complete_cycle(self, cycle_uid: str,
decision: str = None,
comment: Optional[str] = None) -> bool:
"""
Complete an approval cycle with a decision.
Args:
cycle_uid: UID of the approval cycle
decision: Decision value (e.g., 'APPROVED' or 'REJECTED')
comment: Optional comment about the decision
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Validate decision if provided
if decision and decision not in settings.APPROVAL_DECISIONS:
raise ValidationError(f"Invalid approval decision: {decision}")
# Check if all required approvers have completed their assignments
assignments = approval_cycle.get_approver_assignments()
# For sequential approval, verify all steps are complete
if approval_cycle.sequential:
incomplete_assignments = [a for a in assignments
if a.status not in [AssignmentStatus.COMPLETED.value,
AssignmentStatus.REJECTED.value,
AssignmentStatus.SKIPPED.value]]
if incomplete_assignments:
raise BusinessRuleError("Cannot complete approval cycle: Some approvers have not completed their assignments")
# For parallel approval, verify required percentage is met
else:
total = len(assignments)
if total == 0:
raise BusinessRuleError("Cannot complete approval cycle: No approvers assigned")
completed = sum(1 for a in assignments
if a.status == AssignmentStatus.COMPLETED.value)
completion_percentage = (completed / total) * 100
required_percentage = approval_cycle.required_approval_percentage
if completion_percentage < required_percentage:
raise BusinessRuleError(
f"Cannot complete approval cycle: Only {completion_percentage:.0f}% approved "
f"(requires {required_percentage:.0f}%)"
)
# Check for rejections
rejections = [a for a in assignments if a.status == AssignmentStatus.REJECTED.value]
# If rejection exists and decision is not specifically overriding, set to rejected
if rejections and (decision is None or decision != "APPROVED"):
decision = "REJECTED"
# Add final comment if provided
if comment:
approval_cycle.add_comment(
approver=DocUser(uid=approval_cycle.initiated_by_uid) if approval_cycle.initiated_by_uid else None,
text=comment
)
# Complete the cycle with the appropriate decision
is_approved = decision is None or decision == "APPROVED"
success = approval_cycle.complete_approval(is_approved)
if not success:
raise BusinessRuleError("Failed to complete approval cycle")
# Log approval completion
audit_trail.log_event(
event_type="APPROVAL_COMPLETED",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"decision": decision,
"has_comment": comment is not None
}
)
# Update document version status
document_version = approval_cycle.document_version
if document_version:
if decision == "APPROVED":
document_version.status = "APPROVED"
# Also update parent document status
document = document_version.document
if document:
document.status = "APPROVED"
else:
# Set back to draft if rejected
document_version.status = "DRAFT"
# Also update parent document status
document = document_version.document
if document:
document.status = "DRAFT"
return True
except (ResourceNotFoundError, ValidationError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error completing approval cycle: {e}")
raise BusinessRuleError(f"Failed to complete approval cycle: {e}")
@log_controller_action("cancel_cycle")
@require_permission(["MANAGE_APPROVALS", "INITIATE_APPROVAL"])
def cancel_cycle(self, cycle_uid: str, reason: Optional[str] = None) -> bool:
"""
Cancel an approval cycle.
Args:
cycle_uid: UID of the approval cycle
reason: Optional reason for cancellation
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Check if approval can be canceled
if approval_cycle.status in [WorkflowStatus.APPROVED.value, WorkflowStatus.REJECTED.value,
WorkflowStatus.CANCELED.value]:
raise BusinessRuleError(f"Cannot cancel approval cycle in {approval_cycle.status} state")
# Update status to CANCELED
approval_cycle.status = WorkflowStatus.CANCELED.value
# Add cancellation comment if reason provided
if reason:
approval_cycle.add_comment(
approver=DocUser(uid=approval_cycle.initiated_by_uid) if approval_cycle.initiated_by_uid else None,
text=f"Approval cycle canceled: {reason}"
)
# Log cancellation
audit_trail.log_event(
event_type="APPROVAL_CANCELED",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"reason": reason
}
)
# Restore document version to DRAFT status
document_version = approval_cycle.document_version
if document_version:
document_version.status = "DRAFT"
# Also update parent document status
document = document_version.document
if document:
document.status = "DRAFT"
return True
except (ResourceNotFoundError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error canceling approval cycle: {e}")
raise BusinessRuleError(f"Failed to cancel approval cycle: {e}")
@log_controller_action("add_participant")
@require_permission(["MANAGE_APPROVALS", "INITIATE_APPROVAL"])
def add_participant(self, cycle_uid: str,
user_uid: str,
sequence_order: int = 0) -> bool:
"""
Add an approver to an approval cycle.
Args:
cycle_uid: UID of the approval cycle
user_uid: UID of the user to add
sequence_order: Order in the sequence (for sequential approvals)
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Check if user exists
user = DocUser(uid=user_uid)
if not user:
raise ResourceNotFoundError(f"User not found: {user_uid}")
# Check if already an approver
if approval_cycle.is_approver(user_uid):
return True # Already an approver, consider successful
# Add approver
success = approval_cycle.add_approver(user, sequence_order)
if not success:
raise BusinessRuleError("Failed to add approver to approval cycle")
# Log approver addition
audit_trail.log_event(
event_type="APPROVER_ADDED",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"user_uid": user_uid,
"user_name": user.name,
"sequence_order": sequence_order
}
)
# Send notification to the new approver
try:
# Get document info for notification
document_version = approval_cycle.document_version
document = document_version.document if document_version else None
if document:
notification_data = {
'recipient_uid': user_uid,
'title': f"Approval Request: {document.title}",
'message': f"You have been assigned to approve {document.doc_number}: {document.title}",
'link_url': f"/document/{document.uid}/approval/{cycle_uid}",
'notification_type': "APPROVAL_REQUEST",
'source_uid': cycle_uid,
'priority': 'HIGH'
}
notifications.send_notification(notification_data)
except Exception as e:
logger.error(f"Error sending notification to new approver: {e}")
# Continue even if notification fails
return True
except (ResourceNotFoundError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error adding approver to approval cycle: {e}")
raise BusinessRuleError(f"Failed to add approver: {e}")
@log_controller_action("remove_participant")
@require_permission(["MANAGE_APPROVALS", "INITIATE_APPROVAL"])
def remove_participant(self, cycle_uid: str,
user_uid: str,
reason: Optional[str] = None) -> bool:
"""
Remove an approver from an approval cycle.
Args:
cycle_uid: UID of the approval cycle
user_uid: UID of the user to remove
reason: Optional reason for removal
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Check if user exists
user = DocUser(uid=user_uid)
if not user:
raise ResourceNotFoundError(f"User not found: {user_uid}")
# Check if user is an approver
if not approval_cycle.is_approver(user_uid):
return True # Not an approver, consider successful
# Get approver assignment before removal
assignment = approval_cycle.get_approver_assignment(user_uid)
if assignment and reason:
# Record removal reason in assignment
assignment.removal_reason = reason
assignment.removal_date = datetime.now()
assignment.save()
# Remove approver
success = approval_cycle.remove_approver(user)
if not success:
raise BusinessRuleError("Failed to remove approver from approval cycle")
# Log approver removal
audit_trail.log_event(
event_type="APPROVER_REMOVED",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"user_uid": user_uid,
"user_name": user.name,
"reason": reason
}
)
return True
except (ResourceNotFoundError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error removing approver from approval cycle: {e}")
raise BusinessRuleError(f"Failed to remove approver: {e}")
@log_controller_action("update_assignment_status")
@require_permission(["APPROVE_DOCUMENT", "MANAGE_APPROVALS"])
def update_assignment_status(self, assignment_uid: str, status: str) -> bool:
"""
Update the status of an approver assignment.
Args:
assignment_uid: UID of the approver assignment
status: New status value
Returns:
Boolean indicating success
"""
try:
# Get ApproverAssignment instance
assignment = ApproverAssignment(uid=assignment_uid)
if not assignment:
raise ResourceNotFoundError(f"Approver assignment not found: {assignment_uid}")
# Validate the status
if status not in [s.value for s in AssignmentStatus]:
raise ValidationError(f"Invalid status: {status}")
# Update status
old_status = assignment.status
assignment.status = status
# Log status change
audit_trail.log_event(
event_type="APPROVAL_ASSIGNMENT_STATUS_CHANGE",
resource_uid=assignment_uid,
resource_type="ApproverAssignment",
details={
"previous_status": old_status,
"new_status": status
}
)
# Update parent approval cycle status if needed
approval_cycle_uid = assignment.approval_cycle_uid
if approval_cycle_uid:
approval_cycle = ApprovalCycle(uid=approval_cycle_uid)
if approval_cycle:
approval_cycle.update_status()
return True
except (ResourceNotFoundError, ValidationError):
raise
except Exception as e:
logger.error(f"Error updating approver assignment status: {e}")
raise BusinessRuleError(f"Failed to update assignment status: {e}")
@log_controller_action("update_assignment_status")
@require_permission(["APPROVE_DOCUMENT", "MANAGE_APPROVALS"])
def update_assignment_status(self, assignment_uid: str, status: str) -> bool:
"""
Update the status of an approver assignment.
Args:
assignment_uid: UID of the approver assignment
status: New status value
Returns:
Boolean indicating success
"""
try:
# Get ApproverAssignment instance
assignment = ApproverAssignment(uid=assignment_uid)
if not assignment:
raise ResourceNotFoundError(f"Approver assignment not found: {assignment_uid}")
# Validate the status
if status not in [s.value for s in AssignmentStatus]:
raise ValidationError(f"Invalid status: {status}")
# Update status
old_status = assignment.status
assignment.status = status
# Log status change
audit_trail.log_event(
event_type="APPROVAL_ASSIGNMENT_STATUS_CHANGE",
resource_uid=assignment_uid,
resource_type="ApproverAssignment",
details={
"previous_status": old_status,
"new_status": status
}
)
# Update parent approval cycle status if needed
approval_cycle_uid = assignment.approval_cycle_uid
if approval_cycle_uid:
approval_cycle = ApprovalCycle(uid=approval_cycle_uid)
if approval_cycle:
approval_cycle.update_status()
return True
except (ResourceNotFoundError, ValidationError):
raise
except Exception as e:
logger.error(f"Error updating approver assignment status: {e}")
raise BusinessRuleError(f"Failed to update assignment status: {e}")
@log_controller_action("complete_assignment")
@require_permission(["APPROVE_DOCUMENT"])
def complete_assignment(self, assignment_uid: str,
decision: str,
comments: Optional[str] = None) -> bool:
"""
Complete an approver assignment with a decision.
Args:
assignment_uid: UID of the approver assignment
decision: Decision value (e.g., 'APPROVED', 'REJECTED')
comments: Optional comments about the decision
Returns:
Boolean indicating success
"""
try:
# Get ApproverAssignment instance
assignment = ApproverAssignment(uid=assignment_uid)
if not assignment:
raise ResourceNotFoundError(f"Approver assignment not found: {assignment_uid}")
# Validate the decision
if decision not in settings.APPROVAL_DECISIONS:
raise ValidationError(f"Invalid approval decision: {decision}")
# Check if assignment already completed
if assignment.status in [AssignmentStatus.COMPLETED.value, AssignmentStatus.REJECTED.value]:
raise BusinessRuleError("Assignment already completed")
# For sequential approval, check if this is the current step
approval_cycle = ApprovalCycle(uid=assignment.approval_cycle_uid) if assignment.approval_cycle_uid else None
if approval_cycle and approval_cycle.sequential:
sequence_order = assignment.sequence_order or 0
if sequence_order > approval_cycle.current_sequence:
raise BusinessRuleError(f"Cannot complete assignment: Previous steps must be completed first")
# Set decision and comments
assignment.decision = decision
if comments:
assignment.decision_comments = comments
# Set decision date
assignment.decision_date = datetime.now()
# Update status based on decision
if decision == "REJECTED":
assignment.status = AssignmentStatus.REJECTED.value
else:
assignment.status = AssignmentStatus.COMPLETED.value
# Save changes
if not assignment.save():
raise BusinessRuleError("Failed to save approver assignment")
# Add comment to approval cycle if provided
if comments:
if approval_cycle:
# Add comment to cycle
approval_cycle.add_comment(
approver=assignment.approver_uid,
text=comments
)
# Log assignment completion
audit_trail.log_event(
event_type="APPROVAL_ASSIGNMENT_COMPLETED",
resource_uid=assignment_uid,
resource_type="ApproverAssignment",
details={
"decision": decision,
"has_comments": comments is not None,
"approval_cycle_uid": assignment.approval_cycle_uid
}
)
# Update parent approval cycle status
if approval_cycle:
approval_cycle.update_status()
# If sequential and this step is approved, activate next step
if approval_cycle.sequential and decision != "REJECTED":
next_approver = approval_cycle.get_next_approver(sequence_order)
if next_approver:
# Start the next assignment
next_approver.status = AssignmentStatus.PENDING.value
next_approver.save()
# Update cycle's current sequence
approval_cycle.current_sequence = next_approver.sequence_order
approval_cycle.save()
# Notify next approver
self.notify_participant(next_approver.uid)
return True
except (ResourceNotFoundError, ValidationError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error completing approver assignment: {e}")
raise BusinessRuleError(f"Failed to complete assignment: {e}")
@log_controller_action("add_comment")
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 an approval cycle.
Args:
cycle_uid: UID of the approval 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
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Check if user exists
user = DocUser(uid=user_uid)
if not user:
raise ResourceNotFoundError(f"User not found: {user_uid}")
# Check if parent comment exists if provided
parent_comment = None
if parent_comment_uid:
parent_comment = ApprovalComment(uid=parent_comment_uid)
if not parent_comment:
raise ResourceNotFoundError(f"Parent comment not found: {parent_comment_uid}")
# Prepare properties
properties = {
'comment_type': comment_type
}
# Add location if provided
if location:
if 'context' not in properties:
properties['context'] = {}
properties['context']['location'] = location
if 'page' in location:
properties['page'] = location['page']
# Add parent comment reference if provided
if parent_comment:
properties['parent_comment_uid'] = parent_comment_uid
# Create the comment
comment = approval_cycle.add_comment(
approver=user,
text=text,
requires_resolution=requires_resolution,
properties=properties
)
if not comment:
raise BusinessRuleError("Failed to create approval comment")
# Log comment creation
audit_trail.log_event(
event_type="APPROVAL_COMMENT_ADDED",
user=user,
resource_uid=comment.uid,
resource_type="ApprovalComment",
details={
"approval_cycle_uid": cycle_uid,
"requires_resolution": requires_resolution,
"is_reply": parent_comment is not None
}
)
return comment.to_dict()
except (ResourceNotFoundError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error adding approval comment: {e}")
raise BusinessRuleError(f"Failed to add comment: {e}")
@log_controller_action("get_comments")
def get_comments(self, cycle_uid: str) -> List[Dict[str, Any]]:
"""
Get all comments for an approval cycle.
Args:
cycle_uid: UID of the approval cycle
Returns:
List of dictionaries with comment information
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Get comments
comments = approval_cycle.comments
# Convert to dictionaries
return [comment.to_dict() for comment in comments]
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving comments for approval cycle {cycle_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve comments: {e}")
@log_controller_action("resolve_comment")
def resolve_comment(self, comment_uid: str,
resolution_text: str) -> bool:
"""
Resolve an approval comment.
Args:
comment_uid: UID of the approval comment
resolution_text: Resolution text
Returns:
Boolean indicating success
"""
try:
# Get ApprovalComment instance
comment = ApprovalComment(uid=comment_uid)
if not comment:
raise ResourceNotFoundError(f"Approval comment not found: {comment_uid}")
# Check if comment requires resolution
if not comment.requires_resolution:
logger.warning(f"Comment {comment_uid} does not require resolution, but resolving anyway")
# Check if already resolved
if comment.is_resolved:
logger.warning(f"Comment {comment_uid} is already resolved")
return True
# Set resolution
comment.resolution = resolution_text
# Log resolution
audit_trail.log_event(
event_type="APPROVAL_COMMENT_RESOLVED",
resource_uid=comment_uid,
resource_type="ApprovalComment",
details={
"resolution_provided": bool(resolution_text),
"approval_cycle_uid": comment.approval_cycle_uid
}
)
return True
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error resolving approval comment: {e}")
raise BusinessRuleError(f"Failed to resolve comment: {e}")
@log_controller_action("get_comment_by_uid")
def get_comment_by_uid(self, comment_uid: str) -> Optional[Dict[str, Any]]:
"""
Get approval comment information by UID.
Args:
comment_uid: UID of the approval comment
Returns:
Dictionary with comment information or None if not found
"""
try:
# Get ApprovalComment instance
comment = ApprovalComment(uid=comment_uid)
if not comment:
raise ResourceNotFoundError(f"Approval comment not found: {comment_uid}")
# Convert to dictionary
return comment.to_dict()
except ResourceNotFoundError:
raise
except Exception as e:
logger.error(f"Error retrieving approval comment {comment_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve comment: {e}")
@log_controller_action("update_due_date")
@require_permission(["MANAGE_APPROVALS", "INITIATE_APPROVAL"])
def update_due_date(self, cycle_uid: str, due_date: datetime) -> bool:
"""
Update the due date for an approval cycle.
Args:
cycle_uid: UID of the approval cycle
due_date: New due date
Returns:
Boolean indicating success
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
raise ResourceNotFoundError(f"Approval cycle not found: {cycle_uid}")
# Validate due date
if due_date < datetime.now():
raise ValidationError("Due date cannot be in the past")
# Check if approval can be updated
if approval_cycle.status in [WorkflowStatus.APPROVED.value, WorkflowStatus.REJECTED.value,
WorkflowStatus.CANCELED.value]:
raise BusinessRuleError(f"Cannot update due date for approval in {approval_cycle.status} status")
# Update due date
old_due_date = approval_cycle.due_date
approval_cycle.due_date = due_date
# Log due date change
audit_trail.log_event(
event_type="APPROVAL_DUE_DATE_CHANGED",
resource_uid=cycle_uid,
resource_type="ApprovalCycle",
details={
"previous_due_date": old_due_date.isoformat() if old_due_date else None,
"new_due_date": due_date.isoformat()
}
)
return True
except (ResourceNotFoundError, ValidationError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error updating approval due date: {e}")
raise BusinessRuleError(f"Failed to update due date: {e}")
@log_controller_action("get_overdue_assignments")
def get_overdue_assignments(self) -> List[Dict[str, Any]]:
"""
Get all overdue approver assignments.
Returns:
List of dictionaries with overdue assignment information
"""
try:
# Query for overdue approver assignments
query = """
MATCH (a:Approver)
WHERE
a.due_date < datetime() AND
NOT a.status IN ['COMPLETED', 'REJECTED', 'SKIPPED']
RETURN a.UID as uid
"""
result = db.run_query(query)
# Get assignment details
overdue_assignments = []
for record in result:
if 'uid' in record:
assignment_data = self.get_assignment_by_uid(record['uid'])
if assignment_data:
overdue_assignments.append(assignment_data)
return overdue_assignments
except Exception as e:
logger.error(f"Error retrieving overdue assignments: {e}")
raise BusinessRuleError(f"Failed to retrieve overdue assignments: {e}")
@log_controller_action("get_approval_steps")
def get_approval_steps(self, cycle_uid: str) -> List[Dict[str, Any]]:
"""
Get all approval steps for an approval cycle.
Args:
cycle_uid: UID of the approval cycle
Returns:
List of dictionaries with approval step information
"""
try:
# Query for approval steps
query = """
MATCH (a:Approval {UID: $cycle_uid})-[:HAS_STEP]->(s:ApprovalStep)
RETURN s
ORDER BY s.sequence_order
"""
result = db.run_query(query, {"cycle_uid": cycle_uid})
# Process steps
steps = []
for record in result:
if 's' in record:
step_data = record['s']
# Get approvers for this step
approvers_query = """
MATCH (s:ApprovalStep {UID: $step_uid})-[:HAS_APPROVER]->(a:Approver)
RETURN a
"""
approvers_result = db.run_query(approvers_query, {"step_uid": step_data['UID']})
# Process approvers
approvers = []
for approver_record in approvers_result:
if 'a' in approver_record:
approver_data = approver_record['a']
# Get user information
user_uid = approver_data.get('approver_uid')
if user_uid:
user = DocUser(uid=user_uid)
if user:
approver_data['user'] = {
'uid': user.uid,
'name': user.name,
'email': user.email,
'department': user.department
}
approvers.append(approver_data)
# Add approvers to step
step_data['approvers'] = approvers
steps.append(step_data)
return steps
except Exception as e:
logger.error(f"Error retrieving approval steps for cycle {cycle_uid}: {e}")
raise BusinessRuleError(f"Failed to retrieve approval steps: {e}")
@log_controller_action("create_approval_workflow")
@require_permission(["MANAGE_SETTINGS", "MANAGE_APPROVALS"])
def create_approval_workflow(self, document_type: str,
workflow_type: str = 'STANDARD') -> Optional[Dict[str, Any]]:
"""
Create a new approval workflow template.
Args:
document_type: Type of document this workflow applies to
workflow_type: Type of workflow
Returns:
Dictionary with created workflow information or None if failed
"""
try:
# Validate document type
if document_type not in settings.DOCUMENT_TYPES:
raise ValidationError(f"Invalid document type: {document_type}")
# Validate workflow type
if workflow_type not in settings.APPROVAL_WORKFLOW_TYPES:
raise ValidationError(f"Invalid workflow type: {workflow_type}")
# Generate UID for workflow
workflow_uid = str(uuid.uuid4())
# Create workflow node
workflow_data = {
'UID': workflow_uid,
'document_type': document_type,
'workflow_type': workflow_type,
'name': f"{document_type} {workflow_type} Approval",
'description': f"Approval workflow for {document_type} documents",
'created_at': datetime.now().isoformat(),
'updated_at': datetime.now().isoformat(),
'status': 'ACTIVE'
}
success = db.create_node(NodeLabels.APPROVAL_WORKFLOW, workflow_data)
if not success:
raise BusinessRuleError("Failed to create approval workflow")
# Log workflow creation
audit_trail.log_event(
event_type="APPROVAL_WORKFLOW_CREATED",
resource_uid=workflow_uid,
resource_type="ApprovalWorkflow",
details={
"document_type": document_type,
"workflow_type": workflow_type
}
)
return workflow_data
except (ValidationError, BusinessRuleError):
raise
except Exception as e:
logger.error(f"Error creating approval workflow: {e}")
raise BusinessRuleError(f"Failed to create approval workflow: {e}")
@log_controller_action("get_approval_workflow_templates")
def get_approval_workflow_templates(self, document_type: Optional[str] = None) -> List[Dict[str, Any]]:
"""
Get approval workflow templates.
Args:
document_type: Optional document type to filter by
Returns:
List of dictionaries with workflow template information
"""
try:
# Prepare query
if document_type:
query = """
MATCH (w:ApprovalWorkflow)
WHERE w.document_type = $document_type
RETURN w
ORDER BY w.created_at DESC
"""
params = {"document_type": document_type}
else:
query = """
MATCH (w:ApprovalWorkflow)
RETURN w
ORDER BY w.created_at DESC
"""
params = {}
# Execute query
result = db.run_query(query, params)
# Process results
workflows = []
for record in result:
if 'w' in record:
workflows.append(record['w'])
return workflows
except Exception as e:
logger.error(f"Error retrieving approval workflow templates: {e}")
raise BusinessRuleError(f"Failed to retrieve workflow templates: {e}")
@log_controller_action("get_workflow_statistics")
def get_workflow_statistics(self) -> Dict[str, Any]:
"""
Get statistics about approval workflows.
Returns:
Dictionary with statistics information
"""
try:
current_year = datetime.now().year
# Query for approval cycle statistics
query = """
MATCH (a:Approval)
// Count by status
WITH a,
CASE WHEN a.status = 'PENDING' THEN 1 ELSE 0 END as pending,
CASE WHEN a.status = 'IN_PROGRESS' THEN 1 ELSE 0 END as in_progress,
CASE WHEN a.status = 'APPROVED' THEN 1 ELSE 0 END as approved,
CASE WHEN a.status = 'REJECTED' THEN 1 ELSE 0 END as rejected,
CASE WHEN a.status = 'CANCELED' THEN 1 ELSE 0 END as canceled,
CASE WHEN a.created_at >= datetime({year: $year}) THEN 1 ELSE 0 END as this_year,
CASE WHEN a.created_at >= datetime() - duration('P30D') THEN 1 ELSE 0 END as last_30_days
// Aggregate counts
RETURN
count(a) as total_approvals,
sum(pending) as pending_count,
sum(in_progress) as in_progress_count,
sum(approved) as approved_count,
sum(rejected) as rejected_count,
sum(canceled) as canceled_count,
sum(this_year) as this_year_count,
sum(last_30_days) as last_30_days_count
"""
result = db.run_query(query, {"year": current_year})
if not result:
return {
"total": 0,
"by_status": {
"PENDING": 0,
"IN_PROGRESS": 0,
"APPROVED": 0,
"REJECTED": 0,
"CANCELED": 0
},
"this_year": 0,
"last_30_days": 0,
"average_duration_days": 0
}
# Extract statistics
stats = result[0]
# Query for average duration (only completed approvals)
duration_query = """
MATCH (a:Approval)
WHERE a.status IN ['APPROVED', 'REJECTED']
AND a.created_at IS NOT NULL
AND a.completed_at IS NOT NULL
WITH a, duration.between(a.created_at, a.completed_at).days as duration_days
RETURN avg(duration_days) as avg_duration
"""
duration_result = db.run_query(duration_query)
avg_duration = duration_result[0].get('avg_duration', 0) if duration_result else 0
return {
"total": stats.get('total_approvals', 0),
"by_status": {
"PENDING": stats.get('pending_count', 0),
"IN_PROGRESS": stats.get('in_progress_count', 0),
"APPROVED": stats.get('approved_count', 0),
"REJECTED": stats.get('rejected_count', 0),
"CANCELED": stats.get('canceled_count', 0)
},
"this_year": stats.get('this_year_count', 0),
"last_30_days": stats.get('last_30_days_count', 0),
"average_duration_days": avg_duration
}
except Exception as e:
logger.error(f"Error retrieving approval statistics: {e}")
return {
"total": 0,
"by_status": {},
"error": str(e)
}
@log_controller_action("search_cycles")
def search_cycles(self, query: Dict[str, Any],
limit: int = 100,
offset: int = 0) -> Tuple[List[Dict[str, Any]], int]:
"""
Search for approval 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)
"""
try:
# Parse query parameters
document_uid = query.get('document_uid')
status = query.get('status')
initiated_by = query.get('initiated_by')
approver_uid = query.get('approver_uid')
date_from = query.get('date_from')
date_to = query.get('date_to')
doc_type = query.get('doc_type')
department = query.get('department')
# Build Cypher query
cypher_query = """
MATCH (a:Approval)
"""
# Add document filter if provided
if document_uid:
cypher_query += """
MATCH (d:ControlledDocument {UID: $document_uid})-[:HAS_VERSION]->(v:DocumentVersion)<-[:FOR_APPROVAL]-(a)
"""
# Add approver filter if provided
if approver_uid:
cypher_query += """
MATCH (a)-[:APPROVED_BY]->(u:User {UID: $approver_uid})
"""
# Add WHERE clause
conditions = []
if status:
conditions.append("a.status = $status")
if initiated_by:
conditions.append("a.initiated_by_uid = $initiated_by")
if date_from:
conditions.append("a.created_at >= $date_from")
if date_to:
conditions.append("a.created_at <= $date_to")
# Add document type and department filters if provided
if doc_type or department:
if not document_uid: # Only add this match if document_uid not already specified
cypher_query += """
MATCH (a)-[:FOR_APPROVAL]->(v:DocumentVersion)<-[:HAS_VERSION]-(d:ControlledDocument)
"""
if doc_type:
conditions.append("d.docType = $doc_type")
if department:
conditions.append("d.department = $department")
# Add conditions to query
if conditions:
cypher_query += "WHERE " + " AND ".join(conditions)
# Add return statement
cypher_query += """
RETURN a.UID as uid, a.status as status, a.created_at as created_at, a.due_date as due_date
ORDER BY a.created_at DESC
SKIP $skip
LIMIT $limit
"""
# Create parameters
params = {
'document_uid': document_uid,
'status': status,
'initiated_by': initiated_by,
'approver_uid': approver_uid,
'date_from': date_from.isoformat() if isinstance(date_from, datetime) else date_from,
'date_to': date_to.isoformat() if isinstance(date_to, datetime) else date_to,
'doc_type': doc_type,
'department': department,
'skip': offset,
'limit': limit
}
# Run query to get results
result = db.run_query(cypher_query, params)
# Fetch full approval cycles
approvals = []
for record in result:
approval_uid = record.get('uid')
if approval_uid:
approval_data = self.get_cycle_by_uid(approval_uid)
if approval_data:
approvals.append(approval_data)
# Run count query to get total
count_query = cypher_query.split("RETURN")[0] + "RETURN count(a) as total"
count_result = db.run_query(count_query, params)
total_count = count_result[0].get('total', 0) if count_result else 0
return approvals, total_count
except Exception as e:
logger.error(f"Error searching approval cycles: {e}")
return [], 0
@log_controller_action("is_user_participant")
def is_user_participant(self, cycle_uid: str, user_uid: str) -> bool:
"""
Check if a user is a participant in an approval cycle.
Args:
cycle_uid: UID of the approval cycle
user_uid: UID of the user to check
Returns:
Boolean indicating if user is a participant
"""
try:
# Get ApprovalCycle instance
approval_cycle = ApprovalCycle(uid=cycle_uid)
if not approval_cycle:
return False
# Check if user is an approver
return approval_cycle.is_approver(user_uid)
except Exception as e:
logger.error(f"Error checking if user is participant: {e}")
return False
def _send_approver_notifications(self, approval_cycle, document, approver_uids):
"""
Send notifications to approvers.
Args:
approval_cycle: ApprovalCycle instance
document: Document instance
approver_uids: List of approver UIDs
Returns:
None
"""
try:
for approver_uid in approver_uids:
# Get assignment for this approver
assignment = approval_cycle.get_approver_assignment(approver_uid)
if not assignment:
logger.warning(f"No assignment found for approver {approver_uid}")
continue
# Get user
user = DocUser(uid=approver_uid)
if not user:
logger.warning(f"User not found: {approver_uid}")
continue
# Create notification
notification_data = {
'recipient_uid': approver_uid,
'title': f"Approval Request: {document.title}",
'message': f"You have been assigned to approve {document.doc_number}: {document.title}",
'link_url': f"/document/{document.uid}/approval/{approval_cycle.uid}",
'notification_type': "APPROVAL_REQUEST",
'source_uid': approval_cycle.uid,
'priority': 'HIGH'
}
# Send notification
from CDocs.utils.notifications import send_notification
send_notification(notification_data)
except Exception as e:
logger.error(f"Error sending approver notifications: {e}")Parameters
| Name | Type | Default | Kind |
|---|---|---|---|
bases |
ApprovalControllerBase | - |
Parameter Details
bases: Parameter of type ApprovalControllerBase
Return Value
Returns unspecified type
Class Interface
Methods
create_cycle(self, document_version_uid, user_uids, due_date, instructions, sequential, required_approval_percentage, initiated_by_uid, notify_users, approval_type) -> Optional[Dict[str, Any]]
Purpose: Create a new approval cycle for a document version. Args: document_version_uid: UID of the document version to approve user_uids: List of user UIDs to assign as approvers due_date: Optional due date (default: based on settings) instructions: Instructions for approvers sequential: Whether approval should be sequential (default: True) required_approval_percentage: Percentage of approvals required initiated_by_uid: UID of the user who initiated the approval notify_users: Whether to send notifications to approvers approval_type: Type of approval (e.g., "STANDARD", "REGULATORY") Returns: Dictionary with the created approval cycle information or None if failed
Parameters:
document_version_uid: Type: struser_uids: Type: List[str]due_date: Type: Optional[datetime]instructions: Type: strsequential: Type: boolrequired_approval_percentage: Type: intinitiated_by_uid: Type: Optional[str]notify_users: Type: boolapproval_type: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_cycle_by_uid(self, cycle_uid) -> Optional[Dict[str, Any]]
Purpose: Get detailed information about an approval cycle. Args: cycle_uid: UID of the approval cycle Returns: Dictionary with approval cycle information or None if not found
Parameters:
cycle_uid: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_cycles_for_document(self, document_uid, include_active_only) -> List[Dict[str, Any]]
Purpose: Get all approval cycles for a document. Args: document_uid: UID of the document include_active_only: Whether to include only active approval cycles Returns: List of dictionaries with cycle information
Parameters:
document_uid: Type: strinclude_active_only: Type: bool
Returns: Returns List[Dict[str, Any]]
get_cycles_for_document_version(self, version_uid) -> List[Dict[str, Any]]
Purpose: Get all approval cycles for a document version. Args: version_uid: UID of the document version Returns: List of dictionaries with cycle information
Parameters:
version_uid: Type: str
Returns: Returns List[Dict[str, Any]]
get_current_cycle_for_document(self, document_uid) -> Optional[Dict[str, Any]]
Purpose: Get the current active approval cycle for a document. Args: document_uid: UID of the document Returns: Dictionary with cycle information or None if no active cycle
Parameters:
document_uid: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_user_pending_approvals(self, user_uid, include_completed) -> Dict[str, Any]
Purpose: Get pending approvals for a user. Args: user_uid: UID of the user include_completed: Whether to include completed approvals Returns: Dictionary with pending approvals information
Parameters:
user_uid: Type: strinclude_completed: Type: bool
Returns: Returns Dict[str, Any]
get_assignments_for_cycle(self, cycle_uid) -> List[Dict[str, Any]]
Purpose: Get all approver assignments for an approval cycle. Args: cycle_uid: UID of the approval cycle Returns: List of dictionaries with assignment information
Parameters:
cycle_uid: Type: str
Returns: Returns List[Dict[str, Any]]
get_assignment_by_uid(self, assignment_uid) -> Optional[Dict[str, Any]]
Purpose: Get approver assignment information by UID. Args: assignment_uid: UID of the assignment Returns: Dictionary with assignment information or None if not found
Parameters:
assignment_uid: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_assignment_for_user(self, cycle_uid, user_uid) -> Optional[Dict[str, Any]]
Purpose: Get assignment for a specific approver in an approval cycle. Args: cycle_uid: UID of the approval cycle user_uid: UID of the approver Returns: Dictionary with assignment information or None if not found
Parameters:
cycle_uid: Type: struser_uid: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_assignments_for_user(self, user_uid, status_filter, limit, offset) -> Tuple[List[Dict[str, Any]], int]
Purpose: Get all approval 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)
Parameters:
user_uid: Type: strstatus_filter: Type: Optional[List[str]]limit: Type: intoffset: Type: int
Returns: Returns Tuple[List[Dict[str, Any]], int]
update_cycle_status(self, cycle_uid, status) -> bool
Purpose: Update the status of an approval cycle. Args: cycle_uid: UID of the approval cycle status: New status value Returns: Boolean indicating success
Parameters:
cycle_uid: Type: strstatus: Type: str
Returns: Returns bool
complete_cycle(self, cycle_uid, decision, comment) -> bool
Purpose: Complete an approval cycle with a decision. Args: cycle_uid: UID of the approval cycle decision: Decision value (e.g., 'APPROVED' or 'REJECTED') comment: Optional comment about the decision Returns: Boolean indicating success
Parameters:
cycle_uid: Type: strdecision: Type: strcomment: Type: Optional[str]
Returns: Returns bool
cancel_cycle(self, cycle_uid, reason) -> bool
Purpose: Cancel an approval cycle. Args: cycle_uid: UID of the approval cycle reason: Optional reason for cancellation Returns: Boolean indicating success
Parameters:
cycle_uid: Type: strreason: Type: Optional[str]
Returns: Returns bool
add_participant(self, cycle_uid, user_uid, sequence_order) -> bool
Purpose: Add an approver to an approval cycle. Args: cycle_uid: UID of the approval cycle user_uid: UID of the user to add sequence_order: Order in the sequence (for sequential approvals) Returns: Boolean indicating success
Parameters:
cycle_uid: Type: struser_uid: Type: strsequence_order: Type: int
Returns: Returns bool
remove_participant(self, cycle_uid, user_uid, reason) -> bool
Purpose: Remove an approver from an approval cycle. Args: cycle_uid: UID of the approval cycle user_uid: UID of the user to remove reason: Optional reason for removal Returns: Boolean indicating success
Parameters:
cycle_uid: Type: struser_uid: Type: strreason: Type: Optional[str]
Returns: Returns bool
update_assignment_status(self, assignment_uid, status) -> bool
Purpose: Update the status of an approver assignment. Args: assignment_uid: UID of the approver assignment status: New status value Returns: Boolean indicating success
Parameters:
assignment_uid: Type: strstatus: Type: str
Returns: Returns bool
update_assignment_status(self, assignment_uid, status) -> bool
Purpose: Update the status of an approver assignment. Args: assignment_uid: UID of the approver assignment status: New status value Returns: Boolean indicating success
Parameters:
assignment_uid: Type: strstatus: Type: str
Returns: Returns bool
complete_assignment(self, assignment_uid, decision, comments) -> bool
Purpose: Complete an approver assignment with a decision. Args: assignment_uid: UID of the approver assignment decision: Decision value (e.g., 'APPROVED', 'REJECTED') comments: Optional comments about the decision Returns: Boolean indicating success
Parameters:
assignment_uid: Type: strdecision: Type: strcomments: Type: Optional[str]
Returns: Returns bool
add_comment(self, cycle_uid, user_uid, text, requires_resolution, comment_type, parent_comment_uid, location) -> Optional[Dict[str, Any]]
Purpose: Add a comment to an approval cycle. Args: cycle_uid: UID of the approval 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
Parameters:
cycle_uid: Type: struser_uid: Type: strtext: Type: strrequires_resolution: Type: boolcomment_type: Type: strparent_comment_uid: Type: Optional[str]location: Type: Optional[Dict[str, Any]]
Returns: Returns Optional[Dict[str, Any]]
get_comments(self, cycle_uid) -> List[Dict[str, Any]]
Purpose: Get all comments for an approval cycle. Args: cycle_uid: UID of the approval cycle Returns: List of dictionaries with comment information
Parameters:
cycle_uid: Type: str
Returns: Returns List[Dict[str, Any]]
resolve_comment(self, comment_uid, resolution_text) -> bool
Purpose: Resolve an approval comment. Args: comment_uid: UID of the approval comment resolution_text: Resolution text Returns: Boolean indicating success
Parameters:
comment_uid: Type: strresolution_text: Type: str
Returns: Returns bool
get_comment_by_uid(self, comment_uid) -> Optional[Dict[str, Any]]
Purpose: Get approval comment information by UID. Args: comment_uid: UID of the approval comment Returns: Dictionary with comment information or None if not found
Parameters:
comment_uid: Type: str
Returns: Returns Optional[Dict[str, Any]]
update_due_date(self, cycle_uid, due_date) -> bool
Purpose: Update the due date for an approval cycle. Args: cycle_uid: UID of the approval cycle due_date: New due date Returns: Boolean indicating success
Parameters:
cycle_uid: Type: strdue_date: Type: datetime
Returns: Returns bool
get_overdue_assignments(self) -> List[Dict[str, Any]]
Purpose: Get all overdue approver assignments. Returns: List of dictionaries with overdue assignment information
Returns: Returns List[Dict[str, Any]]
get_approval_steps(self, cycle_uid) -> List[Dict[str, Any]]
Purpose: Get all approval steps for an approval cycle. Args: cycle_uid: UID of the approval cycle Returns: List of dictionaries with approval step information
Parameters:
cycle_uid: Type: str
Returns: Returns List[Dict[str, Any]]
create_approval_workflow(self, document_type, workflow_type) -> Optional[Dict[str, Any]]
Purpose: Create a new approval workflow template. Args: document_type: Type of document this workflow applies to workflow_type: Type of workflow Returns: Dictionary with created workflow information or None if failed
Parameters:
document_type: Type: strworkflow_type: Type: str
Returns: Returns Optional[Dict[str, Any]]
get_approval_workflow_templates(self, document_type) -> List[Dict[str, Any]]
Purpose: Get approval workflow templates. Args: document_type: Optional document type to filter by Returns: List of dictionaries with workflow template information
Parameters:
document_type: Type: Optional[str]
Returns: Returns List[Dict[str, Any]]
get_workflow_statistics(self) -> Dict[str, Any]
Purpose: Get statistics about approval workflows. Returns: Dictionary with statistics information
Returns: Returns Dict[str, Any]
search_cycles(self, query, limit, offset) -> Tuple[List[Dict[str, Any]], int]
Purpose: Search for approval 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)
Parameters:
query: Type: Dict[str, Any]limit: Type: intoffset: Type: int
Returns: Returns Tuple[List[Dict[str, Any]], int]
is_user_participant(self, cycle_uid, user_uid) -> bool
Purpose: Check if a user is a participant in an approval cycle. Args: cycle_uid: UID of the approval cycle user_uid: UID of the user to check Returns: Boolean indicating if user is a participant
Parameters:
cycle_uid: Type: struser_uid: Type: str
Returns: Returns bool
_send_approver_notifications(self, approval_cycle, document, approver_uids)
Purpose: Send notifications to approvers. Args: approval_cycle: ApprovalCycle instance document: Document instance approver_uids: List of approver UIDs Returns: None
Parameters:
approval_cycle: Parameterdocument: Parameterapprover_uids: Parameter
Returns: See docstring for return details
Required Imports
from typing import Dict
from typing import List
from typing import Any
from typing import Optional
from typing import Union
Usage Example
# Example usage:
# result = ApprovalController(bases)Tags
Similar Components
AI-powered semantic similarity - components with related functionality:
-
class ReviewController 79.1% similar
-
class ApprovalControllerBase 78.4% similar
-
class ApprovalCycle 70.0% similar
-
class ApprovalCycle_v1 69.2% similar
-
class ApprovalCycle_v2 68.7% similar