2
0
mirror of https://github.com/frappe/bench.git synced 2024-11-13 16:56:33 +00:00

refactor: CLI Rendering Engine

* Hanlde job-step renders
* Use context manager for managing renders and state
* Job and Step remain process agnostic - neither are dependant on the
  other
This commit is contained in:
Gavin D'souza 2021-11-18 18:10:01 +05:30
parent 6d10d75bb1
commit f230778972

View File

@ -5,6 +5,9 @@ from io import StringIO
# imports - third party imports
import click
# imports - module imports
import bench
class Capturing(list):
"""
@ -28,38 +31,81 @@ class Capturing(list):
sys.stdout = self._stdout
class Rendering:
def __init__(self, success, title, is_parent, args, kwargs):
self.dynamic_feed = bench.cli.from_command_line and bench.cli.dynamic_feed
if not self.dynamic_feed:
return
try:
self.kw = args[0].__dict__
except Exception:
self.kw = kwargs
self.is_parent = is_parent
self.title = title
self.success = success
def __enter__(self, *args, **kwargs):
if not self.dynamic_feed:
return
_prefix = click.style('', fg='bright_yellow')
_hierarchy = " " if not self.is_parent else ""
self._title = self.title.format(**self.kw)
click.secho(f"{_hierarchy}{_prefix} {self._title}")
bench.LOG_BUFFER.append(
{"message": self._title, "prefix": _prefix, "color": None, "is_parent": self.is_parent}
)
def __exit__(self, *args, **kwargs):
if not self.dynamic_feed:
return
self._prefix = click.style('', fg='green')
self._success = self.success.format(**self.kw)
self.render_screen()
def render_screen(self):
click.clear()
for l in bench.LOG_BUFFER:
if l["message"] == self._title:
l["prefix"] = self._prefix
l["message"] = self._success
_hierarchy = " " if not l["is_parent"] else ""
click.secho(f'{_hierarchy}{l["prefix"]} {l["message"]}', fg=l["color"])
def job(title: str = None, success: str = None):
"""Supposed to be wrapped around an atomic job in a given process.
For instance, the `get-app` command consists of two jobs: `initializing bench`
and `fetching and installing app`.
"""
def innfn(fn):
def wrapper_fn(*args, **kwargs):
with Rendering(
success=success, title=title, is_parent=True, args=args, kwargs=kwargs,
):
return fn(*args, **kwargs)
return wrapper_fn
return innfn
def step(title: str = None, success: str = None):
"""Supposed to be wrapped around the smallest possible atomic step in a given operation.
For instance, `building assets` is a step in the update operation.
"""
def innfn(fn):
def wrapper_fn(*args, **kwargs):
import bench.cli
if bench.cli.from_command_line and bench.cli.dynamic_feed:
kw = args[0].__dict__
_title = f"{click.style('', fg='bright_yellow')} {title.format(**kw)}"
click.secho(_title)
retval = fn(*args)
if bench.cli.from_command_line and bench.cli.dynamic_feed:
click.clear()
for l in bench.LOG_BUFFER:
click.secho(l["message"], fg=l["color"])
_success = f"{click.style('', fg='green')} {success.format(**kw)}"
click.echo(_success)
bench.LOG_BUFFER.append(
{"message": _success, "color": None,}
)
return retval
with Rendering(
success=success, title=title, is_parent=False, args=args, kwargs=kwargs,
):
return fn(*args, **kwargs)
return wrapper_fn
return innfn