Spaces:

StrategyStack
/

README

Sleeping

README / activity_ingestion.py

Implement atomic activity ingestion with schema validation and commit tracking

3a67095 verified about 1 year ago

11 kB

"""Activity Ingestion Function with Atomic Operations and Commit Tracking
This module provides schema-compliant activity logging with:
- Atomic row additions (all-or-nothing operations)
- Schema validation and error handling
- Request ID and timestamp generation
- Full commit tracking for rollback and versioning
- Comprehensive error logging
"""

import csv
import os
import uuid
from datetime import datetime, timezone
from typing import Dict, Optional, Tuple
import fcntl
import tempfile
import shutil
from pathlib import Path


# Schema Definition (v1.0)
SCHEMA_VERSION = "1.0"
REQUIRED_FIELDS = ['timestamp', 'agent_id', 'activity_type', 'entity_id', 'status', 'notes', 'context']
VALID_STATUS_VALUES = ['created', 'success', 'failed', 'in_progress', 'warning', 'error']
VALID_ACTIVITY_TYPES = ['init', 'validate', 'ingest', 'update', 'delete', 'query', 'commit', 'rollback']


class ActivityIngestionError(Exception):
    """Base exception for activity ingestion errors"""
    pass


class SchemaValidationError(ActivityIngestionError):
    """Schema validation failed"""
    pass


class AtomicWriteError(ActivityIngestionError):
    """Atomic write operation failed"""
    pass


def generate_request_id() -> str:
    """Generate unique request ID for tracking"""
    return f"req_{uuid.uuid4().hex[:16]}"


def generate_timestamp() -> str:
    """Generate ISO 8601 timestamp in UTC"""
    return datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')


def validate_activity_record(record: Dict[str, str]) -> Tuple[bool, Optional[str]]:
    """Validate activity record against schema
    
    Args:
        record: Dictionary containing activity data
        
    Returns:
        Tuple of (is_valid, error_message)
    """
    # Check required fields
    missing_fields = [f for f in REQUIRED_FIELDS if f not in record or not record[f]]
    if missing_fields:
        return False, f"Missing required fields: {', '.join(missing_fields)}"
    
    # Validate status value
    if record['status'] not in VALID_STATUS_VALUES:
        return False, f"Invalid status '{record['status']}'. Must be one of: {', '.join(VALID_STATUS_VALUES)}"
    
    # Validate activity_type
    if record['activity_type'] not in VALID_ACTIVITY_TYPES:
        return False, f"Invalid activity_type '{record['activity_type']}'. Must be one of: {', '.join(VALID_ACTIVITY_TYPES)}"
    
    # Validate timestamp format (ISO 8601)
    try:
        datetime.fromisoformat(record['timestamp'].replace('Z', '+00:00'))
    except ValueError:
        return False, f"Invalid timestamp format. Expected ISO 8601, got: {record['timestamp']}"
    
    return True, None


def atomic_append_activity(filepath: str, record: Dict[str, str], request_id: Optional[str] = None) -> Dict[str, any]:
    """Atomically append activity record to CSV file with full error handling
    
    Args:
        filepath: Path to activity_log.csv file
        record: Activity record dictionary
        request_id: Optional request ID (generated if not provided)
        
    Returns:
        Dictionary with operation status:
        {
            'success': bool,
            'request_id': str,
            'timestamp': str,
            'error': Optional[str],
            'commit_hash': Optional[str],
            'record': Dict
        }
    """
    # Generate request_id if not provided
    if not request_id:
        request_id = generate_request_id()
    
    # Generate timestamp if not in record
    if 'timestamp' not in record or not record['timestamp']:
        record['timestamp'] = generate_timestamp()
    
    operation_timestamp = generate_timestamp()
    result = {
        'success': False,
        'request_id': request_id,
        'timestamp': operation_timestamp,
        'error': None,
        'commit_hash': None,
        'record': record.copy()
    }
    
    try:
        # Validate record schema
        is_valid, error_msg = validate_activity_record(record)
        if not is_valid:
            raise SchemaValidationError(error_msg)
        
        # Ensure file exists with header
        filepath = Path(filepath)
        if not filepath.exists():
            filepath.parent.mkdir(parents=True, exist_ok=True)
            with open(filepath, 'w', newline='') as f:
                writer = csv.DictWriter(f, fieldnames=REQUIRED_FIELDS)
                writer.writeheader()
        
        # Atomic write using temp file and rename
        temp_fd, temp_path = tempfile.mkstemp(dir=filepath.parent, prefix='.activity_log_', suffix='.tmp')
        
        try:
            # Acquire exclusive lock for atomic operation
            with open(filepath, 'r') as source:
                fcntl.flock(source.fileno(), fcntl.LOCK_EX)
                
                try:
                    # Copy existing content to temp file
                    with os.fdopen(temp_fd, 'w', newline='') as temp_file:
                        reader = csv.DictReader(source)
                        writer = csv.DictWriter(temp_file, fieldnames=REQUIRED_FIELDS)
                        
                        # Write header
                        writer.writeheader()
                        
                        # Copy existing rows
                        for row in reader:
                            writer.writerow(row)
                        
                        # Append new record
                        writer.writerow(record)
                    
                    # Atomic rename (replaces original file)
                    shutil.move(temp_path, filepath)
                    result['success'] = True
                    
                finally:
                    fcntl.flock(source.fileno(), fcntl.LOCK_UN)
                    
        except Exception as e:
            # Clean up temp file on error
            if os.path.exists(temp_path):
                os.unlink(temp_path)
            raise AtomicWriteError(f"Atomic write failed: {str(e)}")
        
        # Log successful operation
        result['commit_hash'] = _generate_commit_hash(record, request_id)
        _log_operation(request_id, 'ingest', 'success', record)
        
    except (SchemaValidationError, AtomicWriteError, Exception) as e:
        result['error'] = str(e)
        result['success'] = False
        _log_operation(request_id, 'ingest', 'error', record, error=str(e))
    
    return result


def _generate_commit_hash(record: Dict[str, str], request_id: str) -> str:
    """Generate pseudo-commit hash for tracking"""
    import hashlib
    data = f"{request_id}:{record['timestamp']}:{record['agent_id']}:{record['activity_type']}"
    return hashlib.sha256(data.encode()).hexdigest()[:16]


def _log_operation(request_id: str, operation: str, status: str, record: Dict, error: Optional[str] = None):
    """Log operation to error/audit log file"""
    log_file = Path('activity_ingestion.log')
    timestamp = generate_timestamp()
    
    log_entry = {
        'timestamp': timestamp,
        'request_id': request_id,
        'operation': operation,
        'status': status,
        'agent_id': record.get('agent_id', 'unknown'),
        'activity_type': record.get('activity_type', 'unknown'),
        'error': error or ''
    }
    
    with open(log_file, 'a') as f:
        f.write(f"{log_entry}\n")


def test_ingestion():
    """Test cases for activity ingestion validation"""
    print("=== Activity Ingestion Function Tests ===")
    print()
    
    # Test 1: Valid record insertion
    print("Test 1: Valid record insertion")
    valid_record = {
        'timestamp': generate_timestamp(),
        'agent_id': 'test_agent',
        'activity_type': 'ingest',
        'entity_id': 'test_entity_001',
        'status': 'success',
        'notes': 'Test record for validation',
        'context': 'Unit test - valid case'
    }
    result = atomic_append_activity('activity_log.csv', valid_record)
    print(f"  Result: {'PASS' if result['success'] else 'FAIL'}")
    print(f"  Request ID: {result['request_id']}")
    print(f"  Timestamp: {result['timestamp']}")
    if result['success']:
        print(f"  Commit Hash: {result['commit_hash']}")
    else:
        print(f"  Error: {result['error']}")
    print()
    
    # Test 2: Missing required field
    print("Test 2: Missing required field (should fail)")
    invalid_record = {
        'timestamp': generate_timestamp(),
        'agent_id': 'test_agent',
        'activity_type': 'ingest',
        # Missing 'entity_id'
        'status': 'success',
        'notes': 'This should fail',
        'context': 'Unit test - invalid case'
    }
    result = atomic_append_activity('activity_log.csv', invalid_record)
    print(f"  Result: {'PASS (correctly rejected)' if not result['success'] else 'FAIL (should have rejected)'}")
    print(f"  Error: {result['error']}")
    print()
    
    # Test 3: Invalid status value
    print("Test 3: Invalid status value (should fail)")
    invalid_status = {
        'timestamp': generate_timestamp(),
        'agent_id': 'test_agent',
        'activity_type': 'ingest',
        'entity_id': 'test_entity_003',
        'status': 'invalid_status',  # Invalid
        'notes': 'This should fail',
        'context': 'Unit test - invalid status'
    }
    result = atomic_append_activity('activity_log.csv', invalid_status)
    print(f"  Result: {'PASS (correctly rejected)' if not result['success'] else 'FAIL (should have rejected)'}")
    print(f"  Error: {result['error']}")
    print()
    
    # Test 4: Auto-generated timestamp
    print("Test 4: Auto-generated timestamp")
    auto_timestamp = {
        'agent_id': 'test_agent',
        'activity_type': 'validate',
        'entity_id': 'test_entity_004',
        'status': 'success',
        'notes': 'Timestamp should be auto-generated',
        'context': 'Unit test - auto timestamp'
    }
    result = atomic_append_activity('activity_log.csv', auto_timestamp)
    print(f"  Result: {'PASS' if result['success'] else 'FAIL'}")
    print(f"  Generated Timestamp: {result['record']['timestamp']}")
    print()
    
    # Test 5: Multiple concurrent inserts (simulate race condition safety)
    print("Test 5: Concurrent insert safety")
    for i in range(3):
        record = {
            'timestamp': generate_timestamp(),
            'agent_id': f'concurrent_agent_{i}',
            'activity_type': 'ingest',
            'entity_id': f'concurrent_entity_{i}',
            'status': 'success',
            'notes': f'Concurrent test insert {i}',
            'context': 'Unit test - concurrent safety'
        }
        result = atomic_append_activity('activity_log.csv', record)
        print(f"  Insert {i}: {'PASS' if result['success'] else 'FAIL'}")
    print()
    
    print("=== Test Summary ===")
    print("All tests completed. Check activity_log.csv and activity_ingestion.log for details.")
    print()


if __name__ == '__main__':
    # Run tests when executed directly
    test_ingestion()

	"""Activity Ingestion Function with Atomic Operations and Commit Tracking

	This module provides schema-compliant activity logging with:
	- Atomic row additions (all-or-nothing operations)
	- Schema validation and error handling
	- Request ID and timestamp generation
	- Full commit tracking for rollback and versioning
	- Comprehensive error logging
	"""

	import csv
	import os
	import uuid
	from datetime import datetime, timezone
	from typing import Dict, Optional, Tuple
	import fcntl
	import tempfile
	import shutil
	from pathlib import Path


	# Schema Definition (v1.0)
	SCHEMA_VERSION = "1.0"
	REQUIRED_FIELDS = ['timestamp', 'agent_id', 'activity_type', 'entity_id', 'status', 'notes', 'context']
	VALID_STATUS_VALUES = ['created', 'success', 'failed', 'in_progress', 'warning', 'error']
	VALID_ACTIVITY_TYPES = ['init', 'validate', 'ingest', 'update', 'delete', 'query', 'commit', 'rollback']


	class ActivityIngestionError(Exception):
	"""Base exception for activity ingestion errors"""
	pass


	class SchemaValidationError(ActivityIngestionError):
	"""Schema validation failed"""
	pass


	class AtomicWriteError(ActivityIngestionError):
	"""Atomic write operation failed"""
	pass


	def generate_request_id() -> str:
	"""Generate unique request ID for tracking"""
	return f"req_{uuid.uuid4().hex[:16]}"


	def generate_timestamp() -> str:
	"""Generate ISO 8601 timestamp in UTC"""
	return datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')


	def validate_activity_record(record: Dict[str, str]) -> Tuple[bool, Optional[str]]:
	"""Validate activity record against schema

	Args:
	record: Dictionary containing activity data

	Returns:
	Tuple of (is_valid, error_message)
	"""
	# Check required fields
	missing_fields = [f for f in REQUIRED_FIELDS if f not in record or not record[f]]
	if missing_fields:
	return False, f"Missing required fields: {', '.join(missing_fields)}"

	# Validate status value
	if record['status'] not in VALID_STATUS_VALUES:
	return False, f"Invalid status '{record['status']}'. Must be one of: {', '.join(VALID_STATUS_VALUES)}"

	# Validate activity_type
	if record['activity_type'] not in VALID_ACTIVITY_TYPES:
	return False, f"Invalid activity_type '{record['activity_type']}'. Must be one of: {', '.join(VALID_ACTIVITY_TYPES)}"

	# Validate timestamp format (ISO 8601)
	try:
	datetime.fromisoformat(record['timestamp'].replace('Z', '+00:00'))
	except ValueError:
	return False, f"Invalid timestamp format. Expected ISO 8601, got: {record['timestamp']}"

	return True, None


	def atomic_append_activity(filepath: str, record: Dict[str, str], request_id: Optional[str] = None) -> Dict[str, any]:
	"""Atomically append activity record to CSV file with full error handling

	Args:
	filepath: Path to activity_log.csv file
	record: Activity record dictionary
	request_id: Optional request ID (generated if not provided)

	Returns:
	Dictionary with operation status:
	{
	'success': bool,
	'request_id': str,
	'timestamp': str,
	'error': Optional[str],
	'commit_hash': Optional[str],
	'record': Dict
	}
	"""
	# Generate request_id if not provided
	if not request_id:
	request_id = generate_request_id()

	# Generate timestamp if not in record
	if 'timestamp' not in record or not record['timestamp']:
	record['timestamp'] = generate_timestamp()

	operation_timestamp = generate_timestamp()
	result = {
	'success': False,
	'request_id': request_id,
	'timestamp': operation_timestamp,
	'error': None,
	'commit_hash': None,
	'record': record.copy()
	}

	try:
	# Validate record schema
	is_valid, error_msg = validate_activity_record(record)
	if not is_valid:
	raise SchemaValidationError(error_msg)

	# Ensure file exists with header
	filepath = Path(filepath)
	if not filepath.exists():
	filepath.parent.mkdir(parents=True, exist_ok=True)
	with open(filepath, 'w', newline='') as f:
	writer = csv.DictWriter(f, fieldnames=REQUIRED_FIELDS)
	writer.writeheader()

	# Atomic write using temp file and rename
	temp_fd, temp_path = tempfile.mkstemp(dir=filepath.parent, prefix='.activity_log_', suffix='.tmp')

	try:
	# Acquire exclusive lock for atomic operation
	with open(filepath, 'r') as source:
	fcntl.flock(source.fileno(), fcntl.LOCK_EX)

	try:
	# Copy existing content to temp file
	with os.fdopen(temp_fd, 'w', newline='') as temp_file:
	reader = csv.DictReader(source)
	writer = csv.DictWriter(temp_file, fieldnames=REQUIRED_FIELDS)

	# Write header
	writer.writeheader()

	# Copy existing rows
	for row in reader:
	writer.writerow(row)

	# Append new record
	writer.writerow(record)

	# Atomic rename (replaces original file)
	shutil.move(temp_path, filepath)
	result['success'] = True

	finally:
	fcntl.flock(source.fileno(), fcntl.LOCK_UN)

	except Exception as e:
	# Clean up temp file on error
	if os.path.exists(temp_path):
	os.unlink(temp_path)
	raise AtomicWriteError(f"Atomic write failed: {str(e)}")

	# Log successful operation
	result['commit_hash'] = _generate_commit_hash(record, request_id)
	_log_operation(request_id, 'ingest', 'success', record)

	except (SchemaValidationError, AtomicWriteError, Exception) as e:
	result['error'] = str(e)
	result['success'] = False
	_log_operation(request_id, 'ingest', 'error', record, error=str(e))

	return result


	def _generate_commit_hash(record: Dict[str, str], request_id: str) -> str:
	"""Generate pseudo-commit hash for tracking"""
	import hashlib
	data = f"{request_id}:{record['timestamp']}:{record['agent_id']}:{record['activity_type']}"
	return hashlib.sha256(data.encode()).hexdigest()[:16]


	def _log_operation(request_id: str, operation: str, status: str, record: Dict, error: Optional[str] = None):
	"""Log operation to error/audit log file"""
	log_file = Path('activity_ingestion.log')
	timestamp = generate_timestamp()

	log_entry = {
	'timestamp': timestamp,
	'request_id': request_id,
	'operation': operation,
	'status': status,
	'agent_id': record.get('agent_id', 'unknown'),
	'activity_type': record.get('activity_type', 'unknown'),
	'error': error or ''
	}

	with open(log_file, 'a') as f:
	f.write(f"{log_entry}\n")


	def test_ingestion():
	"""Test cases for activity ingestion validation"""
	print("=== Activity Ingestion Function Tests ===")
	print()

	# Test 1: Valid record insertion
	print("Test 1: Valid record insertion")
	valid_record = {
	'timestamp': generate_timestamp(),
	'agent_id': 'test_agent',
	'activity_type': 'ingest',
	'entity_id': 'test_entity_001',
	'status': 'success',
	'notes': 'Test record for validation',
	'context': 'Unit test - valid case'
	}
	result = atomic_append_activity('activity_log.csv', valid_record)
	print(f" Result: {'PASS' if result['success'] else 'FAIL'}")
	print(f" Request ID: {result['request_id']}")
	print(f" Timestamp: {result['timestamp']}")
	if result['success']:
	print(f" Commit Hash: {result['commit_hash']}")
	else:
	print(f" Error: {result['error']}")
	print()

	# Test 2: Missing required field
	print("Test 2: Missing required field (should fail)")
	invalid_record = {
	'timestamp': generate_timestamp(),
	'agent_id': 'test_agent',
	'activity_type': 'ingest',
	# Missing 'entity_id'
	'status': 'success',
	'notes': 'This should fail',
	'context': 'Unit test - invalid case'
	}
	result = atomic_append_activity('activity_log.csv', invalid_record)
	print(f" Result: {'PASS (correctly rejected)' if not result['success'] else 'FAIL (should have rejected)'}")
	print(f" Error: {result['error']}")
	print()

	# Test 3: Invalid status value
	print("Test 3: Invalid status value (should fail)")
	invalid_status = {
	'timestamp': generate_timestamp(),
	'agent_id': 'test_agent',
	'activity_type': 'ingest',
	'entity_id': 'test_entity_003',
	'status': 'invalid_status', # Invalid
	'notes': 'This should fail',
	'context': 'Unit test - invalid status'
	}
	result = atomic_append_activity('activity_log.csv', invalid_status)
	print(f" Result: {'PASS (correctly rejected)' if not result['success'] else 'FAIL (should have rejected)'}")
	print(f" Error: {result['error']}")
	print()

	# Test 4: Auto-generated timestamp
	print("Test 4: Auto-generated timestamp")
	auto_timestamp = {
	'agent_id': 'test_agent',
	'activity_type': 'validate',
	'entity_id': 'test_entity_004',
	'status': 'success',
	'notes': 'Timestamp should be auto-generated',
	'context': 'Unit test - auto timestamp'
	}
	result = atomic_append_activity('activity_log.csv', auto_timestamp)
	print(f" Result: {'PASS' if result['success'] else 'FAIL'}")
	print(f" Generated Timestamp: {result['record']['timestamp']}")
	print()

	# Test 5: Multiple concurrent inserts (simulate race condition safety)
	print("Test 5: Concurrent insert safety")
	for i in range(3):
	record = {
	'timestamp': generate_timestamp(),
	'agent_id': f'concurrent_agent_{i}',
	'activity_type': 'ingest',
	'entity_id': f'concurrent_entity_{i}',
	'status': 'success',
	'notes': f'Concurrent test insert {i}',
	'context': 'Unit test - concurrent safety'
	}
	result = atomic_append_activity('activity_log.csv', record)
	print(f" Insert {i}: {'PASS' if result['success'] else 'FAIL'}")
	print()

	print("=== Test Summary ===")
	print("All tests completed. Check activity_log.csv and activity_ingestion.log for details.")
	print()


	if __name__ == '__main__':
	# Run tests when executed directly
	test_ingestion()