Source code for hbutils.system.python.package

import functools
import itertools
import os
import pathlib
import subprocess
import sys
from typing import List, Optional

from packaging.requirements import Requirement
from packaging.utils import canonicalize_name

try:
    import importlib.metadata as importlib_metadata
except (ModuleNotFoundError, ImportError):
    import importlib_metadata
from packaging.version import Version

__all__ = [
    'package_version',
    'load_req_file', 'pip',
    'check_reqs', 'check_req_file',
    'pip_install', 'pip_install_req_file',
]


[docs]def package_version(name: str) -> Optional[Version]: """ Overview: Get version of package with given ``name``. :param name: Name of the package, case is not sensitive. :return: A :class:`packing.version.Version` object. If the package is not installed, return ``None``. Examples:: >>> from hbutils.system import package_version >>> >>> package_version('pip') <Version('21.3.1')> >>> package_version('setuptools') <Version('59.6.0')> >>> package_version('not_a_package') None """ try: return Version(importlib_metadata.distribution(canonicalize_name(name)).version) except importlib_metadata.PackageNotFoundError: return None
def _nonblank(text): return text and not text.startswith('#') @functools.singledispatch def yield_lines(iterable): r""" Based on https://github.com/jaraco/jaraco.text/blob/main/jaraco/text/__init__.py#L537 . Yield valid lines of a string or iterable. >>> list(yield_lines('')) [] >>> list(yield_lines(['foo', 'bar'])) ['foo', 'bar'] >>> list(yield_lines('foo\nbar')) ['foo', 'bar'] >>> list(yield_lines('\nfoo\n#bar\nbaz #comment')) ['foo', 'baz #comment'] >>> list(yield_lines(['foo\nbar', 'baz', 'bing\n\n\n'])) ['foo', 'bar', 'baz', 'bing'] """ return itertools.chain.from_iterable(map(yield_lines, iterable)) @yield_lines.register(str) def _(text): return filter(_nonblank, map(str.strip, text.splitlines())) def drop_comment(line): """ Based on https://github.com/jaraco/jaraco.text/blob/main/jaraco/text/__init__.py#L560 . Drop comments. >>> drop_comment('foo # bar') 'foo' A hash without a space may be in a URL. >>> drop_comment('https://example.com/foo#bar') 'https://example.com/foo#bar' """ return line.partition(' #')[0] def join_continuation(lines): r""" Based on https://github.com/jaraco/jaraco.text/blob/main/jaraco/text/__init__.py#L575 . Join lines continued by a trailing backslash. >>> list(join_continuation(['foo \\', 'bar', 'baz'])) ['foobar', 'baz'] >>> list(join_continuation(['foo \\', 'bar', 'baz'])) ['foobar', 'baz'] >>> list(join_continuation(['foo \\', 'bar \\', 'baz'])) ['foobarbaz'] Not sure why, but... The character preceding the backslash is also elided. >>> list(join_continuation(['goo\\', 'dly'])) ['godly'] A terrible idea, but... If no line is available to continue, suppress the lines. >>> list(join_continuation(['foo', 'bar\\', 'baz\\'])) ['foo'] """ lines = iter(lines) for item in lines: while item.endswith('\\'): try: # pragma: no cover item = item[:-2].strip() + next(lines) except StopIteration: return yield item
[docs]def load_req_file(requirements_file: str) -> List[str]: """ Overview: Load requirements items from a ``requirements.txt`` file. :param requirements_file: Requirements file. :return requirements: List of requirements. Examples:: >>> from hbutils.system import load_req_file >>> load_req_file('requirements.txt') ['packaging>=21.3', 'setuptools>=50.0'] """ with pathlib.Path(requirements_file).open() as reqfile: return list(map( lambda x: str(Requirement(x)), join_continuation(map(drop_comment, yield_lines(reqfile))) ))
[docs]def pip(*args, silent: bool = False): """ Overview: Run pip command with code. :param args: Command line arguments for ``pip`` command. :param silent: Do not print anything. Default is false, which means print the output to ``sys.stdout`` \ and ``sys.stderr``. Examples:: >>> from hbutils.system import pip >>> pip('-V') pip 22.3.1 from /home/user/myproject/venv/lib/python3.7/site-packages/pip (python 3.7) >>> pip('-V', silent=True) # nothing will be printed """ process = subprocess.run( [sys.executable, '-m', 'pip', *args], stdin=sys.stdin if not silent else None, stdout=sys.stdout if not silent else subprocess.PIPE, stderr=sys.stderr if not silent else subprocess.PIPE, ) assert not process.returncode, f'Error when calling {process.args!r}{os.linesep}' \ f'Error Code - {process.returncode}{os.linesep}' \ f'Stdout:{os.linesep}' \ f'{process.stdout.decode()}{os.linesep}' \ f'{os.linesep}' \ f'Stderr:{os.linesep}' \ f'{process.stderr.decode()}{os.linesep}' process.check_returncode()
def _yield_reqs_to_install(req: Requirement, current_extra: str = ''): if req.marker and not req.marker.evaluate({'extra': current_extra}): return try: version = importlib_metadata.distribution(req.name).version except importlib_metadata.PackageNotFoundError: # req not installed yield req else: if req.specifier.contains(version): for child_req in (importlib_metadata.metadata(req.name).get_all('Requires-Dist') or []): child_req_obj = Requirement(child_req) need_check, ext = False, None for extra in req.extras: if child_req_obj.marker and child_req_obj.marker.evaluate({'extra': extra}): need_check = True ext = extra break if need_check: # check for extra reqs yield from _yield_reqs_to_install(child_req_obj, ext) else: # main version not match yield req def _check_req(req: Requirement): return not bool(list(itertools.islice(_yield_reqs_to_install(req), 1)))
[docs]def check_reqs(reqs: List[str]) -> bool: """ Overview: Check if the given requirements are all satisfied. :param reqs: List of requirements. :return satisfied: All the requirements in ``reqs`` satisfied or not. Examples:: >>> from hbutils.system import check_reqs >>> check_reqs(['pip>=20.0']) True >>> check_reqs(['pip~=19.2']) False >>> check_reqs(['pip>=20.0', 'setuptools>=50.0']) True .. note:: If a requirement's marker is not satisfied in this environment, **it will be ignored** instead of return ``False``. """ return all(map(lambda x: _check_req(Requirement(x)), reqs))
[docs]def check_req_file(requirements_file: str) -> bool: """ Overview: Check if the requirements in the given ``requirements_file`` is satisfied. :param requirements_file: Requirements file, such as ``requirements.txt``. :return satisfied: All the requirements in ``requirements_file`` satisfied or not. Examples:: >>> from hbutils.system import check_req_file >>> >>> check_req_file('requirements.txt') True >>> check_req_file('requirements-test.txt') True """ return check_reqs(load_req_file(requirements_file))
[docs]def pip_install(reqs: List[str], silent: bool = False, force: bool = False, user: bool = False): """ Overview: Pip install requirements with code. Similar to ``pip install req1 req2 ...``. :param reqs: Requirement items to install. :param silent: Do not print anything. Default is ``False``. :param force: Force execute the ``pip install`` command. Default is ``False`` which means the requirements \ will be checked before installation, and the installation will be only executed when \ some requirements not installed. :param user: User mode, represents ``--user`` option in ``pip``. Examples:: >>> from hbutils.system import pip_install >>> pip_install(['scikit-learn']) # not installed Looking in indexes: https://xxx/simple Collecting scikit-learn Using cached https://xxx/scikit_learn-1.0.2-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (24.8 MB) Installing collected packages: threadpoolctl, scipy, joblib, scikit-learn Successfully installed joblib-1.2.0 scikit-learn-1.0.2 scipy-1.7.3 threadpoolctl-3.1.0 >>> pip_install(['numpy>=1.10.0']) # installed >>> pip_install(['numpy>=1.10.0'], force=True) # force execute Looking in indexes: https://xxx/simple Requirement already satisfied: numpy>=1.10.0 in ./venv/lib/python3.7/site-packages (1.21.6) """ if force or not check_reqs(reqs): pip('install', *(('--user',) if user else ()), *reqs, silent=silent)
[docs]def pip_install_req_file(requirements_file: str, silent: bool = False, force: bool = False, user: bool = False): """ Overview: Pip install requirements from file with code. Similar to ``pip install -r requirements.txt``. :param requirements_file: Requirements file, such as ``requirements.txt``. :param silent: Do not print anything. Default is ``False``. :param force: Force execute the ``pip install`` command. Default is ``False`` which means the requirements \ will be checked before installation, and the installation will be only executed when \ some requirements not installed. :param user: User mode, represents ``--user`` option in ``pip``. Examples:: >>> from hbutils.system import pip_install_req_file >>> pip_install_req_file('requirements.txt') # pip install -r requirements.txt """ if force or not check_req_file(requirements_file): pip('install', *(('--user',) if user else ()), '-r', requirements_file, silent=silent)