from typing import (
Any,
TextIO,
ClassVar,
)
from collections import Counter
from collections.abc import Sequence
from .util import dump
__all__ = (
'Printer',
)
[docs]
class Printer:
"""
A printer for result tables.
Rows are buffered internally and printed together when `flush` is called.
Attributes:
stream: the output stream
max_width: the maximum width of a column
columns: the list of columns
rows: the rows currently in the buffer
Raises:
ValueError: if the maximum width is too small
"""
MIN_WIDTH: ClassVar[int] = 3
"""The minimum width of a column."""
stream: TextIO
max_width: int
columns: list[tuple[str, int]]
rows: list[tuple[Any, ...]]
def __init__(self,
stream: TextIO,
attrs: Sequence[tuple[str | None, str]],
max_width: int = 16,
) -> None:
if max_width <= self.MIN_WIDTH:
raise ValueError(f"max_width must be at least {self.MIN_WIDTH} ({dump(
max_width=max_width,
)}).")
columns = []
counter = Counter([name for _, name in attrs])
for ref, name in attrs:
if counter[name] > 1:
name = f"{ref}.{name}"
else:
name = name
width = min(len(name), max_width)
columns.append((name, width))
self.stream = stream
self.columns = columns
self.max_width = max_width
self.rows = []
def _ellipsis(self, value: str) -> str:
if len(value) <= self.max_width:
return value
return value[:self.max_width - self.MIN_WIDTH] + "." * self.MIN_WIDTH
def _bar(self) -> None:
self.stream.write("+")
for _, width in self.columns:
self.stream.write("-" * width)
self.stream.write("+")
self.stream.write("\n")
[docs]
def append(self, row: tuple[Any, ...]) -> None:
"""
Append a row to the buffer.
Raises:
ValueError: if the number of values in the row does not match the number of attributes
"""
if len(row) != len(self.columns):
raise ValueError(
f"The number of values in the row does not match the number of attributes ({dump(
values=len(row),
attrs=len(self.columns),
)})."
)
values = []
for cid, value in enumerate(row):
name, width = self.columns[cid]
value = str(value)
width = max(width, min(len(value), self.max_width))
self.columns[cid] = (name, width)
values.append(value)
self.rows.append(tuple(values))
[docs]
def flush(self) -> None:
"""
Print all buffered rows and clear the buffer.
"""
self._bar()
self.stream.write("|")
for name, width in self.columns:
self.stream.write(f"{self._ellipsis(name):<{width}}")
self.stream.write("|")
self.stream.write("\n")
self._bar()
for row in self.rows:
self.stream.write("|")
for cid, value in enumerate(row):
_, width = self.columns[cid]
self.stream.write(f"{self._ellipsis(value):<{width}}")
self.stream.write("|")
self.stream.write("\n")
self._bar()
count = len(self.rows)
self.stream.write("".join([
"(total ",
str(count),
" row",
"s" if count != 1 else "",
")\n",
]))
self.rows = []