🔍 Code Extractor

function upload_document_to_filecloud_v1

Maturity: 76

Uploads a document version to FileCloud storage system with metadata, handling file creation, folder structure, and audit logging.

File:
/tf/active/vicechatdev/CDocs single class/controllers/filecloud_controller.py
Lines:
216 - 358
Complexity:
complex

Purpose

This function manages the complete workflow of uploading controlled documents to FileCloud, including document validation, folder structure creation, file upload, metadata attachment, and audit trail logging. It supports both new uploads and version updates, automatically handling metadata conversion and temporary file management. The function is designed for document management systems requiring version control and permission-based access.

Source Code

def upload_document_to_filecloud(
    user: DocUser,
    document: Union[ControlledDocument, str],
    file_content: bytes,
    file_path: Optional[str] = None,
    version_comment: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Upload a document version to FileCloud.
    
    Args:
        user: User performing the upload
        document: ControlledDocument instance or document UID
        file_content: Binary content of the file
        version_comment: Optional comment for version history
        metadata: Optional metadata to attach to the file
        
    Returns:
        Dictionary with upload results
        
    Raises:
        ResourceNotFoundError: If document not found
        PermissionError: If user doesn't have permission
        FileCloudError: If upload fails
    """
    # Get document instance if UID provided
    if isinstance(document, str):
        doc = ControlledDocument(uid=document)
        if not doc:
            raise ResourceNotFoundError(f"Document not found: {document}")
    else:
        doc = document
    
    # Get current version to determine file details
    version = doc.current_version
    if not version:
        raise ResourceNotFoundError(f"No current version found for document {doc.uid}")
    
    # Ensure folder structure exists
    ensure_document_folders(doc)
    
    # Determine file path in FileCloud
    if not file_path:
        file_path = get_filecloud_document_path(doc)
    folder_path, filename = os.path.split(file_path)
    
    # Prepare metadata if not provided
    if metadata is None:
        metadata = {
            "cdoc_uid": doc.uid,
            "doc_number": doc.doc_number,
            "doc_uid": version.uid,
            "version_number": version.version_number,  # Convert to string
            "doc_type": doc.doc_type,
            "department": doc.department,
            "is_cdoc": "true",  # Use string instead of boolean
            "status": doc.status,
            "owner": doc.owner.name if doc.owner else "Unknown"
        }
    else:
        # Convert any non-string values to strings
        for key, value in metadata.items():
            if value is None:
                metadata[key] = ""
            elif not isinstance(value, str):
                if isinstance(value, bool):
                    metadata[key] = "true" if value else "false"
                else:
                    metadata[key] = str(value)
    
    # Create a temporary file for upload
    with tempfile.NamedTemporaryFile(suffix='.'+version.file_type, delete=False) as temp_file:
        temp_path = temp_file.name
        temp_file.write(file_content)
    
    try:
        client = get_filecloud_client()
        
        # Upload the file
        result = client.upload_file(
            local_file_path=temp_path,
            remote_path=folder_path,
            filename=filename,
            overwrite=True
        )
        
        if not result.get('success', False):
            logger.error(f"Failed to upload file to FileCloud: {result.get('message', 'Unknown error')}")
            raise FileCloudError(f"Failed to upload file: {result.get('message', 'Unknown error')}")
        
        
        # Add metadata to file
        catalog = MetadataCatalog(client, debug=True)  # Enable debug mode
        catalog.initialize()
        
        # # First add the metadata set to the file
        # set_result = catalog.add_set_to_file_object(file_path, set_name="CDocs")
        # if not set_result.get('success', False):
        #     logger.warning(f"Failed to add metadata set to file: {set_result.get('message', 'Unknown error')}")
        #     # Continue despite this error - the set might already be attached
        
        # # Wait briefly after adding the set
        # time.sleep(1)
        
        # Then save attribute values
        metadata_result = catalog.save_attribute_values_by_name(file_path, "CDocs", metadata)
        if not metadata_result.get('success', False):
            logger.warning(f"Failed to set metadata on file: {metadata_result.get('message', 'Unknown error')}")
            logger.debug(f"Metadata that failed: {metadata}")
            logger.debug(f"Raw response: {metadata_result.get('raw_response')}")
        
        # Log audit event
        audit_trail.log_version_event(
            event_type="VERSION_UPDATED" if version_comment else "VERSION_CREATED",
            user=user,
            version_uid=version.uid,
            details={
                "file_path": file_path,
                "comment": version_comment,
                "metadata": metadata
            }
        )
        
        return {
            "success": True,
            "document_uid": doc.uid,
            "version_uid": version.uid,
            "file_path": file_path,
            "metadata": metadata
        }
        
    except FileCloudError:
        raise
    except Exception as e:
        logger.error(f"Error uploading document to FileCloud: {e}")
        raise FileCloudError(f"Error uploading document: {e}")
    finally:
        # Clean up temporary file
        try:
            os.unlink(temp_path)
        except:
            pass

Parameters

Name Type Default Kind
user DocUser - positional_or_keyword
document Union[ControlledDocument, str] - positional_or_keyword
file_content bytes - positional_or_keyword
file_path Optional[str] None positional_or_keyword
version_comment Optional[str] None positional_or_keyword
metadata Optional[Dict[str, Any]] None positional_or_keyword

Parameter Details

user: DocUser instance representing the authenticated user performing the upload. Used for permission checks and audit logging.

document: Either a ControlledDocument instance or a string UID of the document. If a UID string is provided, the function will retrieve the document instance. Must reference an existing document with a current version.

file_content: Binary content (bytes) of the file to be uploaded. This is the actual file data that will be written to FileCloud.

file_path: Optional string specifying the remote path in FileCloud where the file should be stored. If not provided, the path is automatically determined using get_filecloud_document_path(). Should include folder path and filename.

version_comment: Optional string comment describing the version changes. Used for audit logging and version history tracking. If provided, triggers a VERSION_UPDATED event instead of VERSION_CREATED.

metadata: Optional dictionary containing custom metadata key-value pairs to attach to the file in FileCloud. If not provided, default metadata is generated from document properties including cdoc_uid, doc_number, version_number, doc_type, department, status, and owner. All values are converted to strings.

Return Value

Type: Dict[str, Any]

Returns a dictionary with keys: 'success' (boolean, always True if no exception), 'document_uid' (string, UID of the document), 'version_uid' (string, UID of the document version), 'file_path' (string, remote path where file was uploaded), and 'metadata' (dict, the metadata that was attached to the file). If the function fails, it raises an exception rather than returning a failure status.

Dependencies

  • logging
  • os
  • json
  • uuid
  • tempfile
  • typing
  • time
  • CDocs
  • FC_api

Required Imports

import logging
import os
import tempfile
import time
from typing import Dict, List, Any, Optional, Union
from CDocs.models.document import ControlledDocument, DocumentVersion
from CDocs.models.user_extensions import DocUser
from CDocs.utils import audit_trail
from CDocs.controllers import require_permission, log_controller_action, ResourceNotFoundError
from CDocs.utils.metadata_catalog import MetadataCatalog
from FC_api import FileCloudAPI

Usage Example

from CDocs.models.user_extensions import DocUser
from CDocs.models.document import ControlledDocument
from CDocs.controllers.document_controller import upload_document_to_filecloud

# Get user and document instances
user = DocUser.query.get(user_id)
document = ControlledDocument.query.filter_by(uid='DOC-12345').first()

# Read file content
with open('path/to/document.pdf', 'rb') as f:
    file_content = f.read()

# Upload with custom metadata
result = upload_document_to_filecloud(
    user=user,
    document=document,
    file_content=file_content,
    version_comment='Updated with latest revisions',
    metadata={
        'custom_field': 'custom_value',
        'review_status': 'approved'
    }
)

print(f"Upload successful: {result['file_path']}")
print(f"Version UID: {result['version_uid']}")

Best Practices

  • Ensure the user has both UPLOAD_DOCUMENT and EDIT_DOCUMENT permissions before calling this function (enforced by decorator)
  • Always provide file_content as bytes, not string or file object
  • The function creates temporary files that are automatically cleaned up, but ensure sufficient disk space is available
  • Metadata values are automatically converted to strings; booleans become 'true'/'false', None becomes empty string
  • The function will overwrite existing files at the same path in FileCloud
  • Handle ResourceNotFoundError if the document or version doesn't exist
  • Handle FileCloudError for upload failures or network issues
  • The function logs audit events automatically; no additional logging needed for version tracking
  • If providing custom metadata, ensure all keys match the CDocs metadata set schema in FileCloud
  • The document must have a current_version set before calling this function

Similar Components

AI-powered semantic similarity - components with related functionality:

  • function upload_document_to_filecloud 99.5% similar

    Uploads a document version to FileCloud storage system with metadata, handling file creation, folder structure, and audit logging.

    From: /tf/active/vicechatdev/CDocs/controllers/filecloud_controller.py
  • function create_document_version_v3 82.2% similar

    Creates a new version of a controlled document by generating version metadata, storing the file in FileCloud, updating the document's revision number, and creating an audit trail entry.

    From: /tf/active/vicechatdev/CDocs/controllers/document_controller.py
  • function create_document_version_v2 80.7% 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
  • function update_document_metadata_in_filecloud 80.5% similar

    Updates metadata for a document stored in FileCloud, merging new metadata with existing values and logging the update in an audit trail.

    From: /tf/active/vicechatdev/CDocs/controllers/filecloud_controller.py
  • function create_document_version_v1 79.1% 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
← Back to Browse