diff --git a/.github/workflows/typing.yml b/.github/workflows/typing.yml new file mode 100644 index 0000000..263e462 --- /dev/null +++ b/.github/workflows/typing.yml @@ -0,0 +1,30 @@ +name: Typing + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_call: + +env: + UV_SYSTEM_PYTHON: 1 + +jobs: + type-check: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: 3.12 + - name: Install dependencies + run: | + pip install -U pip + pip install . + pip install mypy + + - name: Run mypy + run: | + mypy komodoenv diff --git a/komodoenv/__main__.py b/komodoenv/__main__.py index 8b8d721..20d9ab6 100644 --- a/komodoenv/__main__.py +++ b/komodoenv/__main__.py @@ -15,7 +15,21 @@ from komodoenv.statfs import is_nfs -def get_release_maturity_text(release_path): +class KomodoenvNamespace(argparse.Namespace): + """ + Komodoenv argument parser namespace + """ + + force: bool + release: str + track: str | None + no_update: bool + root: str + force_color: bool + destination: str + + +def get_release_maturity_text(release_path: Path) -> str: """Returns a comment informing the user about the maturity of the release that they've chosen. Eg, warn users if they want bleeding, pat them on the back if they want stable, etc. @@ -40,7 +54,7 @@ def get_release_maturity_text(release_path): ) -def distro_suffix(): +def distro_suffix() -> str: # Workaround to make tests pass on Github Actions if ( "GITHUB_ACTIONS" in os.environ @@ -56,12 +70,10 @@ def distro_suffix(): def resolve_release( # noqa: C901 *, root: Path, - name: str, + release_path: Path, no_update: bool = False, ) -> tuple[Path, Path]: """Autodetect komodo release heuristically""" - if not (root / name / "enable").is_file(): - sys.exit(f"'{root / name}' is not a valid komodo release") env = os.environ.copy() if "BASH_ENV" in env: @@ -71,7 +83,7 @@ def resolve_release( # noqa: C901 [ "/bin/bash", "-c", - f"source {root / name / 'enable'};which python;python --version", + f"source {release_path / 'enable'};which python;python --version", ], env=env, ) @@ -92,7 +104,7 @@ def resolve_release( # noqa: C901 major, minor = match.groups() pyver = "-py" + major + minor - base_name = re.match("^(.*?)(?:-py[0-9]+)?(?:-rhel[0-9]+)?$", name) + base_name = re.match("^(.*?)(?:-py[0-9]+)?(?:-rhel[0-9]+)?$", release_path.name) if base_name is None: msg = "Could not find the release." raise ValueError(msg) @@ -120,7 +132,7 @@ def resolve_release( # noqa: C901 ) -def parse_args(args): +def parse_args(args: list[str]) -> KomodoenvNamespace: ap = argparse.ArgumentParser() ap.add_argument( "-f", @@ -163,42 +175,10 @@ def parse_args(args): ) ap.add_argument("destination", type=str, help="Where to create komodoenv") - args = ap.parse_args(args) - - args.root = Path(args.root) - if not args.root.is_dir(): - msg = "The given root is not a directory." - raise ValueError(msg) - - if "/" in args.release: - args.release = Path(args.release) - elif isinstance(args.release, str): - args.release = Path(args.root) / args.release - - if not args.release or not args.track: - args.release, args.track = resolve_release( - root=args.root, - name=str(args.release), - no_update=args.no_update, - ) - args.track = Path(args.track) - args.destination = Path(args.destination).absolute() - - if args.release is None or not args.release.is_dir(): - print( - "Could not automatically detect active Komodo release. " - "Either enable a Komodo release that supports komodoenv " - "or specify release manually with the " - "`--release' argument. ", - sys.stderr, - ) - ap.print_help() - sys.exit(1) - - return args + return ap.parse_args(args, namespace=KomodoenvNamespace()) -def main(args=None): +def main() -> None: texts = { "info": blue( "Info: " @@ -212,17 +192,45 @@ def main(args=None): ), } - if args is None: - args = sys.argv[1:] - args = parse_args(args) + args = parse_args(sys.argv[1:]) + + root = Path(args.root).resolve() + if not root.is_dir(): + msg = f"The given root '{args.root}' is not a directory." + raise ValueError(msg) + + release = root / args.release + + if not (release / "enable").is_file(): + sys.exit(f"'{root / args.release}' is not a valid komodo release!") + + if not args.track: + release, track = resolve_release( + root=root, + release_path=release, + no_update=args.no_update, + ) + else: + track = Path(args.track).resolve() + destination = Path(args.destination).absolute() + + if not release.is_dir(): + print( + "Could not automatically detect active Komodo release. " + "Either enable a Komodo release that supports komodoenv " + "or specify release manually with the " + "`--release' argument. ", + sys.stderr, + ) + sys.exit(1) - if args.destination.is_dir() and args.force: - rmtree(str(args.destination), ignore_errors=True) - elif args.destination.is_dir(): - sys.exit(f"Destination directory already exists: {args.destination}") + if destination.is_dir() and args.force: + rmtree(str(destination), ignore_errors=True) + elif destination.is_dir(): + sys.exit(f"Destination directory already exists: {destination !s}") use_color = args.force_color or (sys.stdout.isatty() and sys.stderr.isatty()) - release_text = get_release_maturity_text(args.track) + release_text = get_release_maturity_text(track) if not use_color: texts = {key: strip_color(val) for key, val in texts.items()} release_text = strip_color(release_text) @@ -230,14 +238,14 @@ def main(args=None): print(texts["info"], file=sys.stderr) print(release_text, file=sys.stderr) - if not is_nfs(args.destination): + if not is_nfs(destination): print(texts["nfs"], file=sys.stderr) creator = Creator( - komodo_root=args.root, - srcpath=args.release, - trackpath=args.track, - dstpath=args.destination, + komodo_root=root, + src_path=release, + track_path=track, + dst_path=destination, use_color=use_color, ) creator.create() diff --git a/komodoenv/creator.py b/komodoenv/creator.py index 6bc753e..e64cd39 100644 --- a/komodoenv/creator.py +++ b/komodoenv/creator.py @@ -1,9 +1,12 @@ +from __future__ import annotations + import os import subprocess -from contextlib import contextmanager +from contextlib import _GeneratorContextManager, contextmanager from importlib.metadata import distribution from pathlib import Path from textwrap import dedent +from typing import IO, Any, Generator import distro @@ -13,7 +16,9 @@ @contextmanager -def open_chmod(path: Path, mode: str = "w", file_mode=0o644): +def open_chmod( + path: Path, mode: str = "w", file_mode: int = 0o644 +) -> Generator[IO[Any], Any, None]: with open(path, mode, encoding="utf-8") as file: yield file path.chmod(file_mode) @@ -25,47 +30,45 @@ class Creator: def __init__( self, *, - komodo_root, - srcpath, - trackpath, - dstpath=None, - use_color=False, - ): + komodo_root: Path, + src_path: Path, + track_path: Path, + dst_path: Path, + use_color: bool = False, + ) -> None: if not use_color: self._fmt_action = strip_color(self._fmt_action) self.komodo_root = komodo_root - self.srcpath = srcpath - self.trackpath = trackpath - self.dstpath = dstpath + self.src_path = src_path + self.track_path = track_path + self.dst_path = dst_path - self.srcpy = Python(srcpath / "root/bin/python") + self.srcpy = Python(src_path / "root/bin/python") self.srcpy.detect() - self.dstpy = self.srcpy.make_dst(dstpath / "root/bin/python") + self.dstpy = self.srcpy.make_dst(dst_path / "root/bin/python") - def print_action(self, action, message): + def print_action(self, action: str, message: str) -> None: print(self._fmt_action.format(action=action, message=message)) - def mkdir(self, path): - self.print_action("mkdir", path + "/") - (self.dstpath / path).mkdir() - - def create_file(self, path, file_mode=0o644): - self.print_action("create", path) - return open_chmod(self.dstpath / path, file_mode=file_mode) + def create_file( + self, path: Path | str, file_mode: int = 0o644 + ) -> _GeneratorContextManager[IO[Any]]: + self.print_action("create", str(path)) + return open_chmod(self.dst_path / path, file_mode=file_mode) - def remove_file(self, path): - if not (self.dstpath / path).is_file(): + def remove_file(self, path: Path) -> None: + if not (self.dst_path / path).is_file(): return - self.print_action("remove", path) - (self.dstpath / path).unlink() + self.print_action("remove", str(path)) + (self.dst_path / path).unlink() - def venv(self): + def venv(self) -> None: self.print_action("venv", f"using {self.srcpy.executable}") - env = {"LD_LIBRARY_PATH": str(self.srcpath / "root" / "lib"), **os.environ} + env = {"LD_LIBRARY_PATH": str(self.src_path / "root" / "lib"), **os.environ} subprocess.check_output( [ str(self.srcpy.executable) @@ -76,14 +79,14 @@ def venv(self): "venv", "--copies", "--without-pip", - str(self.dstpath / "root"), + str(self.dst_path / "root"), ], env=env, ) - def run(self, path): - self.print_action("run", path) - subprocess.check_output([str(self.dstpath / path)]) + def run(self, path: Path) -> None: + self.print_action("run", str(path)) + subprocess.check_output([str(self.dst_path / path)]) def pip_install(self, package: str) -> None: pip_wheel = get_bundled_wheel("pip") @@ -91,11 +94,11 @@ def pip_install(self, package: str) -> None: self.print_action("install", package) env = os.environ.copy() - env["PYTHONPATH"] = pip_wheel + env["PYTHONPATH"] = str(pip_wheel) subprocess.check_output( [ - str(self.dstpath / "root/bin/python"), + str(self.dst_path / "root/bin/python"), "-m", "pip", "install", @@ -108,8 +111,8 @@ def pip_install(self, package: str) -> None: env=env, ) - def create(self): - self.dstpath.mkdir() + def create(self) -> None: + self.dst_path.mkdir() self.venv() @@ -118,8 +121,8 @@ def create(self): f.write( dedent( f"""\ - current-release = {self.srcpath.name} - tracked-release = {self.trackpath.name} + current-release = {self.src_path.name} + tracked-release = {self.track_path.name} mtime-release = 0 python-version = {self.srcpy.version_info[0]}.{self.srcpy.version_info[1]} komodoenv-version = {distribution('komodoenv').version} @@ -130,7 +133,7 @@ def create(self): ) python_paths = [ - pth for pth in self.srcpy.site_paths if pth.startswith(str(self.srcpath)) + pth for pth in self.srcpy.site_paths if pth.startswith(str(self.src_path)) ] # We use zzz_komodo.pth to try and make it the last .pth file to be processed @@ -150,15 +153,15 @@ def create(self): file_mode=0o755, ) as outf: outf.write(inf.read()) - self.run("root/bin/komodoenv-update") + self.run(Path("root/bin/komodoenv-update")) self.pip_install("pip") - self.remove_file("root/shims/komodoenv") + self.remove_file(Path("root/shims/komodoenv")) if os.environ.get("SHELL", "").endswith("csh"): - enable_script = self.dstpath / "enable.csh" + enable_script = self.dst_path / "enable.csh" else: - enable_script = self.dstpath / "enable" + enable_script = self.dst_path / "enable" print( dedent( diff --git a/komodoenv/python.py b/komodoenv/python.py index 3eb0c5e..a90124a 100644 --- a/komodoenv/python.py +++ b/komodoenv/python.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import json import os from pathlib import Path @@ -5,55 +7,47 @@ class Python: - def __init__(self, executable, komodo_prefix=None): + def __init__(self, executable: Path) -> None: """""" self.executable = Path(executable) - if komodo_prefix is not None: - self.komodo_prefix = komodo_prefix - else: - # Assume komodo's root is one directory up from executable - self.komodo_prefix = self.executable.parent.parent + # Assume komodo's root is one directory up from executable + # meaning, # root/bin/python + self.release_root = self.executable.parents[1] - self.root = self.executable.parent.parent - self.version_info = None + self.version_info: tuple[str, str] + self.site_paths: list[str] - def make_dst(self, executable): - py = Python(executable, self.komodo_prefix) + def make_dst(self, executable: Path) -> Python: + py = Python(executable) py.version_info = self.version_info return py - def detect(self): + def detect(self) -> None: """Detects what type of Python installation this is""" # Get python version_info - script = b"import sys,json;print(json.dumps(sys.version_info[:]))" - env = { - "LD_LIBRARY_PATH": f"{self.komodo_prefix}/lib64:{self.komodo_prefix}/lib", - } - self.version_info = tuple(json.loads(self.call(script=script, env=env))) + script = b"import sys,json;print(json.dumps(sys.version_info[0:2]))" + self.version_info = tuple(json.loads(self.call(script=script))) # Get python sys.path script = b"import sys,json;print(json.dumps(sys.path))" - self.site_paths = json.loads(self.call(script=script, env=env)) + self.site_paths = json.loads(self.call(script=script)) @property - def site_packages_path(self): - return self.root / "lib/python{}.{}/site-packages".format(*self.version_info) - - def call(self, args=None, env=None, script=None): - if args is None: - args = [] - if env is None: - env = {} - if script is not None: - # Prepend '-' to tell Python to read from stdin - args = ["-", *args] - + def site_packages_path(self) -> Path: + return ( + self.release_root + / f"lib/python{self.version_info[0]}.{self.version_info[1]}/site-packages" + ) + + def call(self, script: bytes) -> bytes: + env = {} env["PATH"] = f'{self.executable.parent.absolute()}:{os.environ["PATH"]}' - env["LD_LIBRARY_PATH"] = f"{self.komodo_prefix}/lib64:{self.komodo_prefix}/lib" + env["LD_LIBRARY_PATH"] = f"{self.release_root}/lib64:{self.release_root}/lib" - args = [self.executable, *args] - proc = Popen(map(str, args), stdin=PIPE, stdout=PIPE, env=env) + # Prepend '-' to tell Python to read from stdin + args = [str(self.executable), "-"] + proc = Popen(args, stdin=PIPE, stdout=PIPE, env=env) stdout, _ = proc.communicate(script) return stdout diff --git a/komodoenv/statfs.py b/komodoenv/statfs.py index e47d922..8c569e9 100644 --- a/komodoenv/statfs.py +++ b/komodoenv/statfs.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import sys from ctypes import ( CDLL, @@ -36,9 +38,10 @@ class _Statfs(Structure): ) -def _test_fs_type(path, f_type): +def _test_fs_type(path: Path, f_type: int) -> bool: if sys.platform != "linux": - return None + msg = "Komodoenv is only compatible with linux." + raise OSError(msg) path = Path(path) while not path.is_dir(): @@ -52,11 +55,11 @@ def _test_fs_type(path, f_type): return stat.f_type == f_type -def is_tmpfs(path): +def is_tmpfs(path: Path) -> bool: """Test if `path` is on a `tmpfs` filesystem.""" return _test_fs_type(path, _TMPFS_MAGIC) -def is_nfs(path): +def is_nfs(path: Path) -> bool: """Test if `path` is on a `nfs` filesystem.""" return _test_fs_type(path, _NFS_SUPER_MAGIC) diff --git a/komodoenv/update.py b/komodoenv/update.py index d8b3033..60e3d5d 100644 --- a/komodoenv/update.py +++ b/komodoenv/update.py @@ -14,14 +14,14 @@ import shutil import subprocess import sys -from argparse import ArgumentParser +from argparse import ArgumentParser, Namespace from pathlib import Path from textwrap import dedent from typing import Dict, List, Optional, Tuple try: from distro import id as distro_id - from distro import version_parts as distro_versions + from distro import version_parts as distro_versions # type: ignore[misc] except ImportError: # The 'distro' package isn't installed. # @@ -30,18 +30,26 @@ def distro_id() -> str: if ".el7" in platform.release(): - def distro_versions() -> Tuple[str, str, str]: + def distro_versions() -> Tuple[str, str, str]: # type: ignore[misc] return ("7", "0", "0") elif ".el8" in platform.release(): - def distro_versions() -> Tuple[str, str, str]: + def distro_versions() -> Tuple[str, str, str]: # type: ignore[misc] return ("8", "0", "0") else: sys.stderr.write("Warning: komodoenv is only compatible with RHEL7 or RHEL8") +class KomodoenvUpdateNamespace(Namespace): + """ + Komodoenv update argument parser namespace + """ + + check: bool + + ENABLE_BASH = """\ disable_komodo () {{ if [[ -v _PRE_KOMODO_PATH ]]; then @@ -313,7 +321,7 @@ def can_update(config: Dict[str, str]) -> bool: return current_maj == updated_maj -def write_config(config: Dict[str, str]): +def write_config(config: Dict[str, str]) -> None: with open(Path(__file__).parents[2] / "komodoenv.conf", "w", encoding="utf-8") as f: for key, val in config.items(): f.write(f"{key} = {val}\n") @@ -440,10 +448,7 @@ def create_pth(config: Dict[str, str], srcpath: Path, dstpath: Path) -> None: ) -def parse_args(args: List[str]): - if args is None: - args = sys.argv[1:] - +def parse_args(args: List[str]) -> KomodoenvUpdateNamespace: ap = ArgumentParser() ap.add_argument( "--check", @@ -452,11 +457,11 @@ def parse_args(args: List[str]): help="Check if this komodoenv can be updated", ) - return ap.parse_args(args) + return ap.parse_args(args, namespace=KomodoenvUpdateNamespace()) -def main(args: Optional[List[str]] = None) -> None: - args = parse_args(args) +def main() -> None: + args = parse_args(sys.argv[1:]) config = read_config() if not check_same_distro(config): diff --git a/pyproject.toml b/pyproject.toml index d930948..7bfd80e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,6 @@ ignore = [ "EXE001", # Shebang is present but file is not executable "RET505", # superfluous-else-return "PERF203", # `try`-`except` within a loop incurs performance overhead - "ANN", # flake8-annotations "ISC001", # single-line-implicit-string-concatenation (conflict ruff formatter) "COM812", # missing-trailing-comma (conflict ruff formatter) ] @@ -30,4 +29,8 @@ ignore = [ "komodoenv/update.py" = ["FA102", "FA100"] [tool.ruff.lint.pylint] -max-args = 10 \ No newline at end of file +max-args = 10 + +[tool.mypy] +show_error_code_links = true +show_error_context = true \ No newline at end of file