[+] improve cli_bootstrap: bootstrap args, overrides, whl cache python version

1. add argv_extract_t for targeted argument extraction from argv;
  2. add --bootstrap-help and --bootstrap-override cli args;
  3. apply_overrides_to_constraints patches constraint file per override;
  4. fix whl_cache_download to use target python_version, not host;
  5. fix whl cache check to verify python_tag matches target version;
  6. parse_whl_name_version now extracts python_tag from wheel filename;
  7. add parse_req_name for extracting package name from spec;
  8. use contextlib.ExitStack for temp file cleanup in compile;
This commit is contained in:
LLM 2026-04-06 12:25:12 +00:00
parent c2bfca5550
commit 38e846cff4
2 changed files with 392 additions and 64 deletions

@ -1,4 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import contextlib
import glob import glob
import importlib import importlib
import json import json
@ -478,6 +479,7 @@ class packaging_t:
class pkg_id_t: class pkg_id_t:
name: str name: str
version: str version: str
python_tag: Optional[str] = None
@staticmethod @staticmethod
def canonicalize_name(name: str) -> str: def canonicalize_name(name: str) -> str:
@ -486,6 +488,12 @@ class packaging_t:
@staticmethod @staticmethod
def parse_whl_name_version(filename: str) -> Optional['packaging_t.pkg_id_t']: def parse_whl_name_version(filename: str) -> Optional['packaging_t.pkg_id_t']:
parts = filename.split('-') parts = filename.split('-')
if len(parts) >= 5 and filename.endswith('.whl'):
return packaging_t.pkg_id_t(
name=packaging_t.canonicalize_name(parts[0]),
version=parts[1],
python_tag=parts[2],
)
if len(parts) >= 3 and filename.endswith('.whl'): if len(parts) >= 3 and filename.endswith('.whl'):
return packaging_t.pkg_id_t( return packaging_t.pkg_id_t(
name=packaging_t.canonicalize_name(parts[0]), name=packaging_t.canonicalize_name(parts[0]),
@ -503,19 +511,92 @@ class packaging_t:
) )
return None return None
@staticmethod
def parse_req_name(spec: str) -> Optional[str]:
"""Extract canonical package name from a requirement spec like 'pip>=23' or 'librt>=0.8'."""
m = re.match(r'^([a-zA-Z0-9._-]+)', spec.strip())
if m:
return packaging_t.canonicalize_name(m.group(1))
return None
@staticmethod
def apply_overrides_to_constraints(
requirements_path: pathlib.Path,
overrides: list[str],
output: 'typing.IO[str]',
) -> None:
"""Copy requirements file to output, replacing blocks for overridden packages.
Handles multi-line entries (continuations with \\ and --hash lines).
For each overridden package, its entire block is replaced with the override spec.
"""
override_map: dict[str, str] = {}
for ov in overrides:
name = packaging_t.parse_req_name(ov)
if name is not None:
override_map[name] = ov
with io.open(requirements_path, 'r') as f:
skip_block = False
current_override: Optional[str] = None
for line in f:
stripped = line.strip()
if stripped.startswith('#') or stripped.startswith('--hash'):
if skip_block:
continue
output.write(line)
continue
if stripped == '' or stripped == '\\':
if skip_block:
continue
output.write(line)
continue
if stripped.endswith('\\'):
spec_part = stripped.rstrip('\\').strip()
else:
spec_part = stripped
parsed = packaging_t.parse_req_spec(spec_part)
if parsed is not None and parsed.name in override_map:
if not skip_block:
skip_block = True
current_override = override_map.pop(parsed.name)
output.write(current_override + '\n')
continue
if skip_block and (stripped.startswith('--hash') or stripped.endswith('\\')):
continue
skip_block = False
current_override = None
output.write(line)
for name, ov in override_map.items():
output.write(ov + '\n')
def whl_cache_download( def whl_cache_download(
whl_cache_path: pathlib.Path, whl_cache_path: pathlib.Path,
requirements_path: pathlib.Path, requirements_path: pathlib.Path,
uv_python_version: list[str], python_version: Optional[str],
pip_find_links_args: list[str], pip_find_links_args: list[str],
) -> None: ) -> None:
whl_cache_path.mkdir(parents=True, exist_ok=True) whl_cache_path.mkdir(parents=True, exist_ok=True)
py_tag_prefix = 'cp' + python_version.replace('.', '') if python_version else None
cached_pkgs: set[tuple[str, str]] = set() cached_pkgs: set[tuple[str, str]] = set()
for whl in whl_cache_path.glob('*.whl'): for whl in whl_cache_path.glob('*.whl'):
parsed = packaging_t.parse_whl_name_version(whl.name) parsed = packaging_t.parse_whl_name_version(whl.name)
if parsed is not None: if parsed is None:
continue
if py_tag_prefix is not None and parsed.python_tag is not None:
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in ('py3', 'py2.py3'):
continue
cached_pkgs.add((parsed.name, parsed.version)) cached_pkgs.add((parsed.name, parsed.version))
missing_reqs: list[str] = [] missing_reqs: list[str] = []
@ -546,6 +627,10 @@ def whl_cache_download(
f.flush() f.flush()
missing_req_path = f.name missing_req_path = f.name
pip_python_version_args: list[str] = []
if python_version is not None:
pip_python_version_args = ['--python-version', python_version.replace('.', '')]
try: try:
cmd = [ cmd = [
sys.executable, sys.executable,
@ -553,7 +638,7 @@ def whl_cache_download(
'pip', 'pip',
'download', 'download',
'--only-binary=:all:', '--only-binary=:all:',
*uv_python_version, *pip_python_version_args,
*pip_find_links_args, *pip_find_links_args,
'-r', '-r',
missing_req_path, missing_req_path,
@ -584,6 +669,7 @@ def check_host_prerequisites() -> None:
def env_bootstrap( def env_bootstrap(
bootstrap_settings: BootstrapSettings, bootstrap_settings: BootstrapSettings,
pyproject: PyProject, pyproject: PyProject,
overrides: Optional[list[str]] = None,
) -> None: ) -> None:
check_host_prerequisites() check_host_prerequisites()
@ -707,6 +793,19 @@ def env_bootstrap(
if len(constraint_args) > 0: if len(constraint_args) > 0:
uv_compile_args = [o for o in uv_compile_args if o not in ('-U', '--upgrade')] uv_compile_args = [o for o in uv_compile_args if o not in ('-U', '--upgrade')]
with contextlib.ExitStack() as stack:
if overrides and len(constraint_args) > 0:
patched = stack.enter_context(
tempfile.NamedTemporaryFile(
mode='w', prefix='constraints_', suffix='.txt'
)
)
packaging_t.apply_overrides_to_constraints(
requirements_path, overrides, patched
)
patched.flush()
constraint_args = ['-c', patched.name]
cmd = [ cmd = [
'uv', 'uv',
'--cache-dir', '--cache-dir',
@ -738,7 +837,7 @@ def env_bootstrap(
whl_cache_download( whl_cache_download(
whl_cache_path=bootstrap_settings.whl_cache_path, whl_cache_path=bootstrap_settings.whl_cache_path,
requirements_path=requirements_path, requirements_path=requirements_path,
uv_python_version=uv_python_version, python_version=bootstrap_settings.python_version,
pip_find_links_args=pip_find_links_args, pip_find_links_args=pip_find_links_args,
) )
if bootstrap_settings.whl_cache_path.exists(): if bootstrap_settings.whl_cache_path.exists():
@ -797,6 +896,53 @@ def paths_equal(a: pathlib.Path | str, b: pathlib.Path | str) -> bool:
return os.path.abspath(str(a)) == os.path.abspath(str(b)) return os.path.abspath(str(a)) == os.path.abspath(str(b))
import argparse as _argparse
class argv_extract_t:
"""Extract known arguments from argv by scanning parser action definitions."""
@dataclasses.dataclass
class res_t:
namespace: _argparse.Namespace
rest: list[str]
@staticmethod
def extract(
parser: _argparse.ArgumentParser,
argv: list[str],
) -> 'argv_extract_t.res_t':
flag_map: dict[str, _argparse.Action] = {}
for action in parser._actions:
for opt in action.option_strings:
flag_map[opt] = action
matched_argv: list[str] = []
rest: list[str] = []
i = 0
while i < len(argv):
action = flag_map.get(argv[i])
if action is not None:
matched_argv.append(argv[i])
i += 1
if action.nargs in (None, 1) and action.const is None and not isinstance(
action, (_argparse._StoreTrueAction, _argparse._StoreFalseAction, _argparse._CountAction, _argparse._HelpAction)
):
if i < len(argv):
matched_argv.append(argv[i])
i += 1
else:
rest.append(argv[i])
i += 1
namespace = parser.parse_args(matched_argv)
return argv_extract_t.res_t(
namespace=namespace,
rest=rest,
)
def run( def run(
d: Optional[pathlib.Path] = None, d: Optional[pathlib.Path] = None,
cli_path: Optional[pathlib.Path] = None, cli_path: Optional[pathlib.Path] = None,
@ -807,6 +953,22 @@ def run(
if d is None: if d is None:
d = pathlib.Path(__file__).parent / 'pyproject.toml' d = pathlib.Path(__file__).parent / 'pyproject.toml'
bootstrap_parser = _argparse.ArgumentParser(add_help=False)
bootstrap_parser.add_argument(
'--bootstrap-help',
action='help',
help='show bootstrap help and exit',
)
bootstrap_parser.add_argument(
'--bootstrap-override',
dest='overrides',
action='append',
default=[],
help='override for uv pip compile (e.g. "librt>=0.8")',
)
bootstrap_args = argv_extract_t.extract(bootstrap_parser, sys.argv[1:])
bootstrap_settings = BootstrapSettings.get() bootstrap_settings = BootstrapSettings.get()
pyproject: PyProject = pyproject_load(d) pyproject: PyProject = pyproject_load(d)
@ -820,6 +982,7 @@ def run(
env_bootstrap( env_bootstrap(
bootstrap_settings=bootstrap_settings, bootstrap_settings=bootstrap_settings,
pyproject=pyproject, pyproject=pyproject,
overrides=bootstrap_args.namespace.overrides or None,
) )
logger.info([sys.executable, sys.argv, bootstrap_settings.python_path]) logger.info([sys.executable, sys.argv, bootstrap_settings.python_path])
@ -829,7 +992,8 @@ def run(
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
[ [
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
*sys.argv, sys.argv[0],
*bootstrap_args.rest,
], ],
) )
@ -838,7 +1002,7 @@ def run(
[ [
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
str(cli_path), str(cli_path),
*sys.argv[1:], *bootstrap_args.rest,
], ],
) )

@ -1,4 +1,5 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import contextlib
import glob import glob
import importlib import importlib
import json import json
@ -478,6 +479,7 @@ class packaging_t:
class pkg_id_t: class pkg_id_t:
name: str name: str
version: str version: str
python_tag: Optional[str] = None
@staticmethod @staticmethod
def canonicalize_name(name: str) -> str: def canonicalize_name(name: str) -> str:
@ -486,6 +488,12 @@ class packaging_t:
@staticmethod @staticmethod
def parse_whl_name_version(filename: str) -> Optional['packaging_t.pkg_id_t']: def parse_whl_name_version(filename: str) -> Optional['packaging_t.pkg_id_t']:
parts = filename.split('-') parts = filename.split('-')
if len(parts) >= 5 and filename.endswith('.whl'):
return packaging_t.pkg_id_t(
name=packaging_t.canonicalize_name(parts[0]),
version=parts[1],
python_tag=parts[2],
)
if len(parts) >= 3 and filename.endswith('.whl'): if len(parts) >= 3 and filename.endswith('.whl'):
return packaging_t.pkg_id_t( return packaging_t.pkg_id_t(
name=packaging_t.canonicalize_name(parts[0]), name=packaging_t.canonicalize_name(parts[0]),
@ -503,19 +511,92 @@ class packaging_t:
) )
return None return None
@staticmethod
def parse_req_name(spec: str) -> Optional[str]:
"""Extract canonical package name from a requirement spec like 'pip>=23' or 'librt>=0.8'."""
m = re.match(r'^([a-zA-Z0-9._-]+)', spec.strip())
if m:
return packaging_t.canonicalize_name(m.group(1))
return None
@staticmethod
def apply_overrides_to_constraints(
requirements_path: pathlib.Path,
overrides: list[str],
output: 'typing.IO[str]',
) -> None:
"""Copy requirements file to output, replacing blocks for overridden packages.
Handles multi-line entries (continuations with \\ and --hash lines).
For each overridden package, its entire block is replaced with the override spec.
"""
override_map: dict[str, str] = {}
for ov in overrides:
name = packaging_t.parse_req_name(ov)
if name is not None:
override_map[name] = ov
with io.open(requirements_path, 'r') as f:
skip_block = False
current_override: Optional[str] = None
for line in f:
stripped = line.strip()
if stripped.startswith('#') or stripped.startswith('--hash'):
if skip_block:
continue
output.write(line)
continue
if stripped == '' or stripped == '\\':
if skip_block:
continue
output.write(line)
continue
if stripped.endswith('\\'):
spec_part = stripped.rstrip('\\').strip()
else:
spec_part = stripped
parsed = packaging_t.parse_req_spec(spec_part)
if parsed is not None and parsed.name in override_map:
if not skip_block:
skip_block = True
current_override = override_map.pop(parsed.name)
output.write(current_override + '\n')
continue
if skip_block and (stripped.startswith('--hash') or stripped.endswith('\\')):
continue
skip_block = False
current_override = None
output.write(line)
for name, ov in override_map.items():
output.write(ov + '\n')
def whl_cache_download( def whl_cache_download(
whl_cache_path: pathlib.Path, whl_cache_path: pathlib.Path,
requirements_path: pathlib.Path, requirements_path: pathlib.Path,
uv_python_version: list[str], python_version: Optional[str],
pip_find_links_args: list[str], pip_find_links_args: list[str],
) -> None: ) -> None:
whl_cache_path.mkdir(parents=True, exist_ok=True) whl_cache_path.mkdir(parents=True, exist_ok=True)
py_tag_prefix = 'cp' + python_version.replace('.', '') if python_version else None
cached_pkgs: set[tuple[str, str]] = set() cached_pkgs: set[tuple[str, str]] = set()
for whl in whl_cache_path.glob('*.whl'): for whl in whl_cache_path.glob('*.whl'):
parsed = packaging_t.parse_whl_name_version(whl.name) parsed = packaging_t.parse_whl_name_version(whl.name)
if parsed is not None: if parsed is None:
continue
if py_tag_prefix is not None and parsed.python_tag is not None:
if not parsed.python_tag.startswith(py_tag_prefix) and parsed.python_tag not in ('py3', 'py2.py3'):
continue
cached_pkgs.add((parsed.name, parsed.version)) cached_pkgs.add((parsed.name, parsed.version))
missing_reqs: list[str] = [] missing_reqs: list[str] = []
@ -546,6 +627,10 @@ def whl_cache_download(
f.flush() f.flush()
missing_req_path = f.name missing_req_path = f.name
pip_python_version_args: list[str] = []
if python_version is not None:
pip_python_version_args = ['--python-version', python_version.replace('.', '')]
try: try:
cmd = [ cmd = [
sys.executable, sys.executable,
@ -553,7 +638,7 @@ def whl_cache_download(
'pip', 'pip',
'download', 'download',
'--only-binary=:all:', '--only-binary=:all:',
*uv_python_version, *pip_python_version_args,
*pip_find_links_args, *pip_find_links_args,
'-r', '-r',
missing_req_path, missing_req_path,
@ -584,6 +669,7 @@ def check_host_prerequisites() -> None:
def env_bootstrap( def env_bootstrap(
bootstrap_settings: BootstrapSettings, bootstrap_settings: BootstrapSettings,
pyproject: PyProject, pyproject: PyProject,
overrides: Optional[list[str]] = None,
) -> None: ) -> None:
check_host_prerequisites() check_host_prerequisites()
@ -707,6 +793,19 @@ def env_bootstrap(
if len(constraint_args) > 0: if len(constraint_args) > 0:
uv_compile_args = [o for o in uv_compile_args if o not in ('-U', '--upgrade')] uv_compile_args = [o for o in uv_compile_args if o not in ('-U', '--upgrade')]
with contextlib.ExitStack() as stack:
if overrides and len(constraint_args) > 0:
patched = stack.enter_context(
tempfile.NamedTemporaryFile(
mode='w', prefix='constraints_', suffix='.txt'
)
)
packaging_t.apply_overrides_to_constraints(
requirements_path, overrides, patched
)
patched.flush()
constraint_args = ['-c', patched.name]
cmd = [ cmd = [
'uv', 'uv',
'--cache-dir', '--cache-dir',
@ -738,7 +837,7 @@ def env_bootstrap(
whl_cache_download( whl_cache_download(
whl_cache_path=bootstrap_settings.whl_cache_path, whl_cache_path=bootstrap_settings.whl_cache_path,
requirements_path=requirements_path, requirements_path=requirements_path,
uv_python_version=uv_python_version, python_version=bootstrap_settings.python_version,
pip_find_links_args=pip_find_links_args, pip_find_links_args=pip_find_links_args,
) )
if bootstrap_settings.whl_cache_path.exists(): if bootstrap_settings.whl_cache_path.exists():
@ -797,6 +896,53 @@ def paths_equal(a: pathlib.Path | str, b: pathlib.Path | str) -> bool:
return os.path.abspath(str(a)) == os.path.abspath(str(b)) return os.path.abspath(str(a)) == os.path.abspath(str(b))
import argparse as _argparse
class argv_extract_t:
"""Extract known arguments from argv by scanning parser action definitions."""
@dataclasses.dataclass
class res_t:
namespace: _argparse.Namespace
rest: list[str]
@staticmethod
def extract(
parser: _argparse.ArgumentParser,
argv: list[str],
) -> 'argv_extract_t.res_t':
flag_map: dict[str, _argparse.Action] = {}
for action in parser._actions:
for opt in action.option_strings:
flag_map[opt] = action
matched_argv: list[str] = []
rest: list[str] = []
i = 0
while i < len(argv):
action = flag_map.get(argv[i])
if action is not None:
matched_argv.append(argv[i])
i += 1
if action.nargs in (None, 1) and action.const is None and not isinstance(
action, (_argparse._StoreTrueAction, _argparse._StoreFalseAction, _argparse._CountAction, _argparse._HelpAction)
):
if i < len(argv):
matched_argv.append(argv[i])
i += 1
else:
rest.append(argv[i])
i += 1
namespace = parser.parse_args(matched_argv)
return argv_extract_t.res_t(
namespace=namespace,
rest=rest,
)
def run( def run(
d: Optional[pathlib.Path] = None, d: Optional[pathlib.Path] = None,
cli_path: Optional[pathlib.Path] = None, cli_path: Optional[pathlib.Path] = None,
@ -807,6 +953,22 @@ def run(
if d is None: if d is None:
d = pathlib.Path(__file__).parent / 'pyproject.toml' d = pathlib.Path(__file__).parent / 'pyproject.toml'
bootstrap_parser = _argparse.ArgumentParser(add_help=False)
bootstrap_parser.add_argument(
'--bootstrap-help',
action='help',
help='show bootstrap help and exit',
)
bootstrap_parser.add_argument(
'--bootstrap-override',
dest='overrides',
action='append',
default=[],
help='override for uv pip compile (e.g. "librt>=0.8")',
)
bootstrap_args = argv_extract_t.extract(bootstrap_parser, sys.argv[1:])
bootstrap_settings = BootstrapSettings.get() bootstrap_settings = BootstrapSettings.get()
pyproject: PyProject = pyproject_load(d) pyproject: PyProject = pyproject_load(d)
@ -820,6 +982,7 @@ def run(
env_bootstrap( env_bootstrap(
bootstrap_settings=bootstrap_settings, bootstrap_settings=bootstrap_settings,
pyproject=pyproject, pyproject=pyproject,
overrides=bootstrap_args.namespace.overrides or None,
) )
logger.info([sys.executable, sys.argv, bootstrap_settings.python_path]) logger.info([sys.executable, sys.argv, bootstrap_settings.python_path])
@ -829,7 +992,8 @@ def run(
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
[ [
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
*sys.argv, sys.argv[0],
*bootstrap_args.rest,
], ],
) )
@ -838,7 +1002,7 @@ def run(
[ [
str(bootstrap_settings.python_path), str(bootstrap_settings.python_path),
str(cli_path), str(cli_path),
*sys.argv[1:], *bootstrap_args.rest,
], ],
) )