feat: better logging

Author: 01Joseph-Hwang10

jtd_codebuild/cli.py

@@ -1,17 +1,25 @@
+from typing import Callable
 import click
 from click.core import Context
 from .commands import BuildCommand, InitCommand
+from .logger import Logger, ClickLogger
 
 
 @click.group(invoke_without_command=True)
 @click.argument("path", type=click.Path(), required=False)
+@click.option("--verbose", "-v", is_flag=True)
 @click.pass_context
-def cli(ctx: Context, path: str):
+def cli(ctx: Context, path: str, verbose: bool):
     # If no subcommand is provided, run the build command
     if ctx.invoked_subcommand is None:
-        BuildCommand().run(path or ".")
+        logger = _create_logger(verbose)
+        command = BuildCommand(logger=logger)
+        _error_boundary(lambda: command.run(path or "."), logger)
     else:
-        ctx.obj = {"path": path or "."}
+        ctx.obj = {
+            "path": path or ".",
+            "verbose": verbose,
+        }
 
 
 @cli.command("init")
@@ -24,4 +32,21 @@ def cli(ctx: Context, path: str):
 )
 def init(ctx: Context, preset: str):
     path = ctx.obj["path"]
-    InitCommand().run(path, preset)
+    verbose = ctx.obj["verbose"]
+
+    logger = _create_logger(verbose)
+    command = InitCommand(logger=logger)
+    _error_boundary(lambda: command.run(path, preset), logger)
+
+
+def _create_logger(verbose: bool) -> ClickLogger:
+    return ClickLogger(level=1 if verbose else 2)
+
+
+def _error_boundary(func: Callable, logger: Logger):
+    try:
+        func()
+    except Exception as e:
+        for line in str(e).split("\n"):
+            logger.error(line)
+        exit(1)

jtd_codebuild/codebuild.py

@@ -1,5 +1,5 @@
 from os import getcwd
-from os.path import basename
+from os.path import relpath
 from typing import Callable, Type
 from toolz import pipe
 from toolz.curried import map
@@ -45,7 +45,7 @@ def run(  # noqa: C901
         target_path = resolve(cwd, path)
         config = get_project_config(target_path)
 
-        self.logger.info(f"Start building: {basename(target_path)}")
+        self.logger.info(f"Start building: {path}")
 
         self.logger.info("Bundling IDL files...")
         bundler = Bundler(logger=self.logger)
@@ -55,11 +55,10 @@ def run(  # noqa: C901
             inheritance.resolve,
         )
 
-        self.logger.info("Writing bundled IDL file...")
         schema_path = resolve(target_path, config.jtdBundlePath)
         write_json(schema_path, bundled_jtd_schema)
 
-        self.logger.success("Wrote bundled IDL file...")
+        self.logger.success(f"Wrote bundled IDL file at: {relpath(schema_path, cwd)}")
 
         self.logger.info("Generating targets...")
 

jtd_codebuild/generators/_generator.py

@@ -4,6 +4,7 @@
 from jtd_codebuild.logger import Logger
 from jtd_codebuild.config.project.model import Target
 from jtd_codebuild.utils.fs import safe_mkdir, resolve
+from jtd_codebuild.utils.subprocess import stream_logs
 
 
 class JTDCodeGenerator(Component, metaclass=abc.ABCMeta):
@@ -60,6 +61,7 @@ def generate(self, target: Target) -> None:
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
         )
+        stream_logs(process, logger=self.logger, level="debug")
         process.wait()
 
     def get_schema_path(self) -> str:

jtd_codebuild/logger/_click_logger.py

@@ -4,25 +4,35 @@
 
 class ClickLogger(Logger):
     def debug(self, message: str, *args, **kwargs) -> None:
-        kwargs.setdefault("color", "blue")
-        message = f"[jtd-codebuild] | DEBUG | {message}"
-        click.echo(message, *args, **kwargs)
+        if self.level > 1:
+            return
+        kwargs.setdefault("fg", "blue")
+        message = f"[jtd-codebuild] |   DEBUG | {message}"
+        click.echo(click.style(message, *args, **kwargs))
 
     def info(self, message: str, *args, **kwargs) -> None:
-        message = f"[jtd-codebuild] | INFO | {message}"
-        click.echo(message, *args, **kwargs)
+        if self.level > 2:
+            return
+        message = f"[jtd-codebuild] |    INFO | {message}"
+        click.echo(click.style(message, *args, **kwargs))
 
     def success(self, message: str, *args, **kwargs) -> None:
+        if self.level > 3:
+            return
         message = f"[jtd-codebuild] | SUCCESS | {message}"
-        kwargs.setdefault("color", "green")
-        click.echo(message, *args, **kwargs)
+        kwargs.setdefault("fg", "green")
+        click.echo(click.style(message, *args, **kwargs))
 
     def warning(self, message: str, *args, **kwargs) -> None:
+        if self.level > 4:
+            return
         message = f"[jtd-codebuild] | WARNING | {message}"
-        kwargs.setdefault("color", "yellow")
-        click.echo(message, *args, **kwargs)
+        kwargs.setdefault("fg", "yellow")
+        click.echo(click.style(message, *args, **kwargs))
 
     def error(self, message: str, *args, **kwargs) -> None:
-        message = f"[jtd-codebuild] | ERROR | {message}"
-        kwargs.setdefault("color", "red")
-        click.echo(message, *args, **kwargs)
+        if self.level > 5:
+            return
+        message = f"[jtd-codebuild] |   ERROR | {message}"
+        kwargs.setdefault("fg", "red")
+        click.echo(click.style(message, *args, **kwargs))

jtd_codebuild/logger/_logger.py

@@ -1,7 +1,21 @@
 import abc
+from typing import Literal, TypeAlias
+from jtd_codebuild.utils.mapping import Subscriptable
 
+Level: TypeAlias = Literal[
+    1,  # DEBUG
+    2,  # INFO
+    3,  # SUCCESS
+    4,  # WARNING
+    5,  # ERROR
+]
+
+
+class Logger(Subscriptable, metaclass=abc.ABCMeta):
+
+    def __init__(self, level: Level = 2) -> None:
+        self.level = level
 
-class Logger(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def debug(self, message: str, *args, **kwargs) -> None:
         pass

jtd_codebuild/utils/subprocess.py

@@ -0,0 +1,26 @@
+import subprocess
+from jtd_codebuild.logger import Logger
+
+
+def flush_streams_to_logs(
+    process: subprocess.Popen,
+    *,
+    logger: Logger,
+    level: str = "info",
+):
+    pipe_data = process.communicate()
+    for data in pipe_data:
+        if data:
+            for line in data.decode().split("\n"):
+                logger[level](line)
+
+
+def stream_logs(
+    process: subprocess.Popen,
+    *,
+    logger: Logger,
+    level: str = "info",
+):
+    while process.returncode is None:
+        flush_streams_to_logs(process, logger=logger, level=level)
+    flush_streams_to_logs(process, logger=logger, level=level)

poetry.lock

@@ -35,6 +35,7 @@ toolz = "^0.12.1"
 deepmerge = "^1.1.1"
 joblib = "^1.4.0"
 pydantic = "^2.7.1"
+colorama = "^0.4.6"
 
 [tool.poetry.group.dev.dependencies]
 flake8 = "^6.1.0"