🔍 Code Extractor

Model class representing a reviewer assignment within a review cycle, managing reviewer information, status, decisions, and lifecycle tracking for document review processes.

Purpose

ReviewerAssignment manages the complete lifecycle of a reviewer's assignment to a review cycle. It tracks reviewer identity, assignment status (PENDING, ACTIVE, COMPLETED, CANCELED, REMOVED), decisions (APPROVED, REJECTED), activity dates, instructions, sequence order for sequential reviews, and removal information. The class provides database persistence through automatic syncing of property changes and supports querying assignments by user. It inherits from BaseModel and is decorated with @register_model for framework integration.

Source Code

class ReviewerAssignment(BaseModel):
    """Model representing a reviewer assignment within a review cycle."""
    
    def __init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None):
        """
        Initialize a reviewer assignment.
        
        Args:
            data: Dictionary of assignment properties
            uid: Assignment UID to load from database (if data not provided)
        """
        if data is None and uid is not None:
            # Fetch assignment data from database
            data = db.get_node_by_uid(uid)
            
        super().__init__(data or {})
    
    @property
    def review_cycle_uid(self) -> Optional[str]:
        """Get the UID of the review cycle this assignment belongs to."""
        return self._data.get('review_cycle_uid')
    
    @property
    def reviewer_uid(self) -> Optional[str]:
        """Get UID of the assigned reviewer."""
        return self._data.get('reviewer_uid')
    
    @property
    def reviewer_name(self) -> Optional[str]:
        """Get name of the assigned reviewer."""
        name = self._data.get('reviewer_name')
        if not name and self.reviewer_uid:
            # Try to fetch from database
            result = db.run_query(
                "MATCH (u:User {UID: $uid}) RETURN u.Name as name",
                {"uid": self.reviewer_uid}
            )
            if result and 'name' in result[0]:
                name = result[0]['name']
                # Cache it
                self._data['reviewer_name'] = name
        return name
    
    @property
    def status(self) -> str:
        """Get assignment status (PENDING, ACTIVE, COMPLETED, CANCELED, REMOVED)."""
        return self._data.get('status', 'PENDING')
    
    @status.setter
    def status(self, value: str) -> None:
        """Set assignment status."""
        if value not in ['PENDING', 'ACTIVE', 'COMPLETED', 'CANCELED', 'REMOVED']:
            logger.warning(f"Invalid assignment status: {value}")
            return
            
        self._data['status'] = value
        db.update_node(self.uid, {'status': value})

        
    # Add a new property for reviewer-specific instructions
    @property
    def instructions(self) -> Optional[str]:
        """Get reviewer-specific instructions."""
        return self._data.get('instructions', '')
    
    @instructions.setter
    def instructions(self, text: str) -> None:
        """Set reviewer-specific instructions."""
        self._data['instructions'] = text
        db.update_node(self.uid, {'instructions': text})
    
    @property
    def sequence_order(self) -> Optional[int]:
        """Get sequence order for sequential reviews."""
        return self._data.get('sequence_order')
    
    @sequence_order.setter
    def sequence_order(self, value: Optional[int]) -> None:
        """Set sequence order."""
        self._data['sequence_order'] = value
        db.update_node(self.uid, {'sequence_order': value})
    
    @property
    def assigned_date(self) -> Optional[datetime]:
        """Get date when reviewer was assigned."""
        return self._data.get('assigned_date')
    
    @property
    def first_activity_date(self) -> Optional[datetime]:
        """Get date of first reviewer activity."""
        return self._data.get('first_activity_date')
    
    @first_activity_date.setter
    def first_activity_date(self, date: datetime) -> None:
        """Set first activity date."""
        self._data['first_activity_date'] = date
        db.update_node(self.uid, {'first_activity_date': date})
    
    @property
    def decision(self) -> Optional[str]:
        """Get reviewer's decision (APPROVED, REJECTED, etc.)."""
        return self._data.get('decision')
    
    @decision.setter
    def decision(self, value: str) -> None:
        """Set reviewer's decision."""
        if value not in settings.REVIEW_DECISIONS:
            logger.warning(f"Invalid review decision: {value}")
            return
            
        self._data['decision'] = value
        db.update_node(self.uid, {'decision': value})
    
    @property
    def decision_date(self) -> Optional[datetime]:
        """Get date when decision was made."""
        return self._data.get('decision_date')
    
    @decision_date.setter
    def decision_date(self, date: datetime) -> None:
        """Set decision date."""
        self._data['decision_date'] = date
        db.update_node(self.uid, {'decision_date': date})
    
    @property
    def decision_comments(self) -> Optional[str]:
        """Get comments about the decision."""
        return self._data.get('decision_comments')
    
    @decision_comments.setter
    def decision_comments(self, text: str) -> None:
        """Set decision comments."""
        self._data['decision_comments'] = text
        db.update_node(self.uid, {'decision_comments': text})
    
    @property
    def removal_date(self) -> Optional[datetime]:
        """Get date when reviewer was removed."""
        return self._data.get('removal_date')
    
    @removal_date.setter
    def removal_date(self, date: datetime) -> None:
        """Set removal date."""
        self._data['removal_date'] = date
        db.update_node(self.uid, {'removal_date': date})
    
    @property
    def removal_reason(self) -> Optional[str]:
        """Get reason for reviewer removal."""
        return self._data.get('removal_reason')
    
    @removal_reason.setter
    def removal_reason(self, text: str) -> None:
        """Set removal reason."""
        self._data['removal_reason'] = text
        db.update_node(self.uid, {'removal_reason': text})
    
        # Add this method to the ReviewerAssignment 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.REVIEWER_ASSIGNMENT,
                    self._data,
                    self.uid
                )
                
                if created and self._data.get('review_cycle_uid'):
                    # Create ASSIGNMENT relationship from ReviewCycle to ReviewerAssignment
                    db.create_relationship(
                        self._data['review_cycle_uid'],
                        self.uid,
                        RelTypes.ASSIGNMENT
                    )
                return created
            
            # Otherwise update existing node
            return db.update_node(self.uid, self._data)
        except Exception as e:
            logger.error(f"Error saving reviewer assignment: {e}")
            import traceback
            logger.error(traceback.format_exc())
            return False
    
    @classmethod
    def get_user_assignments(cls, user_uid: str, 
                           status_filter: Optional[List[str]] = None,
                           limit: int = 100, 
                           offset: int = 0) -> List['ReviewerAssignment']:
        """
        Get all review assignments for a user.
        
        Args:
            user_uid: UID of the user
            status_filter: Optional list of statuses to filter by
            limit: Maximum number of assignments to return
            offset: Number of assignments to skip
            
        Returns:
            List of ReviewerAssignment instances
        """
        try:
            # Build query
            query = """
            MATCH (a:ReviewerAssignment)
            WHERE a.reviewer_uid = $user_uid
            """
            
            params = {"user_uid": user_uid}
            
            # Add status filter if provided
            if status_filter:
                query += "AND a.status IN $statuses "
                params["statuses"] = status_filter
                
            # Add ordering and pagination
            query += """
            RETURN a
            ORDER BY a.assigned_date DESC
            SKIP $offset
            LIMIT $limit
            """
            
            params["offset"] = offset
            params["limit"] = limit
            
            # Execute query
            result = db.run_query(query, params)
            
            # Convert to assignment instances
            assignments = [cls(record['a']) for record in result if 'a' in record]
            
            return assignments
            
        except Exception as e:
            logger.error(f"Error getting user assignments: {e}")
            return []
    
    @classmethod
    def count_user_assignments(cls, user_uid: str, 
                             status_filter: Optional[List[str]] = None) -> int:
        """
        Count review assignments for a user.
        
        Args:
            user_uid: UID of the user
            status_filter: Optional list of statuses to filter by
            
        Returns:
            Count of matching assignments
        """
        try:
            # Build query
            query = """
            MATCH (a:ReviewerAssignment)
            WHERE a.reviewer_uid = $user_uid
            """
            
            params = {"user_uid": user_uid}
            
            # Add status filter if provided
            if status_filter:
                query += "AND a.status IN $statuses "
                params["statuses"] = status_filter
                
            # Add count
            query += "RETURN count(a) as count"
            
            # Execute query
            result = db.run_query(query, params)
            
            # Get count
            if result and 'count' in result[0]:
                return result[0]['count']
                
            return 0
            
        except Exception as e:
            logger.error(f"Error counting user assignments: {e}")
            return 0

Parameters

Parameter Details

data: Optional dictionary containing assignment properties (review_cycle_uid, reviewer_uid, reviewer_name, status, instructions, sequence_order, assigned_date, first_activity_date, decision, decision_date, decision_comments, removal_date, removal_reason). If None and uid is provided, data will be fetched from database.

uid: Optional unique identifier string for the assignment. If provided without data, the assignment will be loaded from the database using this UID.

Return Value

Instantiation returns a ReviewerAssignment object. The save() method returns a boolean indicating success/failure. get_user_assignments() returns a list of ReviewerAssignment instances. count_user_assignments() returns an integer count. Properties return their respective types (str, datetime, int, or Optional versions).

Class Interface

Methods

Attributes

Dependencies

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

Required Imports

from typing import Dict, List, Any, Optional
from datetime import datetime
from CDocs import db
from CDocs.config import settings
from CDocs.db.schema_manager import NodeLabels, RelTypes
from CDocs.models.base import BaseModel
from CDocs.models.decorators import register_model
import logging

Conditional/Optional Imports

These imports are only needed under specific conditions:

import traceback

Usage Example

# Create new assignment
assignment = ReviewerAssignment(data={
    'review_cycle_uid': 'cycle123',
    'reviewer_uid': 'user456',
    'reviewer_name': 'John Doe',
    'status': 'PENDING',
    'instructions': 'Please review sections 1-3',
    'sequence_order': 1,
    'assigned_date': datetime.now()
})
assignment.save()

# Load existing assignment
assignment = ReviewerAssignment(uid='assignment789')

# Update status
assignment.status = 'ACTIVE'
assignment.first_activity_date = datetime.now()

# Record decision
assignment.decision = 'APPROVED'
assignment.decision_date = datetime.now()
assignment.decision_comments = 'All requirements met'

# Get user's assignments
user_assignments = ReviewerAssignment.get_user_assignments(
    user_uid='user456',
    status_filter=['ACTIVE', 'PENDING'],
    limit=50,
    offset=0
)

# Count assignments
count = ReviewerAssignment.count_user_assignments(
    user_uid='user456',
    status_filter=['COMPLETED']
)

Best Practices

• Always call save() after creating a new assignment to persist it to the database
• Use property setters rather than direct _data manipulation to ensure database synchronization
• Valid status values are: PENDING, ACTIVE, COMPLETED, CANCELED, REMOVED - setting invalid values will be ignored with a warning
• Decision values must be in settings.REVIEW_DECISIONS list
• When loading by uid, ensure the node exists in the database or handle None data appropriately
• The reviewer_name property has lazy loading - it will fetch from database if not cached
• Use get_user_assignments() with pagination (limit/offset) for large result sets
• Set first_activity_date when reviewer first interacts with the assignment
• Set removal_date and removal_reason when removing a reviewer from an assignment
• The save() method creates ASSIGNMENT relationship to review cycle automatically
• Status transitions should follow logical workflow: PENDING -> ACTIVE -> COMPLETED/CANCELED/REMOVED

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class ApproverAssignment 77.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 lifecycle tracking.

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

• class ApproverAssignment_v1 75.8% 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

• function get_user_assigned_reviews 70.5% similar

  Retrieves all review assignments for a specific user with optional filtering by status, including associated review cycle, document, and version details.

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

• class ReviewCycle 64.6% similar

  Model representing a review cycle for a document version.

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

• class ReviewComment 64.6% similar

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

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