🔍 Code Extractor

class ReviewerAssignment

Maturity: 52

Model class representing a reviewer assignment within a review cycle, managing reviewer information, status, decisions, and lifecycle of review assignments.

File:
/tf/active/vicechatdev/CDocs single class/models/review.py
Lines:
1295 - 1596
Complexity:
complex

Purpose

ReviewerAssignment manages individual reviewer assignments within a review cycle. It handles reviewer identification, assignment status tracking, decision recording, activity monitoring, and database persistence. The class provides backward compatibility for legacy field names (reviewer_uid/user_uid, reviewer_name/user_name, assigned_date/assigned_at) and integrates with the database layer for CRUD operations. It supports sequential review workflows through sequence ordering and tracks the complete lifecycle from assignment through decision to potential removal.

Source Code

class ReviewerAssignment(AssignmentBase, 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)
            
        BaseModel.__init__(self, data or {})
        AssignmentBase.__init__(self, data or {})
        
        # Ensure backward compatibility
        if 'user_uid' not in self._data and 'reviewer_uid' in self._data:
            self._data['user_uid'] = self._data['reviewer_uid']
        if 'user_name' not in self._data and 'reviewer_name' in self._data:
            self._data['user_name'] = self._data['reviewer_name']
        if 'assigned_at' not in self._data and 'assigned_date' in self._data:
            try:
                if isinstance(self._data['assigned_date'], datetime):
                    self._data['assigned_at'] = self._data['assigned_date'].isoformat()
                else:
                    self._data['assigned_at'] = self._data['assigned_date']
            except:
                self._data['assigned_at'] = datetime.now().isoformat()
    
    @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') or self._data.get('user_uid')
    
    @property
    def reviewer_name(self) -> Optional[str]:
        """Get name of the assigned reviewer."""
        name = self._data.get('reviewer_name') or self._data.get('user_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
                self._data['user_name'] = name
        return name
    
    @property
    def status(self) -> str:
        """Get assignment status."""
        return self._data.get('status', AssignmentStatus.PENDING)
    
    @status.setter
    def status(self, value: str) -> None:
        """Set assignment status."""
        if value not in [s.value for s in AssignmentStatus]:
            logger.warning(f"Invalid assignment status: {value}")
            return
            
        self._data['status'] = value
        db.update_node(self.uid, {'status': value})
    
    @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."""
        if 'assigned_date' in self._data:
            return self._data['assigned_date']
        elif 'assigned_at' in self._data and self._data['assigned_at']:
            try:
                return datetime.fromisoformat(self._data['assigned_at'].replace('Z', '+00:00'))
            except:
                pass
        return None
    
    @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."""
        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."""
        # Check both fields for backward compatibility
        return self._data.get('decision_comments') or self._data.get('comment')
    
    @decision_comments.setter
    def decision_comments(self, text: str) -> None:
        """Set decision comments."""
        self._data['decision_comments'] = text
        self._data['comment'] = text  # Also set workflow_base style
        db.update_node(self.uid, {'decision_comments': text, 'comment': 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})
    
    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
    
    def _update_cycle_status(self) -> None:
        """Update the parent cycle status."""
        try:
            review_cycle = ReviewCycle(uid=self.review_cycle_uid)
            if review_cycle:
                review_cycle.update_status()
        except Exception as e:
            logger.error(f"Error updating review cycle status: {e}")
    
    @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 OR a.user_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 OR a.user_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

Name Type Default Kind
bases AssignmentBase, BaseModel -

Parameter Details

data: Optional dictionary containing assignment properties including reviewer_uid, review_cycle_uid, status, decision, sequence_order, and dates. 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 with properties accessible via getters/setters. The save() method returns a boolean indicating success/failure. Class methods get_user_assignments() returns a list of ReviewerAssignment instances, and count_user_assignments() returns an integer count.

Class Interface

Methods

__init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None)

Purpose: Initialize a reviewer assignment, either from provided data or by loading from database using uid

Parameters:

  • data: Dictionary of assignment properties
  • uid: Assignment UID to load from database if data not provided

Returns: None (constructor)

@property review_cycle_uid(self) -> Optional[str] property

Purpose: Get the UID of the review cycle this assignment belongs to

Returns: String UID of the parent review cycle or None

@property reviewer_uid(self) -> Optional[str] property

Purpose: Get UID of the assigned reviewer with backward compatibility for user_uid field

Returns: String UID of the reviewer or None

@property reviewer_name(self) -> Optional[str] property

Purpose: Get name of the assigned reviewer, fetching from database if not cached

Returns: String name of the reviewer or None

@property status(self) -> str property

Purpose: Get the current assignment status

Returns: String status value from AssignmentStatus enum, defaults to PENDING

@status.setter status(self, value: str) -> None property

Purpose: Set assignment status and persist to database

Parameters:

  • value: New status value, must be valid AssignmentStatus enum value

Returns: None

@property sequence_order(self) -> Optional[int] property

Purpose: Get sequence order for sequential reviews

Returns: Integer sequence order or None

@sequence_order.setter sequence_order(self, value: Optional[int]) -> None property

Purpose: Set sequence order and persist to database

Parameters:

  • value: Integer sequence order or None

Returns: None

@property assigned_date(self) -> Optional[datetime] property

Purpose: Get date when reviewer was assigned, handling both assigned_date and assigned_at fields

Returns: datetime object of assignment date or None

@property first_activity_date(self) -> Optional[datetime] property

Purpose: Get date of first reviewer activity

Returns: datetime object of first activity or None

@first_activity_date.setter first_activity_date(self, date: datetime) -> None property

Purpose: Set first activity date and persist to database

Parameters:

  • date: datetime object representing first activity

Returns: None

@property decision(self) -> Optional[str] property

Purpose: Get reviewer's decision

Returns: String decision value or None

@decision.setter decision(self, value: str) -> None property

Purpose: Set reviewer's decision and persist to database

Parameters:

  • value: Decision value, must be in settings.REVIEW_DECISIONS

Returns: None

@property decision_date(self) -> Optional[datetime] property

Purpose: Get date when decision was made

Returns: datetime object of decision date or None

@decision_date.setter decision_date(self, date: datetime) -> None property

Purpose: Set decision date and persist to database

Parameters:

  • date: datetime object representing decision date

Returns: None

@property decision_comments(self) -> Optional[str] property

Purpose: Get comments about the decision with backward compatibility for comment field

Returns: String comments or None

@decision_comments.setter decision_comments(self, text: str) -> None property

Purpose: Set decision comments and persist to database, also sets comment field for compatibility

Parameters:

  • text: String comment text

Returns: None

@property removal_date(self) -> Optional[datetime] property

Purpose: Get date when reviewer was removed

Returns: datetime object of removal date or None

@removal_date.setter removal_date(self, date: datetime) -> None property

Purpose: Set removal date and persist to database

Parameters:

  • date: datetime object representing removal date

Returns: None

@property removal_reason(self) -> Optional[str] property

Purpose: Get reason for reviewer removal

Returns: String removal reason or None

@removal_reason.setter removal_reason(self, text: str) -> None property

Purpose: Set removal reason and persist to database

Parameters:

  • text: String reason for removal

Returns: None

save(self) -> bool

Purpose: Save changes to database, creating node if it doesn't exist and establishing relationships

Returns: Boolean indicating success (True) or failure (False)

_update_cycle_status(self) -> None

Purpose: Update the parent review cycle status (internal method)

Returns: None

@classmethod get_user_assignments(cls, user_uid: str, status_filter: Optional[List[str]] = None, limit: int = 100, offset: int = 0) -> List['ReviewerAssignment']

Purpose: Get all review assignments for a user with optional filtering and pagination

Parameters:

  • user_uid: UID of the user to get assignments for
  • status_filter: Optional list of status values to filter by
  • limit: Maximum number of assignments to return (default 100)
  • offset: Number of assignments to skip for pagination (default 0)

Returns: List of ReviewerAssignment instances matching criteria

@classmethod count_user_assignments(cls, user_uid: str, status_filter: Optional[List[str]] = None) -> int

Purpose: Count review assignments for a user with optional status filtering

Parameters:

  • user_uid: UID of the user to count assignments for
  • status_filter: Optional list of status values to filter by

Returns: Integer count of matching assignments

Attributes

Name Type Description Scope
_data Dict[str, Any] Internal dictionary storing all assignment data, inherited from BaseModel instance
uid str Unique identifier for the assignment, inherited from BaseModel instance

Dependencies

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

Required Imports

import logging
import uuid
from typing import Dict, List, Any, Optional, Union, Set
from datetime import datetime, timedelta
from CDocs import db
from CDocs.config import settings
from CDocs.db.schema_manager import NodeLabels, RelTypes
from CDocs.models import BaseModel, register_model
from CDocs.models.workflow_base import AssignmentBase, AssignmentStatus
from CDocs.models.review_cycle import ReviewCycle
import traceback

Usage Example

# Create new reviewer assignment
assignment = ReviewerAssignment(data={
    'review_cycle_uid': 'cycle-123',
    'reviewer_uid': 'user-456',
    'reviewer_name': 'John Doe',
    'status': 'pending',
    'sequence_order': 1
})
assignment.save()

# Load existing assignment
existing = ReviewerAssignment(uid='assignment-789')

# Update status and decision
existing.status = 'in_progress'
existing.decision = 'approved'
existing.decision_date = datetime.now()
existing.decision_comments = 'Looks good to me'
existing.save()

# Get all assignments for a user
user_assignments = ReviewerAssignment.get_user_assignments(
    user_uid='user-456',
    status_filter=['pending', 'in_progress'],
    limit=50,
    offset=0
)

# Count assignments
count = ReviewerAssignment.count_user_assignments(
    user_uid='user-456',
    status_filter=['pending']
)

Best Practices

  • Always call save() after modifying properties to persist changes to the database
  • Use the property setters rather than directly modifying _data to ensure database synchronization
  • Check that status values are valid AssignmentStatus enum values before setting
  • Verify decision values against settings.REVIEW_DECISIONS before setting
  • Handle backward compatibility fields (reviewer_uid vs user_uid) by using the provided properties
  • When creating new assignments, ensure review_cycle_uid is set to establish proper relationships
  • Use get_user_assignments() with pagination (limit/offset) for large result sets
  • The class automatically updates parent ReviewCycle status via _update_cycle_status()
  • first_activity_date should be set when reviewer first interacts with the assignment
  • removal_date and removal_reason should be set together when removing a reviewer
  • The class handles database node creation if it doesn't exist during save()
  • Properties that fetch from database (like reviewer_name) cache results in _data

Tags

review assignment workflow reviewer decision-tracking database-model status-management sequential-review user-assignment

Similar Components

AI-powered semantic similarity - components with related functionality:

  • class ReviewerAssignment_v1 96.2% similar

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

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

  • class ApproverAssignment_v1 77.5% 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 76.6% similar

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

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

  • class AssignmentBase 75.5% similar

    Base class for managing assignment lifecycle in a document review/approval workflow system, tracking status, timestamps, user assignments, and decisions.

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

  • class ApproverAssignment 74.3% similar

    Model class representing an approver assignment within an approval cycle, managing the state and lifecycle of individual approval tasks assigned to users.

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