from __future__ import annotations from collections.abc import Mapping from contextlib import suppress from pathlib import Path from typing import TYPE_CHECKING from typing import Any from typing import ClassVar from typing import Union from conda_lock._vendor.cleo.helpers import option from packaging.utils import canonicalize_name from tomlkit import inline_table from conda_lock._vendor.poetry.console.commands.command import Command from conda_lock._vendor.poetry.console.commands.env_command import EnvCommand from conda_lock._vendor.poetry.utils.dependency_specification import RequirementsParser from conda_lock._vendor.poetry.utils.env.python_manager import Python if TYPE_CHECKING: from conda_lock._vendor.cleo.io.inputs.option import Option from packaging.utils import NormalizedName from conda_lock._vendor.poetry.core.packages.package import Package from tomlkit.items import InlineTable from conda_lock._vendor.poetry.repositories import RepositoryPool Requirements = dict[str, Union[str, Mapping[str, Any]]] class InitCommand(Command): name = "init" description = ( "Creates a basic pyproject.toml file in the current directory." ) options: ClassVar[list[Option]] = [ option("name", None, "Name of the package.", flag=False), option("description", None, "Description of the package.", flag=False), option("author", None, "Author name of the package.", flag=False), option("python", None, "Compatible Python versions.", flag=False), option( "dependency", None, "Package to require, with an optional version constraint, " "e.g. requests:^2.10.0 or requests=2.11.1.", flag=False, multiple=True, ), option( "dev-dependency", None, "Package to require for development, with an optional version" " constraint, e.g. requests:^2.10.0 or requests=2.11.1.", flag=False, multiple=True, ), option("license", "l", "License of the package.", flag=False), ] help = """\ The init command creates a basic pyproject.toml file in the\ current directory. """ def __init__(self) -> None: super().__init__() self._pool: RepositoryPool | None = None def handle(self) -> int: from pathlib import Path project_path = Path.cwd() if self.io.input.option("project"): project_path = Path(self.io.input.option("project")) if not project_path.exists() or not project_path.is_dir(): self.line_error("The --project path is not a directory.") return 1 return self._init_pyproject(project_path=project_path) def _init_pyproject( self, project_path: Path, allow_interactive: bool = True, layout_name: str = "standard", readme_format: str = "md", ) -> int: from conda_lock._vendor.poetry.core.vcs.git import GitConfig from conda_lock._vendor.poetry.config.config import Config from conda_lock._vendor.poetry.layouts import layout from conda_lock._vendor.poetry.pyproject.toml import PyProjectTOML is_interactive = self.io.is_interactive() and allow_interactive pyproject = PyProjectTOML(project_path / "pyproject.toml") if pyproject.file.exists(): if pyproject.is_poetry_project(): self.line_error( "A pyproject.toml file with a project and/or" " a poetry section already exists." ) return 1 if pyproject.data.get("build-system"): self.line_error( "A pyproject.toml file with a defined build-system already" " exists." ) return 1 vcs_config = GitConfig() if is_interactive: self.line("") self.line( "This command will guide you through creating your" " pyproject.toml config." ) self.line("") name = self.option("name") if not name: name = project_path.name.lower() if is_interactive: question = self.create_question( f"Package name [{name}]: ", default=name ) name = self.ask(question) version = "0.1.0" if is_interactive: question = self.create_question( f"Version [{version}]: ", default=version ) version = self.ask(question) description = self.option("description") or "" if not description and is_interactive: description = self.ask(self.create_question("Description []: ", default="")) author = self.option("author") if not author and vcs_config.get("user.name"): author = vcs_config["user.name"] author_email = vcs_config.get("user.email") if author_email: author += f" <{author_email}>" if is_interactive: question = self.create_question( f"Author [{author}, n to skip]: ", default=author ) question.set_validator(lambda v: self._validate_author(v, author)) author = self.ask(question) authors = [author] if author else [] license_name = self.option("license") if not license_name and is_interactive: license_name = self.ask(self.create_question("License []: ", default="")) python = self.option("python") if not python: config = Config.create() python = ( ">=" + Python.get_preferred_python(config, self.io).minor_version.to_string() ) if is_interactive: question = self.create_question( f"Compatible Python versions [{python}]: ", default=python, ) python = self.ask(question) if is_interactive: self.line("") requirements: Requirements = {} if self.option("dependency"): requirements = self._format_requirements( self._determine_requirements(self.option("dependency")) ) question_text = "Would you like to define your main dependencies interactively?" help_message = """\ You can specify a package in the following forms: - A single name (requests): this will search for matches on PyPI - A name and a constraint (requests@^2.23.0) - A git url (git+https://github.com/python-poetry/poetry.git) - A git url with a revision\ (git+https://github.com/python-poetry/poetry.git#develop) - A file path (../my-package/my-package.whl) - A directory (../my-package/) - A url (https://example.com/packages/my-package-0.1.0.tar.gz) """ help_displayed = False if is_interactive and self.confirm(question_text, True): self.line(help_message) help_displayed = True requirements.update( self._format_requirements(self._determine_requirements([])) ) self.line("") dev_requirements: Requirements = {} if self.option("dev-dependency"): dev_requirements = self._format_requirements( self._determine_requirements(self.option("dev-dependency")) ) question_text = ( "Would you like to define your development dependencies interactively?" ) if is_interactive and self.confirm(question_text, True): if not help_displayed: self.line(help_message) dev_requirements.update( self._format_requirements(self._determine_requirements([])) ) self.line("") layout_ = layout(layout_name)( name, version, description=description, author=authors[0] if authors else None, readme_format=readme_format, license=license_name, python=python, dependencies=requirements, dev_dependencies=dev_requirements, ) create_layout = not project_path.exists() if create_layout: layout_.create(project_path, with_pyproject=False) content = layout_.generate_project_content() for section, item in content.items(): pyproject.data.append(section, item) if is_interactive: self.line("Generated file") self.line("") self.line(pyproject.data.as_string().replace("\r\n", "\n")) self.line("") if is_interactive and not self.confirm("Do you confirm generation?", True): self.line_error("Command aborted") return 1 pyproject.save() if create_layout: path = project_path.resolve() with suppress(ValueError): path = path.relative_to(Path.cwd()) self.line( f"Created package {layout_._package_name} in" f" {path.as_posix()}" ) return 0 def _generate_choice_list( self, matches: list[Package], canonicalized_name: NormalizedName ) -> list[str]: choices = [] matches_names = [p.name for p in matches] exact_match = canonicalized_name in matches_names if exact_match: choices.append(matches[matches_names.index(canonicalized_name)].pretty_name) for found_package in matches: if len(choices) >= 10: break if found_package.name == canonicalized_name: continue choices.append(found_package.pretty_name) return choices def _determine_requirements( self, requires: list[str], allow_prereleases: bool | None = None, source: str | None = None, is_interactive: bool | None = None, ) -> list[dict[str, Any]]: if is_interactive is None: is_interactive = self.io.is_interactive() if not requires: result = [] question = self.create_question( "Package to add or search for (leave blank to skip):" ) question.set_validator(self._validate_package) follow_up_question = self.create_question( "\nAdd a package (leave blank to skip):" ) follow_up_question.set_validator(self._validate_package) package = self.ask(question) while package: constraint = self._parse_requirements([package])[0] if ( "git" in constraint or "url" in constraint or "path" in constraint or "version" in constraint ): self.line(f"Adding {package}") result.append(constraint) package = self.ask(follow_up_question) continue canonicalized_name = canonicalize_name(constraint["name"]) matches = self._get_pool().search(canonicalized_name) if not matches: self.line_error("Unable to find package") package = False else: choices = self._generate_choice_list(matches, canonicalized_name) info_string = ( f"Found {len(matches)} packages matching" f" {package}" ) if len(matches) > 10: info_string += "\nShowing the first 10 matches" self.line(info_string) # Default to an empty value to signal no package was selected choices.append("") package = self.choice( "\nEnter package # to add, or the complete package name if" " it is not listed", choices, attempts=3, default=len(choices) - 1, ) if not package: self.line("No package selected") # package selected by user, set constraint name to package name if package: constraint["name"] = package # no constraint yet, determine the best version automatically if package and "version" not in constraint: question = self.create_question( "Enter the version constraint to require " "(or leave blank to use the latest version):" ) question.set_max_attempts(3) question.set_validator(lambda x: (x or "").strip() or None) package_constraint = self.ask(question) if package_constraint is None: _, package_constraint = self._find_best_version_for_package( package ) self.line( f"Using version {package_constraint} for" f" {package}" ) constraint["version"] = package_constraint if package: result.append(constraint) if is_interactive: package = self.ask(follow_up_question) return result result = [] for requirement in self._parse_requirements(requires): if "git" in requirement or "url" in requirement or "path" in requirement: result.append(requirement) continue elif "version" not in requirement: # determine the best version automatically name, version = self._find_best_version_for_package( requirement["name"], allow_prereleases=allow_prereleases, source=source, ) requirement["version"] = version requirement["name"] = name self.line(f"Using version {version} for {name}") else: # check that the specified version/constraint exists # before we proceed name, _ = self._find_best_version_for_package( requirement["name"], requirement["version"], allow_prereleases=allow_prereleases, source=source, ) requirement["name"] = name result.append(requirement) return result def _find_best_version_for_package( self, name: str, required_version: str | None = None, allow_prereleases: bool | None = None, source: str | None = None, ) -> tuple[str, str]: from conda_lock._vendor.poetry.version.version_selector import VersionSelector selector = VersionSelector(self._get_pool()) package = selector.find_best_candidate( name, required_version, allow_prereleases=allow_prereleases, source=source ) if not package: # TODO: find similar raise ValueError(f"Could not find a matching version of package {name}") version = package.version.without_local() return package.pretty_name, f"^{version.to_string()}" def _parse_requirements(self, requirements: list[str]) -> list[dict[str, Any]]: from conda_lock._vendor.poetry.core.pyproject.exceptions import PyProjectError try: cwd = self.poetry.file.path.parent artifact_cache = self.poetry.pool.artifact_cache except (PyProjectError, RuntimeError): cwd = Path.cwd() artifact_cache = self._get_pool().artifact_cache parser = RequirementsParser( artifact_cache=artifact_cache, env=self.env if isinstance(self, EnvCommand) else None, cwd=cwd, ) return [parser.parse(requirement) for requirement in requirements] def _format_requirements(self, requirements: list[dict[str, str]]) -> Requirements: requires: Requirements = {} for requirement in requirements: name = requirement.pop("name") constraint: str | InlineTable if "version" in requirement and len(requirement) == 1: constraint = requirement["version"] else: constraint = inline_table() constraint.trivia.trail = "\n" constraint.update(requirement) requires[name] = constraint return requires @staticmethod def _validate_author(author: str, default: str) -> str | None: from conda_lock._vendor.poetry.core.utils.helpers import combine_unicode from conda_lock._vendor.poetry.core.utils.patterns import AUTHOR_REGEX author = combine_unicode(author or default) if author in ["n", "no"]: return None m = AUTHOR_REGEX.match(author) if not m: raise ValueError( "Invalid author string. Must be in the format: " "John Smith " ) return author @staticmethod def _validate_package(package: str | None) -> str | None: if package and len(package.split()) > 2: raise ValueError("Invalid package definition.") return package def _get_pool(self) -> RepositoryPool: from conda_lock._vendor.poetry.config.config import Config from conda_lock._vendor.poetry.repositories import RepositoryPool from conda_lock._vendor.poetry.repositories.pypi_repository import PyPiRepository if isinstance(self, EnvCommand): return self.poetry.pool if self._pool is None: self._pool = RepositoryPool() pool_size = Config.create().installer_max_workers self._pool.add_repository(PyPiRepository(pool_size=pool_size)) return self._pool