🔍 Code Extractor

Retrieves comprehensive system statistics from a Neo4j graph database for display on an admin dashboard, including user counts, document counts, review cycles, and approval metrics.

Purpose

This function aggregates key system metrics by executing multiple Cypher queries against a Neo4j database. It provides administrators with a snapshot of system health and activity, including total and active users, documents (excluding archived/obsolete), pending reviews, pending approvals, and system uptime. The function is designed to be resilient, returning zero values for all metrics if any errors occur during data retrieval.

Source Code

def get_system_stats() -> Dict[str, Any]:
    """Get system statistics for the admin dashboard."""
    try:
        # Query database for stats
        stats = {}
        
        # Use datetime.datetime.now() instead of datetime.now()
        from datetime import datetime
        
        # User stats
        user_query = """
        MATCH (u:User) 
        RETURN count(u) as total_users,
               count(u) as active_users
        """
        user_result = db.run_query(user_query)
        if user_result:
            stats['total_users'] = user_result[0].get('total_users', 0)
            stats['active_users'] = user_result[0].get('active_users', 0)
        else:
            stats['total_users'] = 0
            stats['active_users'] = 0
        
        # Document stats
        doc_query = """
        MATCH (d:ControlledDocument)
        RETURN count(d) as total_docs,
               count(CASE WHEN d.status <> 'ARCHIVED' AND d.status <> 'OBSOLETE' THEN d END) as active_docs
        """
        doc_result = db.run_query(doc_query)
        if doc_result:
            stats['total_documents'] = doc_result[0].get('total_docs', 0)
            stats['active_documents'] = doc_result[0].get('active_docs', 0)
        else:
            stats['total_documents'] = 0
            stats['active_documents'] = 0
        
        # Review stats
        review_query = """
        MATCH (r:ReviewCycle)
        RETURN count(r) as total_reviews,
               count(CASE WHEN r.status = 'IN_REVIEW' THEN r END) as pending_reviews
        """
        review_result = db.run_query(review_query)
        if review_result:
            stats['total_reviews'] = review_result[0].get('total_reviews', 0)
            stats['pending_reviews'] = review_result[0].get('pending_reviews', 0)
        else:
            stats['total_reviews'] = 0
            stats['pending_reviews'] = 0
        
        # Approval stats
        approval_query = """
        MATCH (a:Approval)
        RETURN count(a) as total_approvals,
               count(CASE WHEN a.status = 'PENDING' THEN a END) as pending_approvals
        """
        approval_result = db.run_query(approval_query)
        if approval_result:
            stats['total_approvals'] = approval_result[0].get('total_approvals', 0)
            stats['pending_approvals'] = approval_result[0].get('pending_approvals', 0)
        else:
            stats['total_approvals'] = 0
            stats['pending_approvals'] = 0
        
        # Add uptime (mock for now)
        # In a real implementation, this would come from system metrics
        stats['uptime'] = int((datetime.now() - datetime(2023, 1, 1)).total_seconds())
        
        return stats
    except Exception as e:
        logger.error(f"Error getting system stats: {e}")
        return {
            'total_users': 0,
            'active_users': 0,
            'total_documents': 0,
            'active_documents': 0,
            'total_reviews': 0,
            'pending_reviews': 0,
            'total_approvals': 0,
            'pending_approvals': 0,
            'uptime': 0
        }

Return Value

Type: Dict[str, Any]

Returns a dictionary containing system statistics with the following keys: 'total_users' (int: total count of User nodes), 'active_users' (int: count of active users), 'total_documents' (int: total count of ControlledDocument nodes), 'active_documents' (int: count of documents not archived or obsolete), 'total_reviews' (int: total count of ReviewCycle nodes), 'pending_reviews' (int: count of reviews with status 'IN_REVIEW'), 'total_approvals' (int: total count of Approval nodes), 'pending_approvals' (int: count of approvals with status 'PENDING'), and 'uptime' (int: system uptime in seconds, currently mocked as time since 2023-01-01). All values default to 0 if errors occur.

Dependencies

• logging
• datetime
• typing
• CDocs.db
• CDocs.models.document
• CDocs.models.user_extensions
• CDocs.db.schema_manager
• CDocs.config
• CDocs.utils.audit_trail

Required Imports

from typing import Dict, Any
from datetime import datetime
import logging

Conditional/Optional Imports

These imports are only needed under specific conditions:

from datetime import datetime

Usage Example

import logging
from typing import Dict, Any
from datetime import datetime
from CDocs import db

# Ensure logger is configured
logger = logging.getLogger(__name__)

# Call the function to get system statistics
stats = get_system_stats()

# Access the statistics
print(f"Total Users: {stats['total_users']}")
print(f"Active Documents: {stats['active_documents']}")
print(f"Pending Reviews: {stats['pending_reviews']}")
print(f"Pending Approvals: {stats['pending_approvals']}")
print(f"System Uptime (seconds): {stats['uptime']}")

# Use in a dashboard context
if stats['pending_approvals'] > 10:
    print("Warning: High number of pending approvals!")

Best Practices

• The function includes comprehensive error handling and returns safe default values (all zeros) if any database query fails
• Each database query is executed independently, so partial failures won't prevent other statistics from being retrieved
• The uptime calculation is currently mocked with a hardcoded start date (2023-01-01); in production, this should be replaced with actual system start time tracking
• Ensure the 'logger' variable is properly initialized in the module scope before calling this function
• The function uses CASE expressions in Cypher queries for conditional counting, which is efficient for aggregating filtered results
• Consider caching the results of this function if called frequently, as it executes multiple database queries
• The function assumes db.run_query() returns a list of records; verify this matches your database wrapper implementation
• For large databases, consider adding query timeouts or pagination to prevent performance issues

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_review_statistics 77.3% similar

  Retrieves comprehensive review statistics from a Neo4j graph database, including counts of review cycles, approval rates, and reviewer decisions.

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

• function get_document_stats 74.7% similar

  Retrieves aggregated statistics about controlled documents from a Neo4j database, including status and type distributions for visualization in charts.

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

• function get_approval_statistics_v1 73.5% similar

  Retrieves comprehensive approval statistics from a Neo4j graph database, including counts of total, pending, completed, approved, and rejected approval cycles and decisions, along with calculated approval rates.

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

• function get_database_statistics 70.5% similar

  Retrieves statistical information about a Neo4j graph database, including counts of nodes grouped by label and counts of relationships grouped by type.

  From: /tf/active/vicechatdev/neo4j_schema/neo4j_python_snippets.py

• function get_users 67.3% similar

  Retrieves user information from a Neo4j graph database for display in an admin panel, with optional filtering by user ID.

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