Module piate.cli.render
Expand source code
import dataclasses
import json
from functools import reduce
from json import JSONEncoder
from typing import Any, List, Iterator
import click
from piate.cli.format import Format
class _IATEResponseEncoder(JSONEncoder):
def default(self, o):
if dataclasses.is_dataclass(o):
if hasattr(o, "items"):
return [i.compact() for i in o.items]
elif hasattr(o, "compact"):
return o.compact()
else:
return dataclasses.asdict(o)
else:
return super().default(o)
def response(obj: Any):
click.echo(json.dumps(obj, cls=_IATEResponseEncoder, indent=4))
def response_lines(obj: List[Any]):
for item in obj:
click.echo(json.dumps(item, cls=_IATEResponseEncoder))
def pages_response_iterator(paginator: Iterator, format: Format):
if format == Format.JSON:
response(list(reduce(lambda items, page: items + page.items, paginator, [])))
elif format == Format.JSON_LINES:
for page in paginator:
response_lines(page.items)
else:
raise RuntimeError(f"Unexpected output format: {format=}")
def page_response(page, format: Format):
if format == Format.JSON:
response(page)
elif format == Format.JSON_LINES:
response_lines(page.items)
else:
raise RuntimeError(f"Unexpected output format: {format=}")
Functions
def page_response(page, format: Format)-
Expand source code
def page_response(page, format: Format): if format == Format.JSON: response(page) elif format == Format.JSON_LINES: response_lines(page.items) else: raise RuntimeError(f"Unexpected output format: {format=}") def pages_response_iterator(paginator: Iterator, format: Format)-
Expand source code
def pages_response_iterator(paginator: Iterator, format: Format): if format == Format.JSON: response(list(reduce(lambda items, page: items + page.items, paginator, []))) elif format == Format.JSON_LINES: for page in paginator: response_lines(page.items) else: raise RuntimeError(f"Unexpected output format: {format=}") def response(obj: Any)-
Expand source code
def response(obj: Any): click.echo(json.dumps(obj, cls=_IATEResponseEncoder, indent=4)) def response_lines(obj: List[Any])-
Expand source code
def response_lines(obj: List[Any]): for item in obj: click.echo(json.dumps(item, cls=_IATEResponseEncoder))