🔍 Code Extractor

Model class representing a specific version of a controlled document in a document management system, handling version metadata, file paths, status tracking, and review workflows.

Purpose

DocumentVersion manages individual versions of controlled documents, including their lifecycle from draft to published states. It handles file storage paths in FileCloud (both editable and PDF formats), tracks version metadata (version number, status, dates), manages relationships to parent documents and authors, supports review cycles, and provides hash-based integrity verification. The class integrates with a Neo4j database for persistence and maintains relationships between versions, documents, and users.

Source Code

class DocumentVersion(BaseModel):
    """Model representing a specific version of a controlled document."""
    
    def __init__(self, data: Optional[Dict[str, Any]] = None, uid: Optional[str] = None):
        """
        Initialize a document version.
        
        Args:
            data: Dictionary of version properties
            uid: Version UID to load from database (if data not provided)
        """
        if data is None and uid is not None:
            # Fetch document data from database
            data = db.get_node_by_uid(uid)
            
        super().__init__(data or {})
        self._parent_doc = None  # Cached parent document
    
    @classmethod
    def create(cls, document_uid: str, version_number: str, 
            created_by: Union[DocUser, str], 
            file_paths: Optional[Dict[str, str]] = None,
            properties: Optional[Dict[str, Any]] = None) -> Optional['DocumentVersion']:
        """
        Create a new document version.
        
        Args:
            document_uid: UID of parent document
            version_number: Version number string (e.g., "1.0")
            created_by: User creating the version or their UID
            file_paths: Dictionary containing file paths in FileCloud
                    Can include 'base', 'word', 'pdf' keys
            properties: Additional properties for the version
            
        Returns:
            New DocumentVersion instance or None if creation failed
        """
        try:
            # Prepare properties
            props = properties or {}
            props.update({
                'versionNumber': version_number,
                'status': 'DRAFT',
                'createdDate': datetime.now(),
            })
            
            # Handle file paths based on the new model
            if file_paths:
                # If base path is directly provided, use it
                if 'base' in file_paths:
                    props['fileCloudPath'] = file_paths['base']
                # Otherwise, try to derive it from word or pdf path
                elif 'word' in file_paths:
                    base_path, ext = os.path.splitext(file_paths['word'])
                    props['fileCloudPath'] = base_path
                    props['file_type'] = ext[1:] if ext and ext.startswith('.') else 'docx'
                elif 'pdf' in file_paths:
                    base_path, _ = os.path.splitext(file_paths['pdf'])
                    props['fileCloudPath'] = base_path
                    props['file_type'] = 'pdf'
            
            # Create node in database
            version_data = db.create_node(
                NodeLabels.DOCUMENT_VERSION,
                props,
                document_uid,
                RelTypes.HAS_VERSION
            )
            
            if not version_data:
                logger.error(f"Failed to create document version for document {document_uid}")
                return None
                
            # Create the version instance
            version = cls(version_data)
            
            # Link to author
            user_uid = created_by.uid if isinstance(created_by, DocUser) else created_by
            db.create_relationship(
                version.uid,
                user_uid,
                RelTypes.AUTHORED_BY
            )
            
            # Link to previous version if any exists
            previous_versions = db.run_query(
                """
                MATCH (d:ControlledDocument {UID: $doc_uid})-[:HAS_VERSION]->(v:DocumentVersion)
                WHERE v.UID <> $version_uid
                RETURN v
                ORDER BY v.versionNumber DESC
                LIMIT 1
                """,
                {"doc_uid": document_uid, "version_uid": version.uid}
            )
            
            if previous_versions and 'v' in previous_versions[0]:
                prev_version = previous_versions[0]['v']
                db.create_relationship(
                    version.uid,
                    prev_version['UID'],
                    RelTypes.PREVIOUS_VERSION
                )
            
            return version
            
        except Exception as e:
            logger.error(f"Error creating document version: {e}")
            return None
    
    @property
    def document_uid(self) -> Optional[str]:
        """Get the UID of the parent document."""
        if hasattr(self, '_document_uid') and self._document_uid:
            return self._document_uid
            
        result = db.run_query(
            """
            MATCH (d:ControlledDocument)-[:HAS_VERSION]->(v:DocumentVersion {UID: $uid})
            RETURN d.UID as doc_uid
            """,
            {"uid": self.uid}
        )
        
        if result and 'doc_uid' in result[0]:
            self._document_uid = result[0]['doc_uid']
            return self._document_uid
            
        return None
    
    @property
    def document(self) -> Optional['ControlledDocument']:
        """Get the parent document instance."""
        if self._parent_doc:
            return self._parent_doc
            
        doc_uid = self.document_uid
        if doc_uid:
            self._parent_doc = ControlledDocument(uid=doc_uid)
            return self._parent_doc
            
        return None
    
    @property
    def version_number(self) -> str:
        """Get version number."""
        return self._data.get('version_number', '')
    
    @property
    def status(self) -> str:
        """Get version status code."""
        return self._data.get('status', 'DRAFT')
    
    @status.setter
    def status(self, value: str) -> None:
        """Set version status using code."""
        # Convert from name to code if needed
        status_code = settings.get_document_status_code(value)
        
        # Validate status code
        if not settings.is_valid_document_status(status_code):
            logger.warning(f"Invalid status: {value}")
            return
            
        self._data['status'] = status_code
        db.update_node(self.uid, {'status': status_code})
    
    def get_status_name(self) -> str:
        """Get the full status name."""
        return settings.get_document_status_name(self.status)
    
    def get_status_color(self) -> str:
        """Get the color for displaying this status."""
        return settings.get_status_color(self.status)
    
    @property
    def is_current(self) -> bool:
        """
        Check if this version is the current version of its document.
        
        Returns:
            bool: True if this is the current version, False otherwise
        """
        try:
            # Use the document method if we have a document reference
            document = self.document
            if document:
                return document.is_current_version(self.uid)
                
            # Otherwise, query directly using the database helper
            result = db.run_query(
                """
                MATCH (d:ControlledDocument)-[:CURRENT_VERSION]->(v:DocumentVersion {UID: $version_uid})
                RETURN COUNT(d) > 0 as is_current
                """,
                {"version_uid": self.uid}
            )
            
            if result and 'is_current' in result[0]:
                return result[0]['is_current']
                
            return False
        except Exception as e:
            logger.error(f"Error checking if version is current: {e}")
            return False

    @property
    def filecloud_path(self) -> Optional[str]:
        """
        Get the base FileCloud path without extension.
        This is the central attribute for file references.
        
        Returns:
            str: Base path in FileCloud without file extension
        """
        return self._data.get('fileCloudPath')
    
    @filecloud_path.setter
    def filecloud_path(self, path: str) -> None:
        """
        Set the base FileCloud path without extension.
        
        Args:
            path: Base path in FileCloud without file extension
        """
        self._data['fileCloudPath'] = path
        db.update_node(self.uid, {'fileCloudPath': path})
    
    @property
    def word_file_path(self) -> Optional[str]:
        """
        Get path to editable file in FileCloud.
        Dynamically constructs path based on filecloud_path and file_type.
        
        Returns:
            str: Path to the editable file or None if base path is not available
        """
        base_path = self.filecloud_path
        if not base_path:
            return None
            
        # Get file extension based on file_type or use a default
        ext = self.file_type
        if not ext.startswith('.'):
            ext = f'.{ext}'
        if ext in ['.docx','xlsx','pptx']:
            return f"{base_path}"
        else:
            return None    
        
    @word_file_path.setter
    def word_file_path(self, path: str) -> None:
        """
        Set the editable file path and extract base path.
        
        Args:
            path: Full path to the editable file
        """
        if not path:
            return
            
        # Extract the base path by removing the extension
        base_path, ext = os.path.splitext(path)
        
        # Store the file type from the extension
        if ext and ext.startswith('.'):
            self._data['file_type'] = ext[1:]  # Remove the leading dot
            db.update_node(self.uid, {'file_type': ext[1:]})
        
        # Store the base path
        self.filecloud_path = base_path
    
    @property
    def pdf_file_path(self) -> Optional[str]:
        """
        Get path to PDF file in FileCloud.
        Dynamically constructs path based on filecloud_path and
        checks if the PDF actually exists in FileCloud.
        
        Returns:
            str: Path to the PDF file or None if base path is not available or PDF doesn't exist
        """
        base_path = self.filecloud_path
        if not base_path:
            return None
            
        # Remove file extension from base path if it has one
        if '.' in os.path.basename(base_path):
            base_path = os.path.splitext(base_path)[0]
        
        # Construct the PDF path
        pdf_path = f"{base_path}.pdf"
        
        # Check if PDF exists in FileCloud
        try:
            # Import here to avoid circular imports
            from CDocs.controllers.filecloud_controller import get_filecloud_client
            
            client = get_filecloud_client()
            if client.check_file_exists(pdf_path):
                return pdf_path
            else:
                # PDF doesn't exist
                return None
        except Exception as e:
            # Log but don't fail - just return the path even if we couldn't verify existence
            logger.warning(f"Error checking if PDF exists at {pdf_path}: {e}")
            return pdf_path
    
    
    @property
    def file_type(self) -> Optional[str]:
        """Get filetype and extension of file"""
        return self._data.get('file_type')
    
    @property
    def file_name(self) -> Optional[str]:
        """Get filetype and extension of file"""
        return self._data.get('file_name')
    
    @pdf_file_path.setter
    def pdf_file_path(self, path: str) -> None:
        """
        Set the PDF file path and extract base path if not already set.
        
        Args:
            path: Full path to the PDF file
        """
        if not path:
            return
            
        # Extract the base path by removing the extension
        base_path, ext = os.path.splitext(path)
        
        # If filecloud_path isn't set yet, set it
        if not self.filecloud_path:
            self.filecloud_path = base_path
        #db.update_node(self.uid, {'fileCloudPath': path})
    
    @property
    def created_date(self) -> Optional[datetime]:
        """Get version creation date."""
        return self._data.get('created_date')
    
    @property
    def effective_date(self) -> Optional[datetime]:
        """Get date when version becomes effective."""
        return self._data.get('effective_date')
    
    @effective_date.setter
    def effective_date(self, date: datetime) -> None:
        """Set effective date."""
        self._data['effectiveDate'] = date
        db.update_node(self.uid, {'effective_date': date})
    
    @property
    def expiry_date(self) -> Optional[datetime]:
        """Get expiry date."""
        return self._data.get('expiryDate')
    
    @expiry_date.setter
    def expiry_date(self, date: datetime) -> None:
        """Set expiry date."""
        self._data['expiryDate'] = date
        db.update_node(self.uid, {'expiryDate': date})
    
    @property
    def author(self) -> Optional[DocUser]:
        """Get author of the version."""
        result = db.run_query(
            """
            MATCH (v:DocumentVersion {UID: $uid})-[:AUTHORED_BY]->(u:User)
            RETURN u
            """,
            {"uid": self.uid}
        )
        
        if result and 'u' in result[0]:
            return DocUser(result[0]['u'])
            
        return None
    
    @property
    def change_summary(self) -> str:
        """Get summary of changes in this version."""
        return self._data.get('changeSummary', '')
    
    @change_summary.setter
    def change_summary(self, summary: str) -> None:
        """Set change summary."""
        self._data['changeSummary'] = summary
        db.update_node(self.uid, {'changeSummary': summary})
    
    @property
    def hash(self) -> Optional[str]:
        """Get document hash."""
        return self._data.get('hash')
    
    @hash.setter
    def hash(self, value: str) -> None:
        """Set document hash."""
        self._data['hash'] = value
        db.update_node(self.uid, {'hash': value})
    
    def calculate_hash(self, file_content: bytes) -> str:
        """
        Calculate SHA-256 hash for file content.
        
        Args:
            file_content: Content of the file as bytes
            
        Returns:
            Hash string
        """
        hasher = hashlib.sha256()
        hasher.update(file_content)
        hash_value = hasher.hexdigest()
        
        # Store hash in the database
        self.hash = hash_value
        
        return hash_value
    
    def get_previous_version(self) -> Optional['DocumentVersion']:
        """
        Get the previous version of the document.
        
        Returns:
            Previous DocumentVersion or None if this is the first version
        """
        result = db.run_query(
            """
            MATCH (v:DocumentVersion {UID: $uid})-[:PREVIOUS_VERSION]->(prev:DocumentVersion)
            RETURN prev
            """,
            {"uid": self.uid}
        )
        
        if result and 'prev' in result[0]:
            return DocumentVersion(result[0]['prev'])
            
        return None
    
    def start_review(self, reviewers: List[Union[DocUser, str]], 
                    due_date: Optional[datetime] = None,
                    instructions: str = '') -> Optional['ReviewCycle']:
        """
        Start a review cycle for this version.
        
        Args:
            reviewers: List of DocUser instances or UIDs of reviewers
            due_date: Date when review should be completed
            instructions: Instructions for reviewers
            
        Returns:
            ReviewCycle instance or None if creation failed
        """
        # Circular import handled by delayed import
        from .review import ReviewCycle
        
        # Generate due date if not provided
        if not due_date:
            due_date = datetime.now() + timedelta(days=settings.DEFAULT_REVIEW_DAYS)
        
        # Create review cycle
        return ReviewCycle.create(self.uid, reviewers, due_date, instructions)
    
    def get_active_review(self) -> Optional[Dict[str, Any]]:
        """
        Get the active review cycle for this version.
        
        Returns:
            Review cycle data or None if no active review
        """
        result = db.run_query(
            """
            MATCH (v:DocumentVersion {UID: $uid})-[:FOR_REVIEW]->(r:ReviewCycle)
            WHERE r.status IN ['PENDING', 'IN_PROGRESS']
            RETURN r
            ORDER BY r.startDate DESC
            LIMIT 1
            """,
            {"uid": self.uid}
        )
        
        if result and 'r' in result[0]:
            return result[0]['r']
            
        return None
    
    def get_reviews(self) -> List[Dict[str, Any]]:
        """
        Get all review cycles for this version.
        
        Returns:
            List of review cycle data
        """
        result = db.run_query(
            """
            MATCH (v:DocumentVersion {UID: $uid})-[:FOR_REVIEW]->(r:ReviewCycle)
            RETURN r
            ORDER BY r.startDate DESC
            """,
            {"uid": self.uid}
        )
        
        return [record['r'] for record in result if 'r' in record]
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary representation."""
        result = super().to_dict()
        
        # Add author information if available
        author = self.author
        if author:
            result['author'] = {
                'UID': author.uid,
                'name': author.name,
                'email': author.email
            }
        
        # Add document ID if available
        document = self.document
        if document:
            result['documentId'] = document.doc_number
            
        return result

    def get_primary_file_path(self) -> Optional[str]:
        """
        Get the primary file path based on document status.
        For published/non-editable documents, returns PDF path.
        For editable documents, returns editable document path.
        
        Returns:
            str: Path to the primary file or None if no files are available
        """
        if not self.current_version:
            return None
            
        if self.is_published():
            return self.current_version.pdf_file_path
        else:
            return self.current_version.word_file_path

    def has_pdf_version(self) -> bool:
        """
        Check if this version has a PDF version.
        
        Returns:
            bool: True if version has PDF file
        """
        return bool(self.pdf_file_path)

    def has_editable_version(self) -> bool:
        """
        Check if this version has an editable version.
        
        Returns:
            bool: True if version has editable file
        """
        return bool(self.word_file_path)

    def get_file_extension(self) -> str:
        """
        Get the file extension for the primary file.
        
        Returns:
            str: File extension (like '.docx', '.pdf')
        """
        if self.word_file_path:
            # Get extension from path
            return os.path.splitext(self.word_file_path)[1].lower()
        elif self.pdf_file_path:
            return '.pdf'
        else:
            return ''

    def is_editable_format(self) -> bool:
        """
        Check if the primary file is in an editable format.
        
        Returns:
            bool: True if editable format
        """
        ext = self.get_file_extension()
        return ext.lower() in ['.docx', '.doc', '.pptx', '.ppt', '.xlsx', '.xls']

Parameters

Parameter Details

data: Optional dictionary containing version properties loaded from database or provided directly. Keys include 'versionNumber', 'status', 'createdDate', 'fileCloudPath', 'file_type', 'changeSummary', 'hash', 'effectiveDate', 'expiryDate'. If None, uid parameter must be provided.

uid: Optional unique identifier string to load an existing version from the database. Used when data is None to fetch version data via db.get_node_by_uid().

Return Value

Constructor returns a DocumentVersion instance. The create() class method returns Optional[DocumentVersion] - a new version instance if creation succeeds, None if it fails. Properties return various types: strings for version_number/status/paths, datetime objects for dates, DocUser for author, bool for is_current, Optional types when values may not exist. Methods like get_previous_version() return Optional[DocumentVersion], start_review() returns Optional[ReviewCycle], and to_dict() returns Dict[str, Any].

Class Interface

Methods

Attributes

Dependencies

• logging
• uuid
• hashlib
• typing
• datetime
• os
• CDocs.db
• CDocs.config.settings
• CDocs.db.schema_manager
• CDocs.models.BaseModel
• CDocs.models.register_model
• CDocs.models.user_extensions.DocUser
• CDocs.utils.audit_trail
• CDocs.models.document_status
• CDocs.models.review.ReviewCycle
• CDocs.controllers.filecloud_controller

Required Imports

import logging
import uuid
import hashlib
from typing import Dict, List, Any, Optional, Union, Tuple
from datetime import datetime, timedelta
import os
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.user_extensions import DocUser
from CDocs.utils import audit_trail
from CDocs.models.document_status import *
from CDocs.models.review import ReviewCycle

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs.controllers.filecloud_controller import get_filecloud_client

Usage Example

# Create a new document version
from CDocs.models.document_version import DocumentVersion
from CDocs.models.user_extensions import DocUser
from datetime import datetime

# Get the user creating the version
user = DocUser(uid='user-123')

# Create version with file paths
version = DocumentVersion.create(
    document_uid='doc-456',
    version_number='1.0',
    created_by=user,
    file_paths={'word': '/path/to/document.docx'},
    properties={'changeSummary': 'Initial version'}
)

if version:
    # Access version properties
    print(f"Version: {version.version_number}")
    print(f"Status: {version.get_status_name()}")
    print(f"Author: {version.author.name}")
    
    # Update status
    version.status = 'IN_REVIEW'
    
    # Start a review cycle
    reviewers = ['reviewer-uid-1', 'reviewer-uid-2']
    review = version.start_review(
        reviewers=reviewers,
        due_date=datetime.now() + timedelta(days=7),
        instructions='Please review for technical accuracy'
    )
    
    # Check file paths
    if version.has_pdf_version():
        pdf_path = version.pdf_file_path
        print(f"PDF available at: {pdf_path}")
    
    # Calculate hash for integrity
    with open('document.docx', 'rb') as f:
        content = f.read()
        hash_value = version.calculate_hash(content)
    
    # Load existing version
    existing = DocumentVersion(uid='version-uid-789')
    print(f"Is current: {existing.is_current}")
    
    # Get previous version
    prev = existing.get_previous_version()
    if prev:
        print(f"Previous version: {prev.version_number}")

Best Practices

• Always use the create() class method to instantiate new versions rather than direct constructor calls, as it properly sets up database relationships
• The class caches the parent document in _parent_doc - be aware this may become stale if the document is modified elsewhere
• File paths use a base path model: filecloud_path stores the base without extension, and word_file_path/pdf_file_path dynamically construct full paths
• When setting file paths, use the property setters (word_file_path or pdf_file_path) which automatically extract and store the base path
• Status changes should use the status property setter which validates against configured status codes
• The is_current property checks if this version is the current version of its parent document via CURRENT_VERSION relationship
• Review cycles are managed through start_review() which creates ReviewCycle instances and links them via FOR_REVIEW relationship
• Hash calculation should be done after file upload to verify integrity - use calculate_hash() with file content bytes
• The class maintains bidirectional relationships: version->document (HAS_VERSION), version->author (AUTHORED_BY), version->previous (PREVIOUS_VERSION)
• PDF file existence is verified dynamically in pdf_file_path getter - it may return None even if base path exists if PDF hasn't been generated
• Use to_dict() for serialization which includes enriched data like author details and document ID
• The get_previous_version() method follows PREVIOUS_VERSION relationships which are automatically created during version creation
• File type is stored separately and used to determine appropriate file extensions - defaults to 'docx' for Word files
• All database updates are immediately persisted via db.update_node() when using property setters

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class DocumentVersion_v1 83.2% similar

  Model representing a specific version of a controlled document.

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

• class DocumentVersion_v1 78.2% similar

  A dataclass that represents a versioned snapshot of a document, capturing its structure, metadata, and change history at a specific point in time.

  From: /tf/active/vicechatdev/vice_ai/models.py

• function create_document_version 74.5% similar

  Creates a new controlled document in a document management system with versioning, audit trails, and relationship management in a Neo4j graph database.

  From: /tf/active/vicechatdev/CDocs copy/controllers/document_controller.py

• function create_document_version_v1 72.8% similar

  Creates a new version of an existing document in a document management system, storing the file content in FileCloud and maintaining version history in Neo4j graph database.

  From: /tf/active/vicechatdev/CDocs copy/controllers/document_controller.py

• function create_document_version_v2 70.6% similar

  Creates a new version of an existing document in a document management system, storing the file in FileCloud and tracking version metadata in Neo4j graph database.

  From: /tf/active/vicechatdev/document_controller_backup.py