Improve error handling

Author: mc-cat-tty

tricore-cli/pyproject.toml

@@ -1,11 +1,11 @@
 [project]
 name = "tricore-cli"
-version = "1.6"
+version = "1.7"
 authors = [{ name = "Francesco Mecatti" }]
 description = "A utility script for interacting with containerized TriCore development environment"
 readme = "README.md"
 requires-python = ">=3.10"
-dependencies = ["docker>=7.0.0"]
+dependencies = ["docker>=7.0.0", "rich>=13.7"]
 
 [project.urls]
 Homepage = "https://github.com/mc-cat-tty/tricore-dev-env"
@@ -15,5 +15,5 @@ Issues = "https://github.com/mc-cat-tty/tricore-dev-env/issues"
 tricorecli = "tricorecli:main"
 
 [build-system]
-requires = ["setuptools>=61.0", "rich>=13.7"]
+requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"

tricore-cli/src/containers.py

@@ -1,9 +1,10 @@
+from __future__ import annotations
 import docker
 import docker.errors
 from utils import *
 from docker.models.containers import Container
 from docker.errors import DockerException
-from docker import DockerClient
+from docker import DockerClient, APIClient
 from docker.types import Mount
 
 IMAGE: str = 'francescomecatti/tricore-dev-env:1.0'
@@ -14,19 +15,28 @@ def __init__(self, image_name: str):
  self._client: DockerClient | None = None
  self._container: Container | None = None
 
- def __enter__(self):
+ def __enter__(self) -> DisposableContainer:
  try:
  self._client = docker.from_env()
  except DockerException:
  print('Docker engine not found. Make sure to have a running Docker engine!')
  exit(ExitCode.ENGINE_NOT_STARTED)
 
- return self._container
+ return self
 
- def __exit__(self, type, value, tb):
+ def __exit__(self, type, value, tb) -> None:
  assert self._container, "Invalid container"
  self._container.stop(timeout=1)
  self._container.remove()
+ 
+ def run_async(self, cmd: str, stream_handler_callback) -> int:
+ if not self._container: raise RuntimeError("Container not initialized")
+
+ client = APIClient()
+ exec_handle = client.exec_create(self._container.id, cmd)
+ stream = client.exec_start(exec_handle, stream=True)
+ stream_handler_callback(stream)
+ return client.exec_inspect(exec_handle['Id']).get('ExitCode')
 
 
 class BuildDisposableContainer(DisposableContainer):
@@ -36,7 +46,7 @@ def __init__(self, folder: str):
  super().__init__(IMAGE)
  self.__folder = folder
 
- def __enter__(self):
+ def __enter__(self) -> DisposableContainer:
  super().__enter__()
  assert self._client, "Invalid client"
 
@@ -56,5 +66,4 @@ def __enter__(self):
  working_dir = BuildDisposableContainer.SRC_DIR
  )
 
- if not self._container: raise RuntimeError("sdfas") 
- return self._container
+ return self

tricore-cli/src/tricorecli.py

@@ -30,17 +30,21 @@ def build(args: BuildHanderArgs) -> None:
  with BuildDisposableContainer(abs_path) as build_container:
  print(f'Building source with {CPUS} jobs from {abs_path}')
 
- res = build_container.exec_run('cmake -B build --toolchain tricore_toolchain.cmake', stream=True)
- print_stream(res)
+ exit_if_error(
+ build_container.run_async('cmake -B build --toolchain tricore_toolchain.cmake', print_stream),
+ "Failed makefiles generation."
+ )
 
- res = build_container.exec_run(f'cmake --build build --parallel {CPUS}', stream=True)
- print_stream(res)
+ exit_if_error(
+ build_container.run_async(f'cmake --build build --parallel {CPUS}', print_stream),
+ "Build failed."
+ )
 
- print(f"Successfully built artifacts in {build_path}")
+ print(f"Successfully built artifacts in {build_path}.")
 
 
 def flash(args: FlashHandlerArgs) -> None:
- raise NotImplementedError('Feature not implemented yet')
+ raise NotImplementedError('Feature not implemented yet.')
 
 
 def main() -> None:

tricore-cli/src/utils.py

@@ -35,10 +35,16 @@ def check_file_or_exit(file_path: str, error_msg: str = "", exit_code: int = 1):
  print(error_msg)
  exit(exit_code)
 
-def print_stream(exec_result):
- if GlobalConfiguration().verbosity_level < 1: return
- 
- for line in exec_result.output:
+def exit_if_error(exit_code: int, msg: str):
+ if exit_code == 0: return
+ print(msg)
+ exit(exit_code)
+
+def print_stream(exec_stream):
+ skip_print: bool = GlobalConfiguration().verbosity_level < 1
+
+ for line in exec_stream:
+ if skip_print: continue
  print(line.decode('utf-8'), end='')
 
 def print_pull_progress(stream: Generator):