Skip to content

core

Core print logic.

Classes⚓︎

Record ⚓︎

Bases: BaseModel

Record Model.

Source code in tail_jsonl/_private/core.py 
class Record(BaseModel):
    """Record Model."""

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

    @classmethod
    @beartype
    def from_line(cls, data: Dict, config: Config) -> 'Record':  # type: ignore[type-arg]
        """Extract 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)

Extract Record from jsonl.

Source code in tail_jsonl/_private/core.py 
@classmethod
@beartype
def from_line(cls, data: Dict, config: Config) -> 'Record':  # type: ignore[type-arg]
    """Extract 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)

Safely find the first key in the data or default to the fallback.

Source code in tail_jsonl/_private/core.py 
@beartype
def pop_key(data: Dict, keys: List[str], fallback: str) -> Any:  # type: ignore[type-arg]
    """Safely find the first key in the data or default to the fallback."""
    return _pop_key(data, copy(keys), fallback)

print_record ⚓︎

print_record(line, console, config)

Format and print the record.

Source code in tail_jsonl/_private/core.py 
@beartype
def print_record(line: str, console: Console, config: Config) -> None:
    """Format and print the record."""
    try:
        record = Record.from_line(json.loads(line), config=config)
    except Exception:
        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

    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]
        # Ensure that there is no repeat keyword arguments
        **{f'_{key}' if key in keys else key: value for key, value in record.data.items()},
    )