"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "poetry/installation/pip_installer.py" between
poetry-1.1.15.tar.gz and poetry-1.2.0.tar.gz

About: Poetry is a tool for dependency management and packaging in Python.

pip_installer.py  (poetry-1.1.15)	:	pip_installer.py  (poetry-1.2.0)
		from __future__ import annotations
import os		import os
import tempfile		import tempfile
		import urllib.parse
		from pathlib import Path
from subprocess import CalledProcessError		from subprocess import CalledProcessError
		from typing import TYPE_CHECKING
from clikit.api.io import IO		from typing import Any
from poetry.core.pyproject.toml import PyProjectTOML		from poetry.core.pyproject.toml import PyProjectTOML
from poetry.repositories.pool import Pool		from poetry.core.semver.version import Version
		from poetry.installation.base_installer import BaseInstaller
		from poetry.repositories.http import HTTPRepository
from poetry.utils._compat import encode		from poetry.utils._compat import encode
from poetry.utils.env import Env		from poetry.utils.helpers import remove_directory
from poetry.utils.helpers import safe_rmtree		from poetry.utils.pip import pip_install
from .base_installer import BaseInstaller		if TYPE_CHECKING:
		from cleo.io.io import IO
		from poetry.core.masonry.builders.builder import Builder
		from poetry.core.packages.package import Package
try:		from poetry.repositories.pool import Pool
import urllib.parse as urlparse		from poetry.utils.env import Env
except ImportError:
import urlparse
class PipInstaller(BaseInstaller):		class PipInstaller(BaseInstaller):
def __init__(self, env, io, pool): # type: (Env, IO, Pool) -> None		def __init__(self, env: Env, io: IO, pool: Pool) -> None:
self._env = env		self._env = env
self._io = io		self._io = io
self._pool = pool		self._pool = pool
def install(self, package, update=False):		def install(self, package: Package, update: bool = False) -> None:
if package.source_type == "directory":		if package.source_type == "directory":
self.install_directory(package)		self.install_directory(package)
return		return
if package.source_type == "git":		if package.source_type == "git":
self.install_git(package)		self.install_git(package)
return		return
args = ["install", "--no-deps"]		args = ["install", "--no-deps"]
if (		if (
package.source_type not in {"git", "directory", "file", "url"}		package.source_type not in {"git", "directory", "file", "url"}
and package.source_url		and package.source_url
):		):
		assert package.source_reference is not None
repository = self._pool.repository(package.source_reference)		repository = self._pool.repository(package.source_reference)
parsed = urlparse.urlparse(package.source_url)		parsed = urllib.parse.urlparse(package.source_url)
if parsed.scheme == "http":		if parsed.scheme == "http":
self._io.error(		assert parsed.hostname is not None
" <warning>Installing from unsecure host: {}</warning>".f		self._io.write_error(
ormat(		" <warning>Installing from unsecure host:"
parsed.hostname		f" {parsed.hostname}</warning>"
)
)		)
args += ["--trusted-host", parsed.hostname]		args += ["--trusted-host", parsed.hostname]
if repository.cert:		if isinstance(repository, HTTPRepository):
args += ["--cert", str(repository.cert)]		certificates = repository.certificates
		if certificates.cert:
		args += ["--cert", str(certificates.cert)]
		if parsed.scheme == "https" and not certificates.verify:
		assert parsed.hostname is not None
		args += ["--trusted-host", parsed.hostname]
		if certificates.client_cert:
		args += ["--client-cert", str(certificates.client_cert)]
if repository.client_cert:		index_url = repository.authenticated_url
args += ["--client-cert", str(repository.client_cert)]
index_url = repository.authenticated_url		args += ["--index-url", index_url]
args += ["--index-url", index_url]		if (
if self._pool.has_default():		self._pool.has_default()
if repository.name != self._pool.repositories[0].name:		and repository.name != self._pool.repositories[0].name
		):
		first_repository = self._pool.repositories[0]
		if isinstance(first_repository, HTTPRepository):
args += [		args += [
"--extra-index-url",		"--extra-index-url",
self._pool.repositories[0].authenticated_url,		first_repository.authenticated_url,
]		]
if update:		if update:
args.append("-U")		args.append("-U")
		req: str | list[str]
if package.files and not package.source_url:		if package.files and not package.source_url:
# Format as a requirements.txt		# Format as a requirements.txt
# We need to create a requirements.txt file		# We need to create a requirements.txt file
# for each package in order to check hashes.		# for each package in order to check hashes.
# This is far from optimal but we do not have any		# This is far from optimal but we do not have any
# other choice since this is the only way for pip		# other choice since this is the only way for pip
# to verify hashes.		# to verify hashes.
req = self.create_temporary_requirement(package)		req = self.create_temporary_requirement(package)
args += ["-r", req]		args += ["-r", req]
skipping to change at line 96		skipping to change at line 119
os.unlink(req)		os.unlink(req)
else:		else:
req = self.requirement(package)		req = self.requirement(package)
if not isinstance(req, list):		if not isinstance(req, list):
args.append(req)		args.append(req)
else:		else:
args += req		args += req
self.run(*args)		self.run(*args)
def update(self, package, target):		def update(self, package: Package, target: Package) -> None:
if package.source_type != target.source_type:		if package.source_type != target.source_type:
# If the source type has changed, we remove the current		# If the source type has changed, we remove the current
# package to avoid perpetual updates in some cases		# package to avoid perpetual updates in some cases
self.remove(package)		self.remove(package)
self.install(target, update=True)		self.install(target, update=True)
def remove(self, package):		def remove(self, package: Package) -> None:
try:		try:
self.run("uninstall", package.name, "-y")		self.run("uninstall", package.name, "-y")
except CalledProcessError as e:		except CalledProcessError as e:
if "not installed" in str(e):		if "not installed" in str(e):
return		return
raise		raise
# This is a workaround for https://github.com/pypa/pip/issues/4176		# This is a workaround for https://github.com/pypa/pip/issues/4176
nspkg_pth_file = self._env.site_packages.path / "{}-nspkg.pth".format(		for nspkg_pth_file in self._env.site_packages.find_distribution_nspkg_pt
package.name		h_files(
)		distribution_name=package.name
if nspkg_pth_file.exists():		):
nspkg_pth_file.unlink()		nspkg_pth_file.unlink()
# If we have a VCS package, remove its source directory		# If we have a VCS package, remove its source directory
if package.source_type == "git":		if package.source_type == "git":
src_dir = self._env.path / "src" / package.name		src_dir = self._env.path / "src" / package.name
if src_dir.exists():		if src_dir.exists():
safe_rmtree(str(src_dir))		remove_directory(src_dir, force=True)
def run(self, *args, **kwargs): # type: (...) -> str		def run(self, *args: Any, **kwargs: Any) -> int | str:
return self._env.run_pip(*args, **kwargs)		return self._env.run_pip(*args, **kwargs)
def requirement(self, package, formatted=False):		def requirement(self, package: Package, formatted: bool = False) -> str | li st[str]:
if formatted and not package.source_type:		if formatted and not package.source_type:
req = "{}=={}".format(package.name, package.version)		req = f"{package.name}=={package.version}"
for f in package.files:		for f in package.files:
hash_type = "sha256"		hash_type = "sha256"
h = f["hash"]		h = f["hash"]
if ":" in h:		if ":" in h:
hash_type, h = h.split(":")		hash_type, h = h.split(":")
req += " --hash {}:{}".format(hash_type, h)		req += f" --hash {hash_type}:{h}"
req += "\n"		req += "\n"
return req		return req
if package.source_type in ["file", "directory"]:		if package.source_type in ["file", "directory"]:
		assert package.source_url is not None
if package.root_dir:		if package.root_dir:
req = (package.root_dir / package.source_url).as_posix()		req = (package.root_dir / package.source_url).as_posix()
else:		else:
req = os.path.realpath(package.source_url)		req = os.path.realpath(package.source_url)
if package.develop and package.source_type == "directory":		if package.develop and package.source_type == "directory":
req = ["-e", req]		return ["-e", req]
return req		return req
if package.source_type == "git":		if package.source_type == "git":
req = "git+{}@{}#egg={}".format(		req = (
package.source_url, package.source_reference, package.name		f"git+{package.source_url}@{package.source_reference}"
		f"#egg={package.name}"
)		)
		if package.source_subdirectory:
		req += f"&subdirectory={package.source_subdirectory}"
if package.develop:		if package.develop:
req = ["-e", req]		return ["-e", req]
return req		return req
if package.source_type == "url":		if package.source_type == "url":
return "{}#egg={}".format(package.source_url, package.name)		return f"{package.source_url}#egg={package.name}"
return "{}=={}".format(package.name, package.version)		return f"{package.name}=={package.version}"
def create_temporary_requirement(self, package):		def create_temporary_requirement(self, package: Package) -> str:
fd, name = tempfile.mkstemp(		fd, name = tempfile.mkstemp("reqs.txt", f"{package.name}-{package.versio
"reqs.txt", "{}-{}".format(package.name, package.version)		n}")
)		req = self.requirement(package, formatted=True)
		if isinstance(req, list):
		req = " ".join(req)
try:		try:
os.write(fd, encode(self.requirement(package, formatted=True)))		os.write(fd, encode(req))
finally:		finally:
os.close(fd)		os.close(fd)
return name		return name
def install_directory(self, package):		def install_directory(self, package: Package) -> str | int:
		from cleo.io.null_io import NullIO
from poetry.factory import Factory		from poetry.factory import Factory
from poetry.io.null_io import NullIO
		assert package.source_url is not None
if package.root_dir:		if package.root_dir:
req = (package.root_dir / package.source_url).as_posix()		req = package.root_dir / package.source_url
else:		else:
req = os.path.realpath(package.source_url)		req = Path(package.source_url).resolve(strict=False)
args = ["install", "--no-deps", "-U"]		if package.source_subdirectory:
		req /= package.source_subdirectory
pyproject = PyProjectTOML(os.path.join(req, "pyproject.toml"))		pyproject = PyProjectTOML(os.path.join(req, "pyproject.toml"))
if pyproject.is_poetry_project():		if pyproject.is_poetry_project():
# Even if there is a build system specified		# Even if there is a build system specified
# some versions of pip (< 19.0.0) don't understand it		# some versions of pip (< 19.0.0) don't understand it
# so we need to check the version of pip to know		# so we need to check the version of pip to know
# if we can rely on the build system		# if we can rely on the build system
legacy_pip = self._env.pip_version < self._env.pip_version.__class__		legacy_pip = self._env.pip_version < Version.from_parts(19, 0, 0)
(
19, 0, 0
)
try:		try:
package_poetry = Factory().create_poetry(pyproject.file.path.par ent)		package_poetry = Factory().create_poetry(pyproject.file.path.par ent)
except RuntimeError:		except RuntimeError:
package_poetry = None		package_poetry = None
if package_poetry is not None:		if package_poetry is not None:
		builder: Builder
if package.develop and not package_poetry.package.build_script:		if package.develop and not package_poetry.package.build_script:
from poetry.masonry.builders.editable import EditableBuilder		from poetry.masonry.builders.editable import EditableBuilder
# This is a Poetry package in editable mode		# This is a Poetry package in editable mode
# we can use the EditableBuilder without going through pip		# we can use the EditableBuilder without going through pip
# to install it, unless it has a build script.		# to install it, unless it has a build script.
builder = EditableBuilder(package_poetry, self._env, NullIO( ))		builder = EditableBuilder(package_poetry, self._env, NullIO( ))
builder.build()		builder.build()
return 0		return 0
skipping to change at line 231		skipping to change at line 261
from poetry.core.masonry.builders.sdist import SdistBuilder		from poetry.core.masonry.builders.sdist import SdistBuilder
# We need to rely on creating a temporary setup.py		# We need to rely on creating a temporary setup.py
# file since the version of pip does not support		# file since the version of pip does not support
# build-systems		# build-systems
# We also need it for non-PEP-517 packages		# We also need it for non-PEP-517 packages
builder = SdistBuilder(package_poetry)		builder = SdistBuilder(package_poetry)
with builder.setup_py():		with builder.setup_py():
if package.develop:		if package.develop:
args.append("-e")		return pip_install(
		path=req,
args.append(req)		environment=self._env,
		upgrade=True,
return self.run(*args)		editable=True,
		)
		return pip_install(
		path=req, environment=self._env, deps=False, upgrade
		=True
		)
if package.develop:		if package.develop:
args.append("-e")		return pip_install(
		path=req, environment=self._env, upgrade=True, editable=True
args.append(req)		)
		return pip_install(path=req, environment=self._env, deps=False, upgrade=
return self.run(*args)		True)
def install_git(self, package):
from poetry.core.packages import Package
from poetry.core.vcs import Git
src_dir = self._env.path / "src" / package.name
if src_dir.exists():
safe_rmtree(str(src_dir))
src_dir.parent.mkdir(exist_ok=True)
git = Git()		def install_git(self, package: Package) -> None:
git.clone(package.source_url, src_dir)		from poetry.core.packages.package import Package
reference = package.source_resolved_reference		from poetry.vcs.git import Git
if not reference:
reference = package.source_reference
git.checkout(reference, src_dir)		assert package.source_url is not None
		source = Git.clone(
		url=package.source_url,
		source_root=self._env.path / "src",
		revision=package.source_resolved_reference or package.source_referen
		ce,
		)
# Now we just need to install from the source directory		# Now we just need to install from the source directory
pkg = Package(package.name, package.version)		pkg = Package(
pkg._source_type = "directory"		name=package.name,
pkg._source_url = str(src_dir)		version=package.version,
pkg.develop = package.develop		source_type="directory",
		source_url=str(source.path),
		source_subdirectory=package.source_subdirectory,
		develop=package.develop,
		)
self.install_directory(pkg)		self.install_directory(pkg)
End of changes. 50 change blocks.
88 lines changed or deleted		121 lines changed or added

---