Core

Core print logic.

Classes

Record dataclass

Record Model.

Source code in tail_jsonl/_private/core.py

@dataclass
class Record:
    """Record Model."""

    timestamp: str
    level: str
    message: str
    data: dict  # type: ignore[type-arg]

    @classmethod
    def from_line(cls, data: dict, config: Config) -> Record:  # type: ignore[type-arg]
        """Return Record from jsonl."""
        return cls(
            timestamp=pop_key(data, config.keys.timestamp, '<no timestamp>'),
            level=pop_key(data, config.keys.level, ''),
            message=pop_key(data, config.keys.message, '<no message>'),
            data=data,
        )

Functions

from_line classmethod

from_line(data, config)

Return Record from jsonl.

Source code in tail_jsonl/_private/core.py

@classmethod
def from_line(cls, data: dict, config: Config) -> Record:  # type: ignore[type-arg]
    """Return Record from jsonl."""
    return cls(
        timestamp=pop_key(data, config.keys.timestamp, '<no timestamp>'),
        level=pop_key(data, config.keys.level, ''),
        message=pop_key(data, config.keys.message, '<no message>'),
        data=data,
    )

Functions

pop_key

pop_key(data, keys, fallback)

Return the first key in the data or default to the fallback.

Source code in tail_jsonl/_private/core.py

def pop_key(data: dict, keys: list[str], fallback: str) -> Any:  # type: ignore[type-arg]
    """Return the first key in the data or default to the fallback."""
    return _pop_key(data, keys, 0, fallback)

print_record

print_record(line, console, config)

Format and print the record.

Source code in tail_jsonl/_private/core.py

def print_record(line: str, console: Console, config: Config) -> None:
    """Format and print the record."""
    try:
        data = json.loads(line)
        record = Record.from_line(data, config=config)
        if config.debug:
            console.print(
                (
                    f'[dim]DEBUG: Parsed keys - timestamp={record.timestamp!r},'
                    f' level={record.level!r}, message={record.message!r}[/dim]'
                ),
                markup=True,
                highlight=False,
            )
    except (json.JSONDecodeError, ValueError, KeyError, TypeError, AttributeError) as exc:
        if config.debug:
            console.print(
                f'[dim red]DEBUG: Failed to parse line as JSON: {exc.__class__.__name__}: {exc}[/dim red]',
                markup=True,
                highlight=False,
            )
        console.print(line.rstrip(), markup=False, highlight=False)  # Print the unmodified line
        return

    if (_this_level := get_level(name=record.level)) == logging.NOTSET and record.level:
        record.data['_level_name'] = record.level

    # PLANNED: Consider moving to Corallium
    for dotted_key in config.keys.on_own_line:
        if '.' not in dotted_key:
            continue
        if value := dotted.get(record.data, dotted_key):
            if config.debug:
                console.print(
                    f'[dim]DEBUG: Promoting dotted key {dotted_key!r} to own line[/dim]',
                    markup=True,
                    highlight=False,
                )
            record.data[dotted_key] = value if isinstance(value, str) else str(value)
            dotted.remove(record.data, dotted_key)

    printer_kwargs = {
        'message': record.message,
        'is_header': False,
        '_this_level': _this_level,
        '_is_text': False,
        '_console': console,
        '_styles': config.styles,
        '_keys_on_own_line': config.keys.on_own_line,
        'timestamp': record.timestamp,
    }
    keys = set(printer_kwargs)
    rich_printer(
        **printer_kwargs,  # type: ignore[arg-type]
        # Try to print all values and avoid name collision
        **{f' {key}' if key in keys else key: value for key, value in record.data.items()},
    )