🔍 Code Extractor

Performs a system-wide startup check to verify and update document permissions for all documents in the system, ensuring permissions align with document status.

Purpose

This function is designed to run on system startup to audit and correct document permissions across the entire document management system. It retrieves all documents, validates their current permissions using the share_controller's manage_document_permissions function, and creates or updates FileCloud shares as needed. The function handles missing shares, updates user permissions based on document status, and provides comprehensive logging and statistics about the operation.

Source Code

def check_document_permissions_on_startup() -> Dict[str, Any]:
    """
    Check and update permissions for all documents on system startup.
    
    This function uses the manage_document_permissions function from share_controller
    to ensure that all documents have correct permissions based on their current status.
    
    Returns:
        Dict with results of permission check
    """
    try:
        start_time = time.time()
        logger.info("Starting document permission check on system startup")
        
        # Get all documents in the system
        from CDocs.controllers.document_controller import get_documents
        # Create admin user to get all documents

        from CDocs.models.user_extensions import DocUser
        admin_user = DocUser.get_by_email('wim@vicebio.com')
        if not admin_user:
            logger.warning("Admin user not found, creating a temporary admin user for permission check")
            admin_user = DocUser({
                'UID': 'temp-admin',
                'username': 'admin',
                'role': 'ADMIN'
            })
        
        # Get all documents with a high limit to ensure we get everything
        docs_result = get_documents(
            user=admin_user,
            limit=10000  # High limit to get all documents
        )
        
        all_documents = []
        if docs_result.get('success', False):
            # Convert dictionary representations to ControlledDocument objects
            for doc_dict in docs_result.get('documents', []):
                doc = ControlledDocument(doc_dict)
                if doc.uid:  # Only add valid documents
                    all_documents.append(doc)
        
        results = {
            'success': True,
            'total_documents': len(all_documents),
            'updated_count': 0,
            'failed_count': 0,
            'versions_with_missing_attributes': 0,
            'versions_with_missing_shares': 0,
            'elapsed_time': 0,
            'details': []
        }
        
        # Process each document - using manage_document_permissions from share_controller
        from CDocs.controllers.share_controller import manage_document_permissions
        
        for document in all_documents:
            try:
                # Skip documents with no current version
                if not document.current_version:
                    logger.warning(f"Document {document.uid} ({document.title}) has no current version")
                    continue
                
                # Ensure document version has required share attributes
                # version_updated = False
                # if not hasattr(document.current_version, 'share_id') or document.current_version.share_id is None or \
                #    not hasattr(document.current_version, 'share_url') or document.current_version.share_url is None:
                #     version_updated = ensure_document_version_share_attributes(document.current_version)
                #     if version_updated:
                #         results['versions_with_missing_attributes'] += 1
                
                logger.info(f"Checking permissions for document: {document.title} [{document.uid}] (status: {document.status})")
                
                # Use manage_document_permissions to handle all permission logic
                permission_result = manage_document_permissions(document)
                
                # Process results from manage_document_permissions
                if permission_result.get('success', False):
                    # Count this as an update if any changes were made
                    if (permission_result.get('users_added', 0) > 0 or 
                        permission_result.get('users_updated', 0) > 0 or 
                        permission_result.get('users_removed', 0) > 0 or
                        permission_result.get('share_created', False)):
                        results['updated_count'] += 1
                    
                    # Check if a share was created
                    if permission_result.get('share_created', False):
                        results['versions_with_missing_shares'] += 1
                    
                    # Add the results to our details
                    result_entry = {
                        'document_id': document.uid,
                        'title': document.title,
                        'status': document.status,
                        'success': True,
                        'share_id': permission_result.get('share_id'),
                        'share_url': permission_result.get('share_url'),
                        'users_added': permission_result.get('users_added', 0),
                        'users_updated': permission_result.get('users_updated', 0),
                        'users_removed': permission_result.get('users_removed', 0),
                        'message': permission_result.get('message', 'Permissions updated')
                    }
                    results['details'].append(result_entry)
                else:
                    # Count failures
                    results['failed_count'] += 1
                    results['details'].append({
                        'document_id': document.uid,
                        'title': document.title,
                        'status': document.status,
                        'success': False,
                        'message': permission_result.get('message', 'Failed to update permissions')
                    })
                
            except Exception as e:
                logger.error(f"Error checking permissions for document {document.uid}: {str(e)}")
                import traceback
                logger.error(traceback.format_exc())
                results['failed_count'] += 1
                results['details'].append({
                    'document_id': document.uid,
                    'title': document.title if hasattr(document, 'title') else 'Unknown',
                    'success': False,
                    'message': f"Error: {str(e)}"
                })
        
        logger.info(f"Document permission check completed. Updated: {results['updated_count']}, Failed: {results['failed_count']}")
        
        # Record elapsed time
        results['elapsed_time'] = time.time() - start_time
        logger.info(f"Permission check completed in {results['elapsed_time']:.2f} seconds")
        
        # Log statistics about missing attributes
        if results['versions_with_missing_attributes'] > 0:
            logger.warning(f"Added missing share attributes to {results['versions_with_missing_attributes']} document versions")
            
        if results['versions_with_missing_shares'] > 0:
            logger.warning(f"Created missing shares for {results['versions_with_missing_shares']} document versions")
        
        return results
        
    except Exception as e:
        logger.error(f"Error during document permission startup check: {str(e)}")
        import traceback
        logger.error(traceback.format_exc())
        return {
            'success': False,
            'message': f"Error during permission check: {str(e)}"
        }

Return Value

Type: Dict[str, Any]

Returns a dictionary containing operation results with keys: 'success' (bool indicating overall success), 'total_documents' (int count of documents processed), 'updated_count' (int count of documents with permission changes), 'failed_count' (int count of failed operations), 'versions_with_missing_attributes' (int count of versions that had missing share attributes), 'versions_with_missing_shares' (int count of versions that needed share creation), 'elapsed_time' (float seconds taken), 'details' (list of dicts with per-document results including document_id, title, status, success flag, share_id, share_url, users_added, users_updated, users_removed, and message), and 'message' (str error message if success is False).

Dependencies

• logging
• time
• threading
• typing
• CDocs.models.document
• CDocs.models.user_extensions
• CDocs.controllers.document_controller
• CDocs.controllers.share_controller
• CDocs.controllers.filecloud_controller
• CDocs.models.document_status
• traceback

Required Imports

import logging
import time
import threading
from typing import Dict, Any, List, Optional, Union
from CDocs.models.document import ControlledDocument, DocumentVersion
from CDocs.models.user_extensions import DocUser
from CDocs.controllers.share_controller import manage_document_permissions
from CDocs.controllers.filecloud_controller import get_filecloud_client, get_filecloud_document_path
from CDocs.models.document_status import STATUS_DRAFT, STATUS_IN_REVIEW, STATUS_IN_APPROVAL, STATUS_APPROVED, STATUS_PUBLISHED, STATUS_EFFECTIVE, STATUS_ARCHIVED, STATUS_OBSOLETE
import traceback

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs.controllers.document_controller import get_documents

from CDocs.models.user_extensions import DocUser

from CDocs.controllers.share_controller import manage_document_permissions

import traceback

Usage Example

# This function is typically called during application startup
# Example usage in application initialization:

import logging
from your_module import check_document_permissions_on_startup

# Configure logging
logging.basicConfig(level=logging.INFO)

# Call during system startup
results = check_document_permissions_on_startup()

# Check results
if results['success']:
    print(f"Permission check completed successfully")
    print(f"Total documents: {results['total_documents']}")
    print(f"Updated: {results['updated_count']}")
    print(f"Failed: {results['failed_count']}")
    print(f"Time elapsed: {results['elapsed_time']:.2f} seconds")
    
    # Review details for specific documents
    for detail in results['details']:
        if not detail['success']:
            print(f"Failed document: {detail['title']} - {detail['message']}")
else:
    print(f"Permission check failed: {results.get('message', 'Unknown error')}")

Best Practices

• This function should only be called during system startup or maintenance windows as it processes all documents in the system
• Ensure adequate system resources as the function retrieves and processes up to 10,000 documents
• Monitor the elapsed_time in results to track performance and identify potential bottlenecks
• Review the 'details' array in results to identify specific documents that failed permission updates
• The function requires an admin user (wim@vicebio.com) to exist; ensure this user is created during initial system setup
• Consider implementing rate limiting or batching if the document count exceeds 10,000
• The function logs extensively; ensure log rotation is configured to prevent disk space issues
• Failed permission updates are logged but don't stop the overall process; review failed_count and details after execution
• The function creates a temporary admin user if the expected admin is not found, but this should be avoided in production
• Consider running this function asynchronously or in a background thread to avoid blocking application startup

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function check_document_permissions_on_startup 80.5% similar

  Validates and fixes document permission issues during application startup, prioritizing active documents (DRAFT, IN_REVIEW, IN_APPROVAL) to ensure proper sharing permissions are configured.

  From: /tf/active/vicechatdev/CDocs/utils/sharing_validator.py

• function run_permission_check_async 73.4% similar

  Launches a background daemon thread to perform document permission checks without blocking the main application thread.

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

• function manage_document_permissions 68.7% similar

  Comprehensive function to manage document sharing and user permissions. This function: 1. Creates a share only if needed for active users 2. Adds/updates users with appropriate permissions based on their roles 3. Removes users who shouldn't have access anymore 4. Cleans up shares that are no longer needed 5. Manages ACL entries for write permissions on the document's folder Args: document: The document to manage permissions for Returns: Dict: Result of permission updates with detailed information

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

• function initialize_system 66.5% similar

  Initializes the CDocs document management system by setting up database connections, FileCloud integration, document sharing validation, and API routes.

  From: /tf/active/vicechatdev/CDocs/initialize_system.py

• function set_document_permissions_in_filecloud 63.1% similar

  Sets user and group permissions for a document in FileCloud by updating access controls on the document's folder path and logging the permission changes to an audit trail.

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