feat: make subprocess executor robust against invalid encoding and ca…

README.md

@@ -38,24 +38,47 @@
 
 My application development modules.
 
-## Installation
+## Start developing
 
-Install this via pip (or your favourite package manager):
+The project uses UV for dependencies management and packaging and the [pypeline](https://github.com/cuinixam/pypeline) for streamlining the development workflow.
+Use pipx (or your favorite package manager) to install the `pypeline` in an isolated environment:
 
-`pip install py-app-dev`
-
-## Usage
+```shell
+pipx install pypeline-runner
+```
 
-Start by importing it:
+To bootstrap the project and run all the steps configured in the `pypeline.yaml` file, execute the following command:
 
-```python
-import py_app_dev
+```shell
+pypeline run
 ```
 
+For those using [VS Code](https://code.visualstudio.com/) there are tasks defined for the most common commands:
 
-## Credits
+- run tests
+- run pre-commit checks (linters, formatters, etc.)
+- generate documentation
+
+See the `.vscode/tasks.json` for more details.
+
+## Committing changes
+
+This repository uses [commitlint](https://github.com/conventional-changelog/commitlint) for checking if the commit message meets the [conventional commit format](https://www.conventionalcommits.org/en).
 
-[![Copier](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/copier-org/copier/master/img/badge/badge-grayscale-inverted-border-orange.json)](https://github.com/copier-org/copier)
+## Contributors ✨
+
+Thanks goes to these wonderful people ([emoji key](https://allcontributors.org/docs/en/emoji-key)):
+
+<!-- prettier-ignore-start -->
+<!-- ALL-CONTRIBUTORS-LIST:START - Do not remove or modify this section -->
+<!-- markdownlint-disable -->
+<!-- markdownlint-enable -->
+<!-- ALL-CONTRIBUTORS-LIST:END -->
+<!-- prettier-ignore-end -->
+
+This project follows the [all-contributors](https://github.com/all-contributors/all-contributors) specification. Contributions of any kind welcome!
+
+## Credits
 
 This package was created with
 [Copier](https://copier.readthedocs.io/) and the

docs/conf.py

@@ -66,9 +66,7 @@
 traceability_collapse_links = True
 
 # The suffix of source filenames.
-source_suffix = [
-    ".md",
-]
+source_suffix = [".md", ".rst"]
 
 templates_path = ["_templates"]
 exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]

src/py_app_dev/core/subprocess.py

@@ -1,88 +1,99 @@
-import shutil
-import subprocess  # nosec
-from pathlib import Path
-from typing import Any
-
-from .exceptions import UserNotificationException
-from .logging import logger
-
-
-def which(app_name: str) -> Path | None:
-    """Return the path to the app if it is in the PATH, otherwise return None."""
-    app_path = shutil.which(app_name)
-    return Path(app_path) if app_path else None
-
-
-class SubprocessExecutor:
-    """
-    Execute a command in a subprocess.
-
-    Args:
-    ----
-        capture_output: If True, the output of the command will be captured.
-        print_output: If True, the output of the command will be printed to the logger.
-                      One can set this to false in order to get the output in the returned CompletedProcess object.
-
-    """
-
-    def __init__(
-        self,
-        command: str | list[str | Path],
-        cwd: Path | None = None,
-        capture_output: bool = True,
-        env: dict[str, str] | None = None,
-        shell: bool = False,
-        print_output: bool = True,
-    ):
-        self.logger = logger.bind()
-        self.command = command
-        self.current_working_directory = cwd
-        self.capture_output = capture_output
-        self.env = env
-        self.shell = shell
-        self.print_output = print_output
-
-    @property
-    def command_str(self) -> str:
-        if isinstance(self.command, str):
-            return self.command
-        return " ".join(str(arg) if not isinstance(arg, str) else arg for arg in self.command)
-
-    def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any] | None:
-        """Execute the command and return the CompletedProcess object if handle_errors is False."""
-        try:
-            completed_process = None
-            stdout = ""
-            stderr = ""
-            self.logger.info(f"Running command: {self.command_str}")
-            cwd_path = (self.current_working_directory or Path.cwd()).as_posix()
-            with subprocess.Popen(
-                args=self.command,
-                cwd=cwd_path,
-                stdout=(subprocess.PIPE if self.capture_output else subprocess.DEVNULL),
-                stderr=(subprocess.STDOUT if self.capture_output else subprocess.DEVNULL),
-                text=True,
-                env=self.env,
-                shell=self.shell,
-            ) as process:  # nosec
-                if self.capture_output and process.stdout is not None:
-                    if self.print_output:
-                        for line in iter(process.stdout.readline, ""):
-                            self.logger.info(line.strip())
-                        process.wait()
-                    else:
-                        stdout, stderr = process.communicate()
-
-            if handle_errors:
-                # Check return code
-                if process.returncode != 0:
-                    raise subprocess.CalledProcessError(process.returncode, self.command_str)
-            else:
-                completed_process = subprocess.CompletedProcess(process.args, process.returncode, stdout, stderr)
-        except subprocess.CalledProcessError as e:
-            raise UserNotificationException(f"Command '{self.command_str}' execution failed with return code {e.returncode}") from None
-        except FileNotFoundError as e:
-            raise UserNotificationException(f"Command '{self.command_str}' could not be executed. Failed with error {e}") from None
-        except KeyboardInterrupt:
-            raise UserNotificationException(f"Command '{self.command_str}' execution interrupted by user") from None
-        return completed_process
+import locale
+import shutil
+import subprocess  # nosec
+from pathlib import Path
+from typing import Any
+
+from .exceptions import UserNotificationException
+from .logging import logger
+
+
+def which(app_name: str) -> Path | None:
+    """Return the path to the app if it is in the PATH, otherwise return None."""
+    app_path = shutil.which(app_name)
+    return Path(app_path) if app_path else None
+
+
+class SubprocessExecutor:
+    """
+    Execute a command in a subprocess.
+
+    Args:
+    ----
+        capture_output: If True, the output of the command will be captured.
+        print_output: If True, the output of the command will be printed to the logger.
+                      One can set this to false in order to get the output in the returned CompletedProcess object.
+
+    """
+
+    def __init__(
+        self,
+        command: str | list[str | Path],
+        cwd: Path | None = None,
+        capture_output: bool = True,
+        env: dict[str, str] | None = None,
+        shell: bool = False,
+        print_output: bool = True,
+    ):
+        self.logger = logger.bind()
+        self.command = command
+        self.current_working_directory = cwd
+        self.capture_output = capture_output
+        self.env = env
+        self.shell = shell
+        self.print_output = print_output
+
+    @property
+    def command_str(self) -> str:
+        if isinstance(self.command, str):
+            return self.command
+        return " ".join(str(arg) if not isinstance(arg, str) else arg for arg in self.command)
+
+    def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any] | None:
+        """Execute the command and return the CompletedProcess object if handle_errors is False."""
+        try:
+            completed_process = None
+            stdout = ""
+            stderr = ""
+            self.logger.info(f"Running command: {self.command_str}")
+            cwd_path = (self.current_working_directory or Path.cwd()).as_posix()
+            with subprocess.Popen(
+                args=self.command,
+                cwd=cwd_path,
+                # Combine both streams to stdout (when captured)
+                stdout=(subprocess.PIPE if self.capture_output else subprocess.DEVNULL),
+                stderr=(subprocess.STDOUT if self.capture_output else subprocess.DEVNULL),
+                # enables line buffering, line is flushed after each \n
+                bufsize=1,
+                text=True,
+                # every new line is a \n
+                universal_newlines=True,
+                # decode bytes to str using current locale/system encoding
+                encoding=locale.getpreferredencoding(False),
+                # replace unknown characters with �
+                errors="replace",
+                env=self.env,
+                shell=self.shell,
+            ) as process:  # nosec
+                if self.capture_output and process.stdout is not None:
+                    if self.print_output:
+                        for line in iter(process.stdout.readline, ""):
+                            self.logger.info(line.strip())
+                            stdout += line
+                        process.wait()
+                    else:
+                        stdout, stderr = process.communicate()
+
+            if handle_errors:
+                # Check return code
+                if process.returncode != 0:
+                    raise subprocess.CalledProcessError(process.returncode, self.command_str)
+            else:
+                completed_process = subprocess.CompletedProcess(process.args, process.returncode, stdout, stderr)
+        except subprocess.CalledProcessError as e:
+            raise UserNotificationException(f"Command '{self.command_str}' execution failed with return code {e.returncode}") from None
+        except FileNotFoundError as e:
+            raise UserNotificationException(f"Command '{self.command_str}' could not be executed. Failed with error {e}") from None
+        except KeyboardInterrupt:
+            raise UserNotificationException(f"Command '{self.command_str}' execution interrupted by user") from None
+        return completed_process