tessl/pypi-eliot-tree

Render Eliot logs as an ASCII tree

—

Pending

Overview

Eval results

Files

Error Handling

Name: tessl/pypi-eliot-tree
Author: tessl

Exception classes for handling parsing errors that can occur when processing Eliot message dictionaries and JSON text. These exceptions provide detailed error context and debugging information to help identify and resolve issues with malformed or invalid log data.

Capabilities

Eliot Parse Error Handling

Exception raised when parsing fails for Eliot message dictionaries during task processing.

class EliotParseError(RuntimeError):
    """
    An error occurred while parsing a particular Eliot message dictionary.

    Attributes:
    - message_dict: dict - The problematic Eliot message dictionary that caused the error
    - exc_info: tuple - Exception information tuple from sys.exc_info()
    """
    
    def __init__(self, message_dict, exc_info):
        """
        Initialize EliotParseError with context information.

        Parameters:
        - message_dict: dict - The Eliot message dictionary that failed to parse
        - exc_info: tuple - Exception information from sys.exc_info()
        """

Usage Example:

from eliottree import tasks_from_iterable, EliotParseError
import sys

def safe_task_parsing(messages):
    """Parse tasks with error handling."""
    try:
        tasks = list(tasks_from_iterable(messages))
        return tasks
    except EliotParseError as e:
        print(f"Failed to parse Eliot message: {e.message_dict}")
        print(f"Original exception: {e.exc_info[1]}")
        
        # Log full exception details
        import traceback
        traceback.print_exception(*e.exc_info)
        
        return []

# Usage with potentially malformed messages
messages = [
    {"timestamp": 1425356936.278875, "action_status": "started"},  # Missing required fields
    {"task_uuid": "invalid", "malformed": True}  # Invalid structure
]

tasks = safe_task_parsing(messages)

JSON Parse Error Handling

Exception raised when JSON parsing fails during file processing, providing context about the specific location and content of the parsing failure.

class JSONParseError(RuntimeError):
    """
    An error occurred while parsing JSON text from a file or input stream.

    Attributes:
    - file_name: str - Name of the file being parsed when error occurred
    - line_number: int - Line number in the file where parsing failed
    - line: str - The actual line content that caused the parsing error
    - exc_info: tuple - Exception information tuple from sys.exc_info()
    """
    
    def __init__(self, file_name, line_number, line, exc_info):
        """
        Initialize JSONParseError with detailed context.

        Parameters:
        - file_name: str - Name of the file being parsed
        - line_number: int - Line number where error occurred
        - line: str - Content of the problematic line
        - exc_info: tuple - Exception information from sys.exc_info()
        """

Usage Example:

import json
from eliottree import JSONParseError

def parse_eliot_log_file(file_path):
    """Parse Eliot log file with comprehensive error handling."""
    messages = []
    
    try:
        with open(file_path, 'r') as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                if not line:
                    continue
                    
                try:
                    message = json.loads(line)
                    messages.append(message)
                except json.JSONDecodeError as e:
                    # Wrap in JSONParseError with context
                    import sys
                    raise JSONParseError(
                        file_name=file_path,
                        line_number=line_num,
                        line=line,
                        exc_info=sys.exc_info()
                    )
                    
    except JSONParseError as e:
        print(f"JSON parsing failed in {e.file_name} at line {e.line_number}")
        print(f"Problematic line: {e.line}")
        print(f"JSON error: {e.exc_info[1]}")
        return []
    
    except IOError as e:
        print(f"File error: {e}")
        return []
        
    return messages

# Usage
messages = parse_eliot_log_file('eliot.log')

Complete Error Handling Pipeline

import json
import sys
from eliottree import (
    tasks_from_iterable, render_tasks, get_theme,
    EliotParseError, JSONParseError
)

def robust_eliot_processing(file_path):
    """
    Complete error-handling pipeline for Eliot log processing.
    """
    
    def parse_json_file(file_path):
        """Parse JSON file with line-by-line error handling."""
        messages = []
        errors = []
        
        try:
            with open(file_path, 'r') as f:
                for line_num, line in enumerate(f, 1):
                    line = line.strip()
                    if not line:
                        continue
                        
                    try:
                        message = json.loads(line)
                        messages.append(message)
                    except json.JSONDecodeError as json_err:
                        error = {
                            'type': 'JSON',
                            'line_number': line_num,
                            'line': line,
                            'error': str(json_err)
                        }
                        errors.append(error)
                        continue
                        
        except IOError as io_err:
            print(f"Cannot read file {file_path}: {io_err}")
            return [], []
            
        return messages, errors
    
    def parse_eliot_tasks(messages):
        """Parse Eliot tasks with error collection."""
        tasks = []
        errors = []
        
        try:
            for task in tasks_from_iterable(messages):
                tasks.append(task)
        except EliotParseError as e:
            error = {
                'type': 'Eliot',
                'message_dict': e.message_dict,
                'error': str(e.exc_info[1])
            }
            errors.append(error)
            
        return tasks, errors
    
    # Main processing
    print(f"Processing {file_path}...")
    
    # Parse JSON
    messages, json_errors = parse_json_file(file_path)
    
    if json_errors:
        print(f"JSON parsing errors: {len(json_errors)}")
        for error in json_errors[:5]:  # Show first 5 errors
            print(f"  Line {error['line_number']}: {error['error']}")
    
    if not messages:
        print("No valid messages found")
        return
    
    # Parse Eliot tasks
    tasks, eliot_errors = parse_eliot_tasks(messages)
    
    if eliot_errors:
        print(f"Eliot parsing errors: {len(eliot_errors)}")
        for error in eliot_errors[:3]:  # Show first 3 errors
            print(f"  Message: {error['message_dict']}")
            print(f"  Error: {error['error']}")
    
    if not tasks:
        print("No valid tasks generated")
        return
    
    # Render with error handling
    def error_writer(text):
        print(f"RENDER ERROR: {text}", file=sys.stderr)
    
    theme = get_theme(dark_background=True)
    
    try:
        render_tasks(
            sys.stdout.write,
            tasks,
            theme=theme,
            write_err=error_writer,
            human_readable=True
        )
        print(f"\nSuccessfully processed {len(tasks)} tasks")
        
    except Exception as e:
        print(f"Rendering failed: {e}")

# Usage
robust_eliot_processing('eliot.log')

Error Context and Debugging

EliotParseError Context

try:
    tasks = list(tasks_from_iterable(messages))
except EliotParseError as e:
    print("=== Eliot Parse Error Details ===")
    print(f"Problematic message: {json.dumps(e.message_dict, indent=2)}")
    print(f"Exception type: {e.exc_info[0].__name__}")
    print(f"Exception message: {e.exc_info[1]}")
    
    # Full traceback
    import traceback
    print("Full traceback:")
    traceback.print_exception(*e.exc_info)

JSONParseError Context

try:
    # JSON parsing code
    pass
except JSONParseError as e:
    print("=== JSON Parse Error Details ===")
    print(f"File: {e.file_name}")
    print(f"Line number: {e.line_number}")
    print(f"Problematic content: {repr(e.line)}")
    print(f"JSON error: {e.exc_info[1]}")
    
    # Show surrounding context
    with open(e.file_name, 'r') as f:
        lines = f.readlines()
        start = max(0, e.line_number - 3)
        end = min(len(lines), e.line_number + 2)
        
        print("Context:")
        for i in range(start, end):
            marker = ">>> " if i + 1 == e.line_number else "    "
            print(f"{marker}{i+1:4d}: {lines[i].rstrip()}")

Common Error Scenarios

Malformed Eliot Messages

# Missing required fields
{"timestamp": 1425356936.278875}  # Missing task_uuid, action_type

# Invalid field types  
{"task_uuid": 123, "timestamp": "invalid"}  # Wrong types

# Corrupted task structure
{"task_level": "not_a_list", "action_status": None}

Invalid JSON Content

# Truncated JSON
{"timestamp": 1425356936.278875, "action_type": "incomplete"

# Invalid escape sequences
{"message": "Invalid \x unicode"}

# Mixed content
Valid JSON line
Not JSON content here
{"another": "valid line"}

File System Issues

File not found or permission denied
Corrupted file content or encoding issues
Very large files causing memory problems
Network interruptions during file access

These error classes provide the necessary context to identify, debug, and resolve issues that commonly occur when processing Eliot log data in production environments.

Install with Tessl CLI