feature: Add pex_set_default_index_url() function and lookup defaul…

…t index URL (#220)

* feature: Add `pex_set_default_index_url()` function

* fmt

* lookup python package index from project settings

* fix mypy

.changelog/_unreleased.toml

@@ -0,0 +1,5 @@
+[[entries]]
+id = "670a651f-b0d2-4d1a-881e-94a70c885749"
+type = "feature"
+description = "Add `pex_set_default_index_url()` function"
+author = "[email protected]"

kraken-build/src/kraken/std/python/tasks/flake8_task.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from pathlib import Path
 
 from kraken.common import Supplier
@@ -41,6 +42,7 @@ def flake8(
     project: Project | None = None,
     config_file: Path | Supplier[Path] | None = None,
     version_spec: str | None = None,
+    additional_requirements: Sequence[str] = (),
 ) -> Flake8Task:
     """Creates a task for linting your Python project with Flake8.
 
@@ -52,7 +54,10 @@ def flake8(
 
     if version_spec is not None:
         flake8_bin = pex_build(
-            "flake8", requirements=[f"flake8{version_spec}"], console_script="flake8", project=project
+            "flake8",
+            requirements=[f"flake8{version_spec}", *additional_requirements],
+            console_script="flake8",
+            project=project,
         ).output_file.map(str)
     else:
         flake8_bin = Supplier.of("flake8")

kraken-build/src/kraken/std/python/tasks/pex_build_task.py

@@ -10,8 +10,10 @@
 from kraken.core.system.project import Project
 from kraken.core.system.property import Property
 from kraken.core.system.task import Task, TaskStatus
+from kraken.std.util.url import inject_url_credentials, redact_url_password
 
 logger = logging.getLogger(__name__)
+default_index_url: str | None = None
 
 
 class PexBuildTask(Task):
@@ -26,6 +28,7 @@ class PexBuildTask(Task):
     venv: Property[Literal["prepend", "append"] | None] = Property.default(None)
     pex_binary: Property[Path | None] = Property.default(None)
     python: Property[Path | None] = Property.default(None)
+    index_url: Property[str | None] = Property.default(None)
 
     #: The path to the built PEX file will be written to this property.
     output_file: Property[Path] = Property.output()
@@ -70,6 +73,7 @@ def execute(self) -> TaskStatus | None:
                 venv=self.venv.get(),
                 pex_binary=self.pex_binary.get(),
                 python=self.python.get(),
+                index_url=self.index_url.get() or _get_default_index_url(self.project),
             )
         except subprocess.CalledProcessError as exc:
             return TaskStatus.from_exit_code(exc.cmd, exc.returncode)
@@ -88,6 +92,7 @@ def _build_pex(
     inject_env: Mapping[str, str] | None = None,
     pex_binary: Path | None = None,
     python: Path | None = None,
+    index_url: str | None = None,
     log: logging.Logger | None = None,
 ) -> None:
     """Invokes the `pex` CLI to build a PEX file and write it to :param:`output_file`.
@@ -105,12 +110,7 @@ def _build_pex(
     if pex_binary is not None:
         command = [str(pex_binary)]
     else:
-        command = [
-            str(python or sys.executable),
-            "-m",
-            "pex",
-            "-v",
-        ]
+        command = [str(python or sys.executable), "-m", "pex", "-v"]
 
     command += [
         "--pip-version",
@@ -132,7 +132,12 @@ def _build_pex(
     for key, value in (inject_env or {}).items():
         command += ["--inject-env", f"{key}={value}"]
 
-    (log or logging).info("Building PEX $ %s", " ".join(map(shlex.quote, command)))
+    safe_command = list(command)
+    if index_url is not None:
+        command += ["--index-url", index_url]
+        safe_command += ["--index-url", redact_url_password(index_url)]
+
+    (log or logging).info("Building PEX $ %s", " ".join(map(shlex.quote, safe_command)))
     subprocess.run(command, check=True)
 
 
@@ -144,6 +149,7 @@ def pex_build(
     console_script: str | None = None,
     interpreter_constraint: str | None = None,
     venv: Literal["prepend", "append"] | None = None,
+    index_url: str | None = None,
     task_name: str | None = None,
     project: Project | None = None,
 ) -> PexBuildTask:
@@ -161,6 +167,7 @@ def pex_build(
         and existing_task.console_script.get() == console_script
         and existing_task.interpreter_constraint.get() == interpreter_constraint
         and existing_task.venv.get() == venv
+        and existing_task.index_url.get() == index_url
     ):
         return existing_task
 
@@ -171,4 +178,32 @@ def pex_build(
     task.console_script = console_script
     task.interpreter_constraint = interpreter_constraint
     task.venv = venv
+    task.index_url = index_url
     return task
+
+
+def pex_set_default_index_url(url: str) -> None:
+    """Set the default index URL for Pex globally."""
+
+    global default_index_url
+    default_index_url = url
+
+
+def _get_default_index_url(project: Project | None) -> str | None:
+    """Looks up the default Python package index in the Python settings of the current project, or falls back
+    to the global `default_index_url`."""
+
+    from kraken.std.python.pyproject import PackageIndex
+    from kraken.std.python.settings import python_settings
+
+    settings = python_settings(project=project)
+    for idx in settings.package_indexes.values():
+        if idx.priority == PackageIndex.Priority.default:
+            break
+    else:
+        return default_index_url
+
+    if idx.credentials:
+        return inject_url_credentials(idx.index_url, *idx.credentials)
+
+    return idx.index_url

kraken-build/src/kraken/std/util/url.py

@@ -0,0 +1,18 @@
+from urllib.parse import urlparse
+
+
+def redact_url_password(url: str, placeholder: str = "[REDACTED]") -> str:
+    """Redacts the password in a URL with the given placeholder, if any."""
+
+    parsed = urlparse(url)
+    if parsed.password:
+        replaced = parsed._replace(netloc=f"{parsed.username}:{placeholder}@{parsed.hostname}")
+    return replaced.geturl()
+
+
+def inject_url_credentials(url: str, username: str, password: str) -> str:
+    """Injects a username and password into a URL."""
+
+    parsed = urlparse(url)
+    replaced = parsed._replace(netloc=f"{username}:{password}@{parsed.hostname}")
+    return replaced.geturl()