Automate JSON

pyproject.toml

@@ -87,6 +87,7 @@ devel = [
   {include-group = "docs"},
   {include-group = "repl"},
   # {include-group = "publish"},
+  "cython>=3.1.6",
 ]
 
 

src/docbuild/cli/cmd_metadata/metaprocess.py

@@ -4,7 +4,6 @@
 import logging
 from pathlib import Path
 import shlex
-import tempfile
 
 from lxml import etree
 from rich.console import Console
@@ -14,6 +13,7 @@
 from ...models.deliverable import Deliverable
 from ...models.doctype import Doctype
 from ...utils.contextmgr import PersistentOnErrorTemporaryDirectory
+from ..commands import run_daps, run_git
 from ..context import DocBuildContext
 
 # Set up rich consoles for output
@@ -91,59 +91,53 @@ async def process_deliverable(
     outputdir.mkdir(parents=True, exist_ok=True)
 
     prefix = (
-        f'{deliverable.productid}-{deliverable.docsetid}-'
-        f'{deliverable.lang}--{deliverable.dcfile}'
+        f'clone-{deliverable.productid}-{deliverable.docsetid}-'
+        f'{deliverable.lang}--{deliverable.dcfile}_'
     )
 
     try:
         async with PersistentOnErrorTemporaryDirectory(
             dir=str(temp_repo_dir),
-            prefix=f'clone-{prefix}_',
+            prefix=prefix,
         ) as worktree_dir:
             # 1. Create a temporary clone from the bare repo and checkout a branch.
-            clone_cmd = [
-                'git',
+            result = await run_git(
                 'clone',
                 '--local',
                 f'--branch={deliverable.branch}',
                 str(bare_repo_path),
                 str(worktree_dir),
-            ]
-            clone_process = await asyncio.create_subprocess_exec(
-                *clone_cmd,
                 stdin=asyncio.subprocess.DEVNULL,
                 stderr=asyncio.subprocess.PIPE,
             )
-            _, stderr = await clone_process.communicate()
-            if clone_process.returncode != 0:
-                # Raise an exception on failure to let the context manager know.
+            if result.returncode != 0:
                 raise RuntimeError(
-                    f'Failed to clone {bare_repo_path}: {stderr.decode().strip()}'
+                    f'Failed to clone {bare_repo_path}: {result.stderr.strip()}'
                 )
 
             # The source file for daps might be in a subdirectory
             dcfile_path = Path(deliverable.subdir) / deliverable.dcfile
 
             # 2. Run the daps command
-            cmd = shlex.split(
+            # ! Keep in mind that the config entry does not need the "daps"
+            # ! command just the arguments.
+            # TODO: Check for both cases, with and without the command?
+            args = shlex.split(
                 dapstmpl.format(
                     builddir=str(worktree_dir),
                     dcfile=str(worktree_dir / dcfile_path),
                     output=str(outputdir / deliverable.dcfile),
                 )
             )
-            console_out.print(f'  command: {cmd}')
-            daps_process = await asyncio.create_subprocess_exec(
-                *cmd,
+            result = await run_daps(
+                *args,
                 stdin=asyncio.subprocess.DEVNULL,
                 stderr=asyncio.subprocess.PIPE,
             )
-            _, stderr = await daps_process.communicate()
-            if daps_process.returncode != 0:
-                # Raise an exception on failure.
+            if result.returncode != 0:
                 raise RuntimeError(
                     f'DAPS command failed for {deliverable.full_id}: '
-                    f'{stderr.decode().strip()}'
+                    f'{result.stderr.strip()}'
                 )
 
         console_out.print(f'> Processed deliverable: {deliverable.pdlangdc}')

src/docbuild/cli/cmd_repo/process.py

@@ -6,9 +6,10 @@
 
 from ...cli.context import DocBuildContext
 from ...config.xml.stitch import create_stitchfile
+from ...constants import GITLOGGER_NAME
 from ...models.repo import Repo
 from ...utils.contextmgr import make_timer
-from ...constants import GITLOGGER_NAME
+from ..commands import run_git
 
 log = logging.getLogger(GITLOGGER_NAME)
 
@@ -29,7 +30,7 @@ async def clone_repo(repo: Repo, base_dir: Path) -> bool:
     # Use create_subprocess_exec for security (avoids shell injection)
     # and pass arguments as a sequence.
     cmd_args = [
-        'git',
+        # 'git',
         '-c',
         'color.ui=never',
         'clone',
@@ -38,11 +39,9 @@ async def clone_repo(repo: Repo, base_dir: Path) -> bool:
         str(repo),
         str(repo_path),
     ]
-
-    process = await asyncio.create_subprocess_exec(
+    process = await run_git(
         *cmd_args,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
+        cwd=base_dir,
         env={
             'LANG': 'C',
             'LC_ALL': 'C',
@@ -51,9 +50,6 @@ async def clone_repo(repo: Repo, base_dir: Path) -> bool:
         },
     )
 
-    # Wait for the process to finish and capture output
-    stdout, stderr = await process.communicate()
-
     if process.returncode == 0:
         log.info("Cloned '%s' successfully", repo)
         return True
@@ -63,9 +59,9 @@ async def clone_repo(repo: Repo, base_dir: Path) -> bool:
             repo,
             process.returncode,
         )
-        if stderr:
+        if process.stderr:
             # Log each line of stderr with a prefix for better readability
-            error_output = stderr.decode().strip()
+            error_output = process.stderr.strip()
             for line in error_output.splitlines():
                 log.error('  [git] %s', line)
         return False
@@ -127,4 +123,4 @@ async def process(context: DocBuildContext, repos: tuple[str, ...]) -> int:
     # Return 0 for success (all clones succeeded), 1 for failure.
     if all(results):
         return 0
-    return 1
+    return 1

src/docbuild/cli/commands.py

@@ -0,0 +1,139 @@
+""" """
+
+import asyncio
+from collections.abc import Mapping, Sequence
+import os
+import shutil
+from subprocess import CompletedProcess
+from typing import IO, Any, cast
+
+# Type aliases for better readability
+type StrOrBytesPath = str | bytes | os.PathLike[str] | os.PathLike[bytes]
+type _TXT = str | bytes
+type _ENV = Mapping[bytes, _TXT] | Mapping[str, _TXT]
+type AnyPath = str | bytes | os.PathLike[str] | os.PathLike[bytes]
+type _CMD = _TXT | Sequence[AnyPath]
+
+
+def resolve_command(program: StrOrBytesPath, *args: StrOrBytesPath) -> tuple[str, ...]:
+    """Resolve the program to an absolute path and return the full command tuple.
+
+    If the program is already an absolute path, it is used as is. Otherwise,
+    it is searched for in the system's PATH.
+
+    :param program: The program to execute.
+    :param args: The arguments to pass to the program.
+    :return: A tuple containing the resolved program path and the arguments.
+    :raises FileNotFoundError: If the program is not found or not executable.
+    """
+    # shutil.which can handle absolute paths, so we don't need to check for them.
+    # It will return the path if it's executable, or None otherwise.
+    resolved_program = shutil.which(str(program))
+
+    if resolved_program is None:
+        raise FileNotFoundError(
+            f"Program '{program}' not found in PATH or not executable."
+        )
+
+    # Cast args to string to create a uniform tuple
+    str_args = tuple(str(arg) for arg in args)
+    return (resolved_program, *str_args)
+
+
+async def run_command(
+    program: StrOrBytesPath,
+    *args: StrOrBytesPath,
+    stdin: IO[Any] | int | None = None,
+    stdout: IO[Any] | int | None = None,
+    stderr: IO[Any] | int | None = None,
+    cwd: StrOrBytesPath | None = None,
+    env: _ENV | None = None,
+    **kwds,  # noqa: ANN003
+) -> CompletedProcess:
+    """Run a command asynchronously and return the result.
+
+    :param program: The command to run.
+    :param args: The arguments to pass to git.
+    :param stdin: The standard input stream to use.
+    :param stdout: The standard output stream to use.
+    :param stderr: The standard error stream to use.
+    :param cwd: The working directory to run the command in.
+    :param env: The environment variables to pass to the command.
+    :param kwds: Additional keyword arguments to pass to the subprocess.
+    :return: The class:`~subprocess.CompletedProcess` object.
+    """
+    program = str(program)
+    args = tuple(str(arg) for arg in args)
+    clone_process = await asyncio.create_subprocess_exec(
+        program,
+        *args,
+        cwd=cwd,
+        env=env,
+        stdin=stdin,
+        stderr=stderr,
+        stdout=stdout,
+        **kwds,
+    )
+    stdout_data, stderr_data = await clone_process.communicate()
+    return CompletedProcess(
+        args=[program, *args],
+        returncode=cast(int, clone_process.returncode),
+        stdout=stdout_data.decode() if stdout_data else None,
+        stderr=stderr_data.decode() if stderr_data else None,
+    )
+
+
+async def run_git(
+    *args: StrOrBytesPath,
+    cwd: StrOrBytesPath | None = None,
+    env: _ENV | None = None,
+    **kwds,  # noqa: ANN003
+) -> CompletedProcess:
+    """Run a git command asynchronously.
+
+    The :command:`git` command is expected to be in the system PATH.
+
+    :param args: The arguments to pass to git without the command.
+    :param cwd: The working directory to run the command in.
+    :param env: The environment variables to pass to the command.
+    :param kwds: Additional keyword arguments to pass to the subprocess.
+    :return: The class:`~subprocess.CompletedProcess` object.
+    :raise FileNotFoundError: If the git executable is not found in PATH.
+    """
+    program, *resolved_args = resolve_command('git', *args)
+
+    return await run_command(
+        program,
+        *resolved_args,
+        cwd=cwd,
+        env=env,
+        **kwds,
+    )
+
+
+async def run_daps(
+    *args: StrOrBytesPath,
+    cwd: StrOrBytesPath | None = None,
+    env: _ENV | None = None,
+    **kwds,  # noqa: ANN003
+) -> CompletedProcess:
+    """Run the DAPS command asynchronously.
+
+    The :command:`daps` command is expected to be in the system PATH.
+
+    :param args: The arguments to pass to DAPS without the command.
+    :param cwd: The working directory to run the command in.
+    :param env: The environment variables to pass to the command.
+    :param kwds: Additional keyword arguments to pass to the subprocess.
+    :return: The class:`~subprocess.CompletedProcess` object.
+    :raises FileNotFoundError: If the DAPS executable is not found in PATH.
+    """
+    program, *resolved_args = resolve_command('daps', *args)
+
+    return await run_command(
+        program,
+        *resolved_args,
+        cwd=cwd,
+        env=env,
+        **kwds,
+    )