diff --git a/python/online/fxreader/pr34/commands.py b/python/online/fxreader/pr34/commands.py index 2f7c095..38661d1 100644 --- a/python/online/fxreader/pr34/commands.py +++ b/python/online/fxreader/pr34/commands.py @@ -3881,16 +3881,23 @@ class Command(enum.StrEnum): pip_resolve = 'pip_resolve' def pip_resolve(args: list[str]) -> None: + from online.fxreader.pr34.commands_typed.pip import pip_resolve, pip_resolve_t + parser = argparse.ArgumentParser() parser.add_argument( '-m', '--mode', - choices=['copy_paste', 'monkey_patch'], + choices=[ + o.value + for o in pip_resolve_t.kwargs_t.mode_t + ], required=True, ) options, argv = parser.parse_known_args(args) - from online.fxreader.pr34.commands_typed.pip import pip_resolve + options.mode = pip_resolve_t.kwargs_t.mode_t( + options.mode + ) resolve_res = pip_resolve( argv, diff --git a/python/online/fxreader/pr34/commands_typed/pip.py b/python/online/fxreader/pr34/commands_typed/pip.py index 8816858..e99bb85 100644 --- a/python/online/fxreader/pr34/commands_typed/pip.py +++ b/python/online/fxreader/pr34/commands_typed/pip.py @@ -1,21 +1,26 @@ import contextlib import pathlib +import enum import dataclasses -import pip._internal.commands.show -import pip._internal.commands.download -import pip._internal.cli.main_parser -import pip._internal.models.index -import pip._internal.utils.temp_dir -import pip._internal.cli.main -import pip._internal.network.download -import pip._internal.resolution.base -import pip._internal.resolution.resolvelib.resolver -import pip._internal.operations.prepare + import unittest.mock import logging +import typing + +if typing.TYPE_CHECKING: + import pip._internal.commands.show + import pip._internal.commands.download + import pip._internal.cli.main_parser + import pip._internal.models.index + import pip._internal.utils.temp_dir + import pip._internal.cli.main + import pip._internal.network.download + import pip._internal.resolution.base + import pip._internal.resolution.resolvelib.resolver + import pip._internal.operations.prepare from typing import ( - Literal, + Literal, Optional, Iterable, Any, ) logger = logging.getLogger(__name__) @@ -23,7 +28,8 @@ logger = logging.getLogger(__name__) def pip_show( argv: list[str], -) -> list[pip._internal.commands.show._PackageInfo]: +) -> list['pip._internal.commands.show._PackageInfo']: + import pip._internal.commands.show return list( pip._internal.commands.show.search_packages_info( argv, @@ -38,6 +44,7 @@ class pip_resolve_t: monkey_patch = "monkey_patch" uv_pip_compile = "uv_pip_compile" + @dataclasses.dataclass class res_t: @dataclasses.dataclass class download_info_t: @@ -67,8 +74,24 @@ def pip_resolve_entries_to_txt( def pip_resolve( argv: list[str], mode: pip_resolve_t.kwargs_t.mode_t, -) -> list[pip_resolve_t.res_t.download_info_t]: +) -> pip_resolve_t.res_t: if mode is pip_resolve_t.kwargs_t.mode_t.copy_paste: + import pip._internal.commands.show + import pip._internal.commands.download + import pip._internal.cli.cmdoptions + import pip._internal.cli.main_parser + import pip._internal.models.index + import pip._internal.utils.temp_dir + import pip._internal.cli.main + import pip._internal.network.download + import pip._internal.resolution.base + import pip._internal.req.req_install + import pip._internal.resolution.resolvelib.resolver + import pip._internal.operations.prepare + import pip._internal.utils.temp_dir + import pip._internal.operations.build.build_tracker + import pip._internal.models.direct_url + with contextlib.ExitStack() as stack: stack.enter_context(pip._internal.utils.temp_dir.global_tempdir_manager()) @@ -103,12 +126,12 @@ def pip_resolve( options.check_build_deps = None options.progress_bar = True options.require_hashes = None - options.ignore_requires_python = None + options.ignore_requires_python = False # options.cache_dir - pip._internal.commands.download.cmdoptions.check_dist_restriction(options) + pip._internal.cli.cmdoptions.check_dist_restriction(options) # t1._in_main_context = True session = t1.get_default_session(options) - target_python = pip._internal.commands.download.make_target_python(options) + target_python = pip._internal.cli.cmdoptions.make_target_python(options) finder = t1._build_package_finder( options=options, session=session, @@ -116,7 +139,7 @@ def pip_resolve( ignore_requires_python=options.ignore_requires_python, ) build_tracker = t1.enter_context( - pip._internal.commands.download.get_build_tracker() + pip._internal.operations.build.build_tracker.get_build_tracker() ) reqs = t1.get_requirements( [ @@ -127,8 +150,8 @@ def pip_resolve( finder, session, ) - pip._internal.commands.download.check_legacy_setup_py_options(options, reqs) - directory = pip._internal.commands.download.TempDirectory( + pip._internal.req.req_install.check_legacy_setup_py_options(options, reqs) + directory = pip._internal.utils.temp_dir.TempDirectory( delete=True, kind="download", globally_managed=True ) preparer = t1.make_requirement_preparer( @@ -152,59 +175,128 @@ def pip_resolve( t1.trace_basic_info(finder) requirement_set = resolver.resolve(reqs, check_supported_wheels=True) - res = pip_resolve_t.res_t( - entries=[ + res = pip_resolve_t.res_t() + + res.entries = [] + + for k, v in requirement_set.requirements.items(): + assert not v.download_info is None + assert isinstance( + v.download_info.info, + pip._internal.models.direct_url.ArchiveInfo, + ) + assert not v.download_info.info.hashes is None + + res.entries.append( pip_resolve_t.res_t.download_info_t( constraint=k, sha256=v.download_info.info.hashes["sha256"], url=v.download_info.url, ) - for k, v in requirement_set.requirements.items() - ], - ) + ) + res.txt = pip_resolve_entries_to_txt( res.entries ) return res elif mode is pip_resolve_t.kwargs_t.mode_t.monkey_patch: + import pip._internal.commands.show + import pip._internal.commands.download + import pip._internal.cli.main_parser + import pip._internal.models.index + import pip._internal.models.link + from pip._internal.models.link import ( + Link, + ) + import pip._internal.utils.temp_dir + from pip._internal.metadata.base import ( + BaseDistribution, + ) + import pip._internal.cli.main + import pip._internal.network.download + import pip._internal.resolution.base + import pip._internal.resolution.resolvelib.resolver + import pip._internal.operations.prepare + from pip._internal.network.download import ( + Downloader, + ) + from pip._internal.operations.prepare import ( + File, + ) + from pip._internal.req.req_set import RequirementSet + from pip._internal.utils.hashes import Hashes + from pip._internal.req.req_install import InstallRequirement + downloader_call_def = pip._internal.network.download.Downloader.__call__ - def downloader_call(*args): + def downloader_call( + _self: pip._internal.network.download.Downloader, + link: pip._internal.models.link.Link, + location: str, + ) -> tuple[str, str]: logger.info( dict( - url=args[1].url, + url=link.url, ) ) - return downloader_call_def(*args) + return downloader_call_def( + _self, + link, location, + ) batch_downloader_call_def = ( pip._internal.network.download.BatchDownloader.__call__ ) - def batch_downloader_call(*args): + def batch_downloader_call( + _self: pip._internal.network.download.BatchDownloader, + links: Iterable[pip._internal.models.link.Link], + location: str, + ) -> Iterable[ + tuple[ + pip._internal.models.link.Link, + tuple[str, str] + ] + ]: # print(args) logger.info( dict( - args=args, + links=links, + location=location, ) ) - return [(o, ("/dev/null", None)) for o in args[1]] + return [ + (o, ("/dev/null", '')) + for o in links + ] # base_resolver_resolve_def = pip._internal.resolution.base.BaseResolver.resolve base_resolver_resolve_def = ( pip._internal.resolution.resolvelib.resolver.Resolver.resolve ) - result_requirements = [] + result_requirements : list[ + RequirementSet | InstallRequirement + ] = [] - def base_resolver_resolve(*args, **kwargs): + def base_resolver_resolve( + _self: pip._internal.resolution.resolvelib.resolver.Resolver, + root_reqs: list[ + InstallRequirement, + ], + check_supported_wheels: bool, + ) -> RequirementSet: # print(args, kwargs) - res = base_resolver_resolve_def(*args, **kwargs) + res = base_resolver_resolve_def( + _self, + root_reqs, + check_supported_wheels + ) result_requirements.append(res) raise NotImplementedError @@ -212,7 +304,12 @@ def pip_resolve( get_http_url_def = pip._internal.operations.prepare.get_http_url - def get_http_url(link, download, download_dir=None, hashes=None): + def get_http_url( + link: Link, + download: Downloader, + download_dir: Optional[str] = None, + hashes: Optional[Hashes] = None, + ) -> File: logger.info( dict( url=link.url, @@ -224,23 +321,44 @@ def pip_resolve( print("blah") hashes = None - return "/dev/null", None + return File( + "/dev/null", + '', + ) else: - return get_http_url_def(link, download, download_dir, hashes) + return get_http_url_def( + link, + download, + download_dir, + hashes + ) prepare_linked_requirements_more_def = pip._internal.operations.prepare.RequirementPreparer.prepare_linked_requirements_more - def prepare_linked_requirements_more(_self, reqs, *args, **kwargs): - result_requirements.append(reqs) + def prepare_linked_requirements_more( + _self: pip._internal.resolution.resolvelib.resolver.Resolver, + reqs: Iterable[InstallRequirement], + parallel_builds: bool = False, + ) -> None: + result_requirements.extend( + reqs + ) raise NotImplementedError _complete_partial_requirements_def = pip._internal.operations.prepare.RequirementPreparer._complete_partial_requirements - def _complete_partial_requirements(_self, reqs, *args, **kwargs): - result_requirements.append(reqs) + def _complete_partial_requirements( + _self: pip._internal.resolution.resolvelib.resolver.Resolver, + partially_downloaded_reqs: Iterable[InstallRequirement], + parallel_builds: bool = False, + ) -> None: + result_requirements.extend( + partially_downloaded_reqs + ) raise NotImplementedError - patches = [] + patches : list[Any] = [] + patches.append( unittest.mock.patch.object( pip._internal.network.download.Downloader, "__call__", downloader_call