diff --git a/bench/utils/render.py b/bench/utils/render.py index 11a6e0f6..9c862b4b 100644 --- a/bench/utils/render.py +++ b/bench/utils/render.py @@ -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