[PULP-1283] Use podman container for OpenAPI schema generation

.pre-commit-config.yaml

@@ -13,11 +13,11 @@ repos:
   - repo: "https://github.com/astral-sh/ruff-pre-commit"
     rev: "v0.11.10"
     hooks:
-      # Run the linter.
-      - id: "ruff-check"
-        args: ["--show-fixes"]
       # Run the formatter.
       - id: "ruff-format"
+      # Run the linter.
+      - id: "ruff-check"
+        args: ["--show-fixes", "--fix"]
 
   - repo: "https://github.com/adrienverge/yamllint"
     rev: "v1.37.1"

src/pulp_docs/cli.py

@@ -7,7 +7,7 @@
 from mkdocs.__main__ import cli as mkdocs_cli
 
 from pulp_docs.context import ctx_blog, ctx_docstrings, ctx_draft, ctx_dryrun, ctx_path
-from pulp_docs.plugin import ComponentLoader, default_lookup_paths
+from pulp_docs.plugin import ComponentLoader, ComponentSpec, default_lookup_paths
 
 
 def blog_callback(ctx: click.Context, param: click.Parameter, value: bool) -> bool:
@@ -94,6 +94,53 @@ async def clone_repository(repo_url: str) -> None:
 )
 
 
+def fetch_repositories(
+    dest: Path,
+    config_file: Path | None = None,
+    component_filter: list[str] | None = None,
+    fetch_all: bool = False,
+) -> list[ComponentSpec]:
+    """Fetch missing repositories and return component specs.
+
+    Args:
+        dest: Destination directory for cloned repositories
+        config_file: Path to mkdocs.yml config file (defaults to bundled config)
+        component_filter: Optional list of component names to fetch (fetches all if None)
+        fetch_all: If True, fetch all components from all_specs (not just missing)
+
+    Returns:
+        List of ComponentSpec objects for the fetched/available components
+    """
+    if config_file is None:
+        config_file = get_default_mkdocs()
+
+    lookup_paths = default_lookup_paths()
+    component_loader = ComponentLoader(lookup_paths, mkdocs_config=config_file)
+    load_result = component_loader.load_all()
+
+    # Determine which components to fetch
+    if fetch_all:
+        to_fetch = load_result.all_specs
+    else:
+        to_fetch = load_result.missing
+
+    if component_filter:
+        to_fetch = [c for c in to_fetch if c.component_name in component_filter]
+
+    repos_to_fetch = {comp.git_url for comp in to_fetch}
+
+    if repos_to_fetch:
+        dest.mkdir(parents=True, exist_ok=True)
+        asyncio.run(clone_repositories(repos_to_fetch, dest))
+
+    # Return specs for requested components
+    specs = load_result.all_specs
+    if component_filter:
+        specs = [s for s in specs if s.component_name in component_filter]
+
+    return specs
+
+
 @click.command()
 @click.option(
     "--dest",
@@ -121,13 +168,7 @@ async def clone_repository(repo_url: str) -> None:
 def fetch(dest, config_file, path_exclude):
     """Fetch repositories to destination dir."""
     dest_path = Path(dest)
-    lookup_paths = default_lookup_paths()
-    component_loader = ComponentLoader(lookup_paths, mkdocs_config=config_file)
-    missing_comps = component_loader.load_all().missing
-    missing_repos = {comp.git_url for comp in missing_comps}
-    if not dest_path.exists():
-        dest_path.mkdir(parents=True)
-    asyncio.run(clone_repositories(missing_repos, dest_path))
+    fetch_repositories(dest_path, config_file)
 
 
 main = mkdocs_cli

src/pulp_docs/openapi.py

@@ -2,143 +2,179 @@
 Module for generating open-api json files for selected Pulp plugins.
 """
 
-import argparse
+from __future__ import annotations
+
 import os
 import shutil
 import subprocess
-import tempfile
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Optional
+from typing import NamedTuple, Optional
 
-from pulp_docs.cli import get_default_mkdocs
-from pulp_docs.plugin import ComponentLoader, ComponentSpec, default_lookup_paths
+CONTAINER_IMAGE = "quay.io/pulp/pulp-minimal:stable"
+CONTAINER_NAME_PREFIX = "pulpdocs-openapi"
+CONTAINER_OUTPUT_PATH = "/output"
 
-BASE_TMPDIR_NAME = "pulpdocs_tmp"
-CURRENT_DIR = Path(__file__).parent.absolute()
 
+class PodmanError(Exception):
+    """Raised when a podman command fails."""
 
-def main(output_dir: Path, filter_list: Optional[list[str]] = None, dry_run: bool = False):
-    """Creates openapi json files for found plugins in the output_dir.
+    pass
 
-    Optionally filter the found plugins with a filter list.
-    """
 
-    def select_component_fn(comp: ComponentSpec) -> bool:
-        name = comp.component_name
-        return (bool(filter_list) and name in filter_list) or name == "pulpcore"
+class PluginInstallError(Exception):
+    """Raised when plugin installation fails."""
+
+    def __init__(self, repository_paths: list[Path]):
+        plugins_str = ", ".join(str(path.name) for path in repository_paths)
+        message = (
+            "Failed to install plugins.\n"
+            "Are all these components Pulp projects with REST APIs?\n"
+            f"Trying with: {plugins_str}"
+        )
+        super().__init__(message)
+        self.repository_paths = repository_paths
+
 
-    mkdocs_config = get_default_mkdocs()
-    lookup_paths = default_lookup_paths()
-    component_loader = ComponentLoader(lookup_paths, mkdocs_config=mkdocs_config)
-    all_specs = component_loader.load_all().all_specs
-    selected = list(filter(select_component_fn, all_specs))
-    openapi = OpenAPIGenerator(plugins=selected, dry_run=dry_run)
-    openapi.generate(target_dir=output_dir)
+class OpenApiPlugin(NamedTuple):
+    repository_path: Path
+    plugin_label: str
 
 
 class OpenAPIGenerator:
-    """
-    Responsible for setting up a python environment with the required
-    Pulp packages to generate openapi schemas for all registered plugins.
+    """Generate openapi schemas for all registered plugins.
 
     Args:
-        plugin_remotes: A list of git remote urls of the required Pulp packages.
+        plugins: A list of OpenApiPlugin with repository paths and labels.
         dry_run: Whether it should execute the commands or just show them.
+        image: The container image to use.
     """
 
-    def __init__(self, plugins: list[ComponentSpec], dry_run=False):
-        self.pulpcore = next(filter(lambda p: p.component_name == "pulpcore", plugins))
-        self.plugins = plugins + [self.pulpcore]
+    def __init__(
+        self,
+        plugins: list[OpenApiPlugin],
+        dry_run: bool = False,
+        image: str = CONTAINER_IMAGE,
+    ):
+        self.plugins = plugins
+        self.repository_paths = list({p.repository_path for p in plugins if p.repository_path})
         self.dry_run = dry_run
+        self.image = image
 
-        # setup working tmpdir
-        self.tmpdir = Path(tempfile.gettempdir()) / BASE_TMPDIR_NAME / "openapi"
-        self.venv_path = os.path.join(self.tmpdir, "venv")
-
-        shutil.rmtree(self.tmpdir, ignore_errors=True)
-        os.makedirs(self.tmpdir, exist_ok=True)
-
-    def generate(self, target_dir: Path):
+    def generate(self, output_dir: Path):
         """Generate openapi json files at target directory."""
+        if not self.repository_paths:
+            return  # nothing to do here
+        self._check_podman()
+        output_dir.mkdir(parents=True, exist_ok=True)
+        container = self._init_container(output_dir)
+        with container.run():
+            self._install_plugins(container)
+            self._generate_schemas(container)
+
+    def _init_container(self, output_dir: Path):
+        abs_target = str(output_dir.resolve())
+        volumes = {abs_target: CONTAINER_OUTPUT_PATH}
+
+        # Mount repository paths as volumes using repository names
+        for repo_path in self.repository_paths:
+            container_repo_path = f"/repos/{repo_path.name}"
+            volumes[str(repo_path.resolve())] = container_repo_path
+
+        return PodmanContainer(
+            image=self.image,
+            volumes=volumes,
+            env={"PULP_CONTENT_ORIGIN": "NONE"},
+            dry_run=self.dry_run,
+        )
+
+    def _install_plugins(self, container: PodmanContainer):
+        pip_args = [f"/repos/{repo_path.name}" for repo_path in self.repository_paths]
+        try:
+            container.exec("pip", "install", *pip_args)
+        except PodmanError as e:
+            raise PluginInstallError(self.repository_paths) from e
+
+    def _check_podman(self):
+        if not self.dry_run and not shutil.which("podman"):
+            raise RuntimeError("podman is required but was not found on PATH. ")
+
+    def _generate_schemas(self, container: PodmanContainer):
         for plugin in self.plugins:
-            self.setup_venv(plugin)
-            outfile = str(target_dir / f"{plugin.label}-api.json")
-            self.run_python(
+            outfile = f"{CONTAINER_OUTPUT_PATH}/{plugin.plugin_label}-api.json"
+            container.exec(
                 "pulpcore-manager",
                 "openapi",
                 "--component",
-                plugin.label,
+                plugin.plugin_label,
                 "--file",
                 outfile,
             )
 
-    def setup_venv(self, plugin: ComponentSpec):
-        """
-        Creates virtualenv with plugin.
-        """
-        create_venv_cmd = ("python", "-m", "venv", self.venv_path)
-        # setuptools provides distutils for python >=3.12.
-        install_cmd = ["pip", "install", f"git+{plugin.git_url}", "setuptools"]
-
-        if self.dry_run is True:
-            print(" ".join(create_venv_cmd))
-        else:
-            shutil.rmtree(self.venv_path, ignore_errors=True)
-            subprocess.run(create_venv_cmd, check=True)
-
-        self.run_python(*install_cmd)
-
-    def run_python(self, *cmd: str) -> str:
-        """Run a binary command from within the tmp venv.
-
-        Basically: $tmp-venv/bin/{first-arg} {remaining-args}
-        """
-        cmd_bin = os.path.join(self.venv_path, f"bin/{cmd[0]}")
-        final_cmd = [cmd_bin] + list(cmd[1:])
-        if self.dry_run is True:
-            cmd_str = " ".join(final_cmd)
-            print(cmd_str)
-            return cmd_str
-
-        os.environ["PULP_CONTENT_ORIGIN"] = "NONE"
-        result = subprocess.run(final_cmd, check=True)
-        return result.stdout.decode() if result.stdout else ""
-
-
-def parse_args():
-    parser = argparse.ArgumentParser(
-        "pulp-docs openapi generation",
-        description="Creates a venv for each plugin and generate its openapi-json to output_dir.",
-    )
-    parser.add_argument(
-        "output_dir", help="The directory where the {plugin}-api.json will be stored."
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",  # default False
-        help="Dont run the commands, only output how they are constructed.",
-    )
-    parser.add_argument(
-        "-l",
-        "--plugin-list",
-        type=str,
-        help="List of plugins that should be used. Use all if omitted.",
-    )
-    args = parser.parse_args()
-
-    # validation
-    if not os.path.isdir(args.output_dir):
-        raise TypeError("Must provide an existing directory.")
-    return args
-
-
-if __name__ == "__main__":
-    args = parse_args()
-    dry_run = args.dry_run
-    dest = Path(args.output_dir)
-
-    filter_list = []
-    if args.plugin_list:
-        filter_list = [str(p) for p in args.plugin_list.split(",") if p]
-
-    main(dest, filter_list, dry_run)
+
+class PodmanContainer:
+    """Manage a podman container lifecycle (create, start, exec, remove).
+
+    Use the `run` context manager to ensure cleanup::
+
+        container = PodmanContainer(image="myimage", volumes={"/host": "/container"})
+        with container.run():
+            container.exec("pip", "install", "some-package")
+            container.exec("my-command", "--flag", "value")
+    """
+
+    def __init__(
+        self,
+        image: str,
+        volumes: Optional[dict[str, str]] = None,
+        env: Optional[dict[str, str]] = None,
+        name: Optional[str] = None,
+        dry_run: bool = False,
+    ):
+        self.image = image
+        self.volumes = volumes or {}
+        self.env = env or {}
+        self.name = name or f"{CONTAINER_NAME_PREFIX}-{os.getpid()}"
+        self.dry_run = dry_run
+
+    @contextmanager
+    def run(self):
+        """Start the container and remove it on exit."""
+        self._create()
+        self._start()
+        try:
+            yield self
+        finally:
+            self._remove()
+
+    def exec(self, *cmd: str):
+        """Run a command inside the container."""
+        self._run_podman("exec", self.name, *cmd)
+
+    def _create(self):
+        cmd = ["create", "--name", self.name]
+        for key, value in self.env.items():
+            cmd.extend(["-e", f"{key}={value}"])
+        for host_path, container_path in self.volumes.items():
+            cmd.extend(["-v", f"{host_path}:{container_path}:Z"])
+        cmd.extend([self.image, "sleep", "infinity"])
+        self._run_podman(*cmd)
+
+    def _start(self):
+        self._run_podman("start", self.name)
+
+    def _remove(self):
+        self._run_podman("rm", "--force", self.name)
+
+    def _run_podman(self, *args: str):
+        cmd = ["podman", *args]
+        if self.dry_run:
+            print(" ".join(cmd))
+            return
+        try:
+            subprocess.run(cmd, check=True)
+        except subprocess.CalledProcessError as e:
+            cmd_str = " ".join(cmd)
+            raise PodmanError(
+                f"Podman command failed: {cmd_str}\nReturn code: {e.returncode}"
+            ) from e

src/pulp_docs/plugin.py

@@ -454,6 +454,7 @@ def on_config(self, config: MkDocsConfig) -> MkDocsConfig | None:
         mkdocs_file = self.mkdocs_yml_dir / "mkdocs.yml"
         log_pulp_config(mkdocs_file, lookup_paths, self.loaded_comps, config.site_dir)
 
+        # Configure mkdocs plugins
         mkdocstrings_config = config.plugins["mkdocstrings"].config
         components_var = []
         for component in self.loaded_comps: