Implement Python management

Author: ofek

docs/config/hatch.md

@@ -205,14 +205,14 @@ Any type of environment that is not explicitly defined will default to `<DATA_DI
     python = "..."
     ```
 
-This determines where to install specific versions of Python, with the full path being `<VALUE>/pythons`.
+This determines where to install specific versions of Python.
 
-The following values have special meanings.
+The following values have special meanings:
 
 | Value | Path |
 | --- | --- |
-| `isolated` (default) | `<DATA_DIR>/pythons` |
-| `shared` | `~/.pythons` |
+| `shared` (default) | `~/.pythons` |
+| `isolated` | `<DATA_DIR>/pythons` |
 
 ## Terminal
 

hatch.toml

@@ -123,9 +123,20 @@ HATCH_BUILD_CLEAN = "true"
 build = "python -m build backend"
 publish = "hatch publish backend/dist"
 version = "cd backend && hatch version {args}"
-update-data = [
+
+[envs.upkeep]
+detached = true
+dependencies = [
+  "httpx",
+]
+[envs.upkeep.scripts]
+update-hatch = [
+  "update-distributions",
+]
+update-hatchling = [
   "update-licenses",
 ]
+update-distributions = "python scripts/update_distributions.py"
 update-licenses = "python backend/scripts/update_licenses.py"
 
 [envs.release]

pyproject.toml

@@ -51,6 +51,7 @@ dependencies = [
   "tomlkit>=0.11.1",
   "userpath~=1.7",
   "virtualenv>=20.16.2",
+  "zstandard<1",
 ]
 dynamic = ["version"]
 

scripts/update_distributions.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import re
+from ast import literal_eval
+from collections import defaultdict
+
+import httpx
+from utils import ROOT
+
+URL = 'https://raw.githubusercontent.com/ofek/pyapp/master/build.rs'
+OUTPUT_FILE = ROOT / 'src' / 'hatch' / 'python' / 'distributions.py'
+ARCHES = {('linux', 'x86'): 'i686', ('windows', 'x86_64'): 'amd64', ('windows', 'x86'): 'i386'}
+
+# system, architecture, ABI, variant
+MAX_IDENTIFIER_COMPONENTS = 4
+
+
+def parse_distributions(contents: str, constant: str):
+    match = re.search(f'^const {constant}.+?^];$', contents, flags=re.DOTALL | re.MULTILINE)
+    if not match:
+        message = f'Could not find {constant} in {URL}'
+        raise ValueError(message)
+
+    block = match.group(0).replace('",\n', '",')
+    for line in block.splitlines()[1:-1]:
+        line = line.strip()
+        if not line or line.startswith('//'):
+            continue
+
+        identifier, *data, source = literal_eval(line[:-1])
+        os, arch = data[:2]
+        if arch == 'powerpc64':
+            arch = 'ppc64le'
+
+        # Force everything to have a variant to maintain structure
+        if len(data) != MAX_IDENTIFIER_COMPONENTS:
+            data.append('')
+
+        data[1] = ARCHES.get((os, arch), arch)
+        yield identifier, tuple(data), source
+
+
+def main():
+    response = httpx.get(URL)
+    response.raise_for_status()
+
+    contents = response.text
+    distributions = defaultdict(list)
+    ordering_data = defaultdict(dict)
+
+    for i, distribution_type in enumerate(('DEFAULT_CPYTHON_DISTRIBUTIONS', 'DEFAULT_PYPY_DISTRIBUTIONS')):
+        for identifier, data, source in parse_distributions(contents, distribution_type):
+            ordering_data[i][identifier] = None
+            distributions[identifier].append((data, source))
+
+    ordered = [identifier for identifiers in ordering_data.values() for identifier in reversed(identifiers)]
+    output = [
+        'from __future__ import annotations',
+        '',
+        '# fmt: off',
+        'ORDERED_DISTRIBUTIONS: tuple[str, ...] = (',
+    ]
+    for identifier in ordered:
+        output.append(f'    {identifier!r},')
+    output.append(')')
+
+    output.append('DISTRIBUTIONS: dict[str, dict[tuple[str, ...], str]] = {')
+    for identifier, data in distributions.items():
+        output.append(f'    {identifier!r}: {{')
+
+        for d, source in data:
+            output.append(f'        {d!r}:')
+            output.append(f'            {source!r},')
+
+        output.append('    },')
+
+    output.append('}')
+    output.append('')
+    output = '\n'.join(output)
+
+    with open(OUTPUT_FILE, 'w') as f:
+        f.write(output)
+
+
+if __name__ == '__main__':
+    main()

src/hatch/cli/__init__.py

@@ -12,6 +12,7 @@
 from hatch.cli.new import new
 from hatch.cli.project import project
 from hatch.cli.publish import publish
+from hatch.cli.python import python
 from hatch.cli.run import run
 from hatch.cli.shell import shell
 from hatch.cli.status import status
@@ -22,7 +23,9 @@
 from hatch.utils.fs import Path
 
 
-@click.group(context_settings={'help_option_names': ['-h', '--help']}, invoke_without_command=True)
+@click.group(
+    context_settings={'help_option_names': ['-h', '--help'], 'max_content_width': 120}, invoke_without_command=True
+)
 @click.option(
     '--env',
     '-e',
@@ -37,20 +40,6 @@
     envvar=ConfigEnvVars.PROJECT,
     help='The name of the project to work on [env var: `HATCH_PROJECT`]',
 )
-@click.option(
-    '--color/--no-color',
-    default=None,
-    help='Whether or not to display colored output (default is auto-detection) [env vars: `FORCE_COLOR`/`NO_COLOR`]',
-)
-@click.option(
-    '--interactive/--no-interactive',
-    envvar=AppEnvVars.INTERACTIVE,
-    default=None,
-    help=(
-        'Whether or not to allow features like prompts and progress bars (default is auto-detection) '
-        '[env var: `HATCH_INTERACTIVE`]'
-    ),
-)
 @click.option(
     '--verbose',
     '-v',
@@ -65,6 +54,20 @@
     count=True,
     help='Decrease verbosity (can be used additively) [env var: `HATCH_QUIET`]',
 )
+@click.option(
+    '--color/--no-color',
+    default=None,
+    help='Whether or not to display colored output (default is auto-detection) [env vars: `FORCE_COLOR`/`NO_COLOR`]',
+)
+@click.option(
+    '--interactive/--no-interactive',
+    envvar=AppEnvVars.INTERACTIVE,
+    default=None,
+    help=(
+        'Whether or not to allow features like prompts and progress bars (default is auto-detection) '
+        '[env var: `HATCH_INTERACTIVE`]'
+    ),
+)
 @click.option(
     '--data-dir',
     envvar=ConfigEnvVars.DATA,
@@ -83,7 +86,7 @@
 )
 @click.version_option(version=__version__, prog_name='Hatch')
 @click.pass_context
-def hatch(ctx: click.Context, env_name, project, color, interactive, verbose, quiet, data_dir, cache_dir, config_file):
+def hatch(ctx: click.Context, env_name, project, verbose, quiet, color, interactive, data_dir, cache_dir, config_file):
     """
     \b
      _   _       _       _
@@ -194,6 +197,7 @@ def hatch(ctx: click.Context, env_name, project, color, interactive, verbose, qu
 hatch.add_command(new)
 hatch.add_command(project)
 hatch.add_command(publish)
+hatch.add_command(python)
 hatch.add_command(run)
 hatch.add_command(shell)
 hatch.add_command(status)

src/hatch/cli/application.py

@@ -2,6 +2,7 @@
 
 import os
 import sys
+from functools import cached_property
 from typing import TYPE_CHECKING, cast
 
 from hatch.cli.terminal import Terminal
@@ -214,6 +215,27 @@ def get_env_directory(self, environment_type):
         else:
             return self.data_dir / 'env' / environment_type
 
+    def get_python_manager(self, directory: str | None = None):
+        from hatch.python.core import PythonManager
+
+        configured_dir = directory or self.config.dirs.python
+        if configured_dir == 'shared':
+            return PythonManager(Path.home() / '.pythons')
+        elif configured_dir == 'isolated':
+            return PythonManager(self.data_dir / 'pythons')
+        else:
+            return PythonManager(Path(configured_dir).expand())
+
+    @cached_property
+    def shell_data(self) -> tuple[str, str]:
+        import shellingham
+
+        try:
+            return shellingham.detect_shell()
+        except shellingham.ShellDetectionFailure:
+            path = self.platform.default_shell
+            return Path(path).stem, path
+
     def abort(self, text='', code=1, **kwargs):
         if text:
             self.display_error(text, **kwargs)

src/hatch/cli/python/__init__.py

@@ -0,0 +1,17 @@
+import click
+
+from hatch.cli.python.install import install
+from hatch.cli.python.remove import remove
+from hatch.cli.python.show import show
+from hatch.cli.python.update import update
+
+
+@click.group(short_help='Manage Python installations')
+def python():
+    pass
+
+
+python.add_command(install)
+python.add_command(remove)
+python.add_command(show)
+python.add_command(update)

src/hatch/cli/python/install.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from hatch.cli.application import Application
+
+
+def ensure_path_public(path: str, shells: list[str]) -> None:
+    import userpath
+
+    if userpath.in_current_path(path) or userpath.in_new_path(path, shells):
+        return
+
+    userpath.append(path, shells=shells)
+
+
+@click.command(short_help='Install Python distributions')
+@click.argument('names', required=True, nargs=-1)
+@click.option('--private', is_flag=True, help='Do not add distributions to the user PATH')
+@click.option('--update', '-u', is_flag=True, help='Update existing installations')
+@click.option(
+    '--dir', '-d', 'directory', help='The directory in which to install distributions, overriding configuration'
+)
+@click.pass_obj
+def install(app: Application, *, names: tuple[str, ...], private: bool, update: bool, directory: str | None):
+    """
+    Install Python distributions.
+
+    You may select `all` to install all compatible distributions:
+
+    \b
+    ```
+    hatch python install all
+    ```
+    """
+    from hatch.errors import PythonDistributionResolutionError, PythonDistributionUnknownError
+    from hatch.python.distributions import ORDERED_DISTRIBUTIONS
+    from hatch.python.resolve import get_distribution
+
+    shells = []
+    if not private and not app.platform.windows:
+        shell_name, _ = app.shell_data
+        shells.append(shell_name)
+
+    manager = app.get_python_manager(directory)
+    installed = manager.get_installed()
+    selection = ORDERED_DISTRIBUTIONS if 'all' in names else names
+    unknown = []
+    compatible = []
+    incompatible = []
+    for name in selection:
+        if name in installed:
+            compatible.append(name)
+            continue
+
+        try:
+            get_distribution(name)
+        except PythonDistributionUnknownError:
+            unknown.append(name)
+        except PythonDistributionResolutionError:
+            incompatible.append(name)
+        else:
+            compatible.append(name)
+
+    if unknown:
+        app.abort(f'Unknown distributions: {", ".join(unknown)}')
+    elif incompatible and (not compatible or 'all' not in names):
+        app.abort(f'Incompatible distributions: {", ".join(incompatible)}')
+
+    for name in compatible:
+        needs_update = False
+        if name in installed:
+            needs_update = installed[name].needs_update()
+            if not needs_update:
+                app.display_warning(f'The latest version is already installed: {name}')
+                continue
+            elif not (update or app.confirm(f'Update {name}?')):
+                app.abort(f'Distribution is already installed: {name}')
+
+        with app.status(f'{"Updating" if needs_update else "Installing"} {name}'):
+            dist = manager.install(name)
+            if not private:
+                ensure_path_public(str(dist.python_path.parent), shells=shells)
+
+        app.display_success(f'{"Updated" if needs_update else "Installed"} {name} @ {dist.path}')

src/hatch/cli/python/remove.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from hatch.cli.application import Application
+
+
+@click.command(short_help='Remove Python distributions')
+@click.argument('names', required=True, nargs=-1)
+@click.option('--dir', '-d', 'directory', help='The directory in which distributions reside')
+@click.pass_obj
+def remove(app: Application, *, names: tuple[str, ...], directory: str | None):
+    """
+    Remove Python distributions.
+
+    You may select `all` to remove all installed distributions:
+
+    \b
+    ```
+    hatch python install all
+    ```
+    """
+    manager = app.get_python_manager(directory)
+    installed = manager.get_installed()
+    selection = tuple(installed) if 'all' in names else names
+    for name in selection:
+        if name not in installed:
+            app.display_warning(f'Distribution is not installed: {name}')
+            continue
+
+        dist = installed[name]
+        with app.status(f'Removing {name}'):
+            manager.remove(dist)