is a tool for dependency management and packaging in Python.
| installer.py (poetry-1.1.15) | : | installer.py (poetry-1.2.0) |
|---|
|
| from typing import List | | from __future__ import annotations |
| from typing import Optional | | |
| from typing import Union | | from typing import TYPE_CHECKING |
| | | |
| from clikit.api.io import IO | | from cleo.io.null_io import NullIO |
| | | from packaging.utils import canonicalize_name |
| from poetry.config.config import Config | | |
| from poetry.core.packages.project_package import ProjectPackage | | from poetry.installation.executor import Executor |
| from poetry.io.null_io import NullIO | | from poetry.installation.operations import Install |
| from poetry.packages import Locker | | from poetry.installation.operations import Uninstall |
| | from poetry.installation.operations import Update |
| | from poetry.installation.pip_installer import PipInstaller |
| from poetry.repositories import Pool | | from poetry.repositories import Pool |
| from poetry.repositories import Repository | | from poetry.repositories import Repository |
| from poetry.repositories.installed_repository import InstalledRepository | | from poetry.repositories.installed_repository import InstalledRepository |
|
| | from poetry.repositories.lockfile_repository import LockfileRepository |
| from poetry.utils.extras import get_extra_package_names | | from poetry.utils.extras import get_extra_package_names |
|
| from poetry.utils.helpers import canonicalize_name | | from poetry.utils.helpers import pluralize |
| | |
|
| from .base_installer import BaseInstaller | | if TYPE_CHECKING: |
| from .executor import Executor | | from collections.abc import Iterable |
| from .operations import Install | | from collections.abc import Sequence |
| from .operations import Uninstall | | |
| from .operations import Update | | from cleo.io.io import IO |
| from .operations.operation import Operation | | from poetry.core.packages.project_package import ProjectPackage |
| from .pip_installer import PipInstaller | | |
| | from poetry.config.config import Config |
| | from poetry.installation.base_installer import BaseInstaller |
| | from poetry.installation.operations.operation import Operation |
| | from poetry.packages import Locker |
| | from poetry.utils.env import Env |
| | |
| class Installer: | | class Installer: |
| def __init__( | | def __init__( |
| self, | | self, |
|
| io, # type: IO | | io: IO, |
| env, | | env: Env, |
| package, # type: ProjectPackage | | package: ProjectPackage, |
| locker, # type: Locker | | locker: Locker, |
| pool, # type: Pool | | pool: Pool, |
| config, # type: Config | | config: Config, |
| installed=None, # type: Union[InstalledRepository, None] | | installed: Repository | None = None, |
| executor=None, # type: Optional[Executor] | | executor: Executor | None = None, |
| ): | | ) -> None: |
| self._io = io | | self._io = io |
| self._env = env | | self._env = env |
| self._package = package | | self._package = package |
| self._locker = locker | | self._locker = locker |
| self._pool = pool | | self._pool = pool |
| | |
| self._dry_run = False | | self._dry_run = False |
|
| self._remove_untracked = False | | self._requires_synchronization = False |
| self._update = False | | self._update = False |
| self._verbose = False | | self._verbose = False |
| self._write_lock = True | | self._write_lock = True |
|
| self._dev_mode = True | | self._groups: Iterable[str] | None = None |
| | |
| self._execute_operations = True | | self._execute_operations = True |
| self._lock = False | | self._lock = False |
| | |
|
| self._whitelist = [] | | self._whitelist: list[str] = [] |
| | |
|
| self._extras = [] | | self._extras: list[str] = [] |
| | |
| if executor is None: | | if executor is None: |
| executor = Executor(self._env, self._pool, config, self._io) | | executor = Executor(self._env, self._pool, config, self._io) |
| | |
| self._executor = executor | | self._executor = executor |
| self._use_executor = False | | self._use_executor = False |
| | |
| self._installer = self._get_installer() | | self._installer = self._get_installer() |
| if installed is None: | | if installed is None: |
| installed = self._get_installed() | | installed = self._get_installed() |
| | |
| self._installed_repository = installed | | self._installed_repository = installed |
| | |
| @property | | @property |
|
| def executor(self): | | def executor(self) -> Executor: |
| return self._executor | | return self._executor |
| | |
| @property | | @property |
|
| def installer(self): | | def installer(self) -> BaseInstaller: |
| return self._installer | | return self._installer |
| | |
|
| def set_package(self, package): # type: (ProjectPackage) -> Installer | | def set_package(self, package: ProjectPackage) -> Installer: |
| self._package = package | | self._package = package |
| | |
| return self | | return self |
| | |
|
| def set_locker(self, locker): # type: (Locker) -> Installer | | def set_locker(self, locker: Locker) -> Installer: |
| self._locker = locker | | self._locker = locker |
| | |
| return self | | return self |
| | |
|
| def run(self): | | def run(self) -> int: |
| # Check if refresh | | # Check if refresh |
| if not self._update and self._lock and self._locker.is_locked(): | | if not self._update and self._lock and self._locker.is_locked(): |
| return self._do_refresh() | | return self._do_refresh() |
| | |
| # Force update if there is no lock file present | | # Force update if there is no lock file present |
| if not self._update and not self._locker.is_locked(): | | if not self._update and not self._locker.is_locked(): |
| self._update = True | | self._update = True |
| | |
| if self.is_dry_run(): | | if self.is_dry_run(): |
| self.verbose(True) | | self.verbose(True) |
| self._write_lock = False | | self._write_lock = False |
| self._execute_operations = False | | self._execute_operations = False |
| | |
|
| local_repo = Repository() | | return self._do_install() |
| | | |
| return self._do_install(local_repo) | | |
| | |
|
| def dry_run(self, dry_run=True): # type: (bool) -> Installer | | def dry_run(self, dry_run: bool = True) -> Installer: |
| self._dry_run = dry_run | | self._dry_run = dry_run |
| self._executor.dry_run(dry_run) | | self._executor.dry_run(dry_run) |
| | |
| return self | | return self |
| | |
|
| def is_dry_run(self): # type: () -> bool | | def is_dry_run(self) -> bool: |
| return self._dry_run | | return self._dry_run |
| | |
|
| def remove_untracked(self, remove_untracked=True): # type: (bool) -> Instal | | def requires_synchronization( |
| ler | | self, requires_synchronization: bool = True |
| self._remove_untracked = remove_untracked | | ) -> Installer: |
| | self._requires_synchronization = requires_synchronization |
| | |
| return self | | return self |
| | |
|
| def is_remove_untracked(self): # type: () -> bool | | def verbose(self, verbose: bool = True) -> Installer: |
| return self._remove_untracked | | |
| | |
| def verbose(self, verbose=True): # type: (bool) -> Installer | | |
| self._verbose = verbose | | self._verbose = verbose |
| self._executor.verbose(verbose) | | self._executor.verbose(verbose) |
| | |
| return self | | return self |
| | |
|
| def is_verbose(self): # type: () -> bool | | def is_verbose(self) -> bool: |
| return self._verbose | | return self._verbose |
| | |
|
| def dev_mode(self, dev_mode=True): # type: (bool) -> Installer | | def only_groups(self, groups: Iterable[str]) -> Installer: |
| self._dev_mode = dev_mode | | self._groups = groups |
| | |
| return self | | return self |
| | |
|
| def is_dev_mode(self): # type: () -> bool | | def update(self, update: bool = True) -> Installer: |
| return self._dev_mode | | |
| | |
| def update(self, update=True): # type: (bool) -> Installer | | |
| self._update = update | | self._update = update |
| | |
| return self | | return self |
| | |
|
| def lock(self, update=True): # type: (bool) -> Installer | | def lock(self, update: bool = True) -> Installer: |
| """ | | """ |
| Prepare the installer for locking only. | | Prepare the installer for locking only. |
| """ | | """ |
| self.update(update=update) | | self.update(update=update) |
| self.execute_operations(False) | | self.execute_operations(False) |
| self._lock = True | | self._lock = True |
| | |
| return self | | return self |
| | |
|
| def is_updating(self): # type: () -> bool | | def is_updating(self) -> bool: |
| return self._update | | return self._update |
| | |
|
| def execute_operations(self, execute=True): # type: (bool) -> Installer | | def execute_operations(self, execute: bool = True) -> Installer: |
| self._execute_operations = execute | | self._execute_operations = execute |
| | |
| if not execute: | | if not execute: |
| self._executor.disable() | | self._executor.disable() |
| | |
| return self | | return self |
| | |
|
| def whitelist(self, packages): # type: (dict) -> Installer | | def whitelist(self, packages: Iterable[str]) -> Installer: |
| self._whitelist = [canonicalize_name(p) for p in packages] | | self._whitelist = [canonicalize_name(p) for p in packages] |
| | |
| return self | | return self |
| | |
|
| def extras(self, extras): # type: (list) -> Installer | | def extras(self, extras: list[str]) -> Installer: |
| self._extras = extras | | self._extras = extras |
| | |
| return self | | return self |
| | |
|
| def use_executor(self, use_executor=True): # type: (bool) -> Installer | | def use_executor(self, use_executor: bool = True) -> Installer: |
| self._use_executor = use_executor | | self._use_executor = use_executor |
| | |
| return self | | return self |
| | |
|
| def _do_refresh(self): | | def _do_refresh(self) -> int: |
| from poetry.puzzle import Solver | | from poetry.puzzle.solver import Solver |
| | |
| # Checking extras | | # Checking extras |
| for extra in self._extras: | | for extra in self._extras: |
| if extra not in self._package.extras: | | if extra not in self._package.extras: |
|
| raise ValueError("Extra [{}] is not specified.".format(extra)) | | raise ValueError(f"Extra [{extra}] is not specified.") |
| | |
|
| locked_repository = self._locker.locked_repository(True) | | locked_repository = self._locker.locked_repository() |
| solver = Solver( | | solver = Solver( |
| self._package, | | self._package, |
| self._pool, | | self._pool, |
|
| locked_repository, | | locked_repository.packages, |
| locked_repository, | | locked_repository.packages, |
| self._io, # noqa | | self._io, |
| ) | | ) |
| | |
|
| ops = solver.solve(use_latest=[]) | | with solver.provider.use_source_root( |
| | source_root=self._env.path.joinpath("src") |
| | ): |
| | ops = solver.solve(use_latest=[]).calculate_operations() |
| | |
|
| local_repo = Repository() | | lockfile_repo = LockfileRepository() |
| self._populate_local_repo(local_repo, ops) | | self._populate_lockfile_repo(lockfile_repo, ops) |
| | |
|
| self._write_lock_file(local_repo, force=True) | | self._write_lock_file(lockfile_repo, force=True) |
| | |
| return 0 | | return 0 |
| | |
|
| def _do_install(self, local_repo): | | def _do_install(self) -> int: |
| from poetry.puzzle import Solver | | from poetry.puzzle.solver import Solver |
| | |
|
| locked_repository = Repository() | | locked_repository = Repository("poetry-locked") |
| if self._update: | | if self._update: |
| if self._locker.is_locked() and not self._lock: | | if self._locker.is_locked() and not self._lock: |
|
| locked_repository = self._locker.locked_repository(True) | | locked_repository = self._locker.locked_repository() |
| | |
| # If no packages have been whitelisted (The ones we want to upda
te), | | # If no packages have been whitelisted (The ones we want to upda
te), |
| # we whitelist every package in the lock file. | | # we whitelist every package in the lock file. |
| if not self._whitelist: | | if not self._whitelist: |
| for pkg in locked_repository.packages: | | for pkg in locked_repository.packages: |
| self._whitelist.append(pkg.name) | | self._whitelist.append(pkg.name) |
| | |
| # Checking extras | | # Checking extras |
| for extra in self._extras: | | for extra in self._extras: |
| if extra not in self._package.extras: | | if extra not in self._package.extras: |
|
| raise ValueError("Extra [{}] is not specified.".format(extra
)) | | raise ValueError(f"Extra [{extra}] is not specified.") |
| | |
| self._io.write_line("<info>Updating dependencies</>") | | self._io.write_line("<info>Updating dependencies</>") |
| solver = Solver( | | solver = Solver( |
| self._package, | | self._package, |
| self._pool, | | self._pool, |
|
| self._installed_repository, | | self._installed_repository.packages, |
| locked_repository, | | locked_repository.packages, |
| self._io, | | self._io, |
|
| remove_untracked=self._remove_untracked, | | |
| ) | | ) |
| | |
|
| ops = solver.solve(use_latest=self._whitelist) | | with solver.provider.use_source_root( |
| | source_root=self._env.path.joinpath("src") |
| | ): |
| | ops = solver.solve(use_latest=self._whitelist).calculate_operati |
| | ons() |
| else: | | else: |
| self._io.write_line("<info>Installing dependencies from lock file</>
") | | self._io.write_line("<info>Installing dependencies from lock file</>
") |
| | |
|
| locked_repository = self._locker.locked_repository(True) | | locked_repository = self._locker.locked_repository() |
| | |
| if not self._locker.is_fresh(): | | if not self._locker.is_fresh(): |
|
| self._io.write_line( | | self._io.write_error_line( |
| "<warning>" | | "<warning>" |
|
| "Warning: The lock file is not up to date with " | | "Warning: poetry.lock is not consistent with pyproject.toml. |
| "the latest changes in pyproject.toml. " | | " |
| "You may be getting outdated dependencies. " | | "You may be getting improper dependencies. " |
| "Run update to update them." | | "Run `poetry lock [--no-update]` to fix it." |
| "</warning>" | | "</warning>" |
| ) | | ) |
| | |
| for extra in self._extras: | | for extra in self._extras: |
| if extra not in self._locker.lock_data.get("extras", {}): | | if extra not in self._locker.lock_data.get("extras", {}): |
|
| raise ValueError("Extra [{}] is not specified.".format(extra
)) | | raise ValueError(f"Extra [{extra}] is not specified.") |
| | |
| # If we are installing from lock | | # If we are installing from lock |
| # Filter the operations by comparing it with what is | | # Filter the operations by comparing it with what is |
| # currently installed | | # currently installed |
| ops = self._get_operations_from_lock(locked_repository) | | ops = self._get_operations_from_lock(locked_repository) |
| | |
|
| self._populate_local_repo(local_repo, ops) | | lockfile_repo = LockfileRepository() |
| | self._populate_lockfile_repo(lockfile_repo, ops) |
| | |
| if self._update: | | if self._update: |
|
| self._write_lock_file(local_repo) | | self._write_lock_file(lockfile_repo) |
| | |
| if self._lock: | | if self._lock: |
| # If we are only in lock mode, no need to go any further | | # If we are only in lock mode, no need to go any further |
| return 0 | | return 0 |
| | |
|
| root = self._package | | if self._groups is not None: |
| if not self.is_dev_mode(): | | root = self._package.with_dependency_groups(list(self._groups), only |
| root = root.clone() | | =True) |
| del root.dev_requires[:] | | else: |
| | root = self._package.without_optional_dependency_groups() |
| | |
| if self._io.is_verbose(): | | if self._io.is_verbose(): |
| self._io.write_line("") | | self._io.write_line("") |
| self._io.write_line( | | self._io.write_line( |
| "<info>Finding the necessary packages for the current system</>" | | "<info>Finding the necessary packages for the current system</>" |
| ) | | ) |
| | |
| # We resolve again by only using the lock file | | # We resolve again by only using the lock file |
| pool = Pool(ignore_repository_names=True) | | pool = Pool(ignore_repository_names=True) |
| | |
| # Making a new repo containing the packages | | # Making a new repo containing the packages |
| # newly resolved and the ones from the current lock file | | # newly resolved and the ones from the current lock file |
|
| repo = Repository() | | repo = Repository("poetry-repo") |
| for package in local_repo.packages + locked_repository.packages: | | for package in lockfile_repo.packages + locked_repository.packages: |
| if not repo.has_package(package): | | if not package.is_direct_origin() and not repo.has_package(package): |
| repo.add_package(package) | | repo.add_package(package) |
| | |
| pool.add_repository(repo) | | pool.add_repository(repo) |
| | |
| solver = Solver( | | solver = Solver( |
| root, | | root, |
| pool, | | pool, |
|
| self._installed_repository, | | self._installed_repository.packages, |
| locked_repository, | | locked_repository.packages, |
| NullIO(), | | NullIO(), |
|
| remove_untracked=self._remove_untracked, | | |
| ) | | ) |
| # Everything is resolved at this point, so we no longer need | | # Everything is resolved at this point, so we no longer need |
| # to load deferred dependencies (i.e. VCS, URL and path dependencies) | | # to load deferred dependencies (i.e. VCS, URL and path dependencies) |
| solver.provider.load_deferred(False) | | solver.provider.load_deferred(False) |
| | |
| with solver.use_environment(self._env): | | with solver.use_environment(self._env): |
|
| ops = solver.solve(use_latest=self._whitelist) | | ops = solver.solve(use_latest=self._whitelist).calculate_operations( |
| | with_uninstalls=self._requires_synchronization, |
| | synchronize=self._requires_synchronization, |
| | ) |
| | |
| | if not self._requires_synchronization: |
| | # If no packages synchronisation has been requested we need |
| | # to calculate the uninstall operations |
| | from poetry.puzzle.transaction import Transaction |
| | |
| | transaction = Transaction( |
| | locked_repository.packages, |
| | [(package, 0) for package in lockfile_repo.packages], |
| | installed_packages=self._installed_repository.packages, |
| | root_package=root, |
| | ) |
| | |
| | ops = [ |
| | op |
| | for op in transaction.calculate_operations(with_uninstalls=True) |
| | if op.job_type == "uninstall" |
| | ] + ops |
| | |
| # We need to filter operations so that packages | | # We need to filter operations so that packages |
| # not compatible with the current system, | | # not compatible with the current system, |
| # or optional and not requested, are dropped | | # or optional and not requested, are dropped |
|
| self._filter_operations(ops, local_repo) | | self._filter_operations(ops, lockfile_repo) |
| | |
| # Execute operations | | # Execute operations |
| return self._execute(ops) | | return self._execute(ops) |
| | |
|
| def _write_lock_file(self, repo, force=True): # type: (Repository, bool) -> | | def _write_lock_file(self, repo: LockfileRepository, force: bool = False) -> |
| None | | None: |
| if force or (self._update and self._write_lock): | | if self._write_lock and (force or self._update): |
| updated_lock = self._locker.set_lock_data(self._package, repo.packag
es) | | updated_lock = self._locker.set_lock_data(self._package, repo.packag
es) |
| | |
| if updated_lock: | | if updated_lock: |
| self._io.write_line("") | | self._io.write_line("") |
| self._io.write_line("<info>Writing lock file</>") | | self._io.write_line("<info>Writing lock file</>") |
| | |
|
| def _execute(self, operations): | | def _execute(self, operations: list[Operation]) -> int: |
| if self._use_executor: | | if self._use_executor: |
| return self._executor.execute(operations) | | return self._executor.execute(operations) |
| | |
| if not operations and (self._execute_operations or self._dry_run): | | if not operations and (self._execute_operations or self._dry_run): |
| self._io.write_line("No dependencies to install or update") | | self._io.write_line("No dependencies to install or update") |
| | |
| if operations and (self._execute_operations or self._dry_run): | | if operations and (self._execute_operations or self._dry_run): |
| installs = 0 | | installs = 0 |
| updates = 0 | | updates = 0 |
| uninstalls = 0 | | uninstalls = 0 |
| | |
| skipping to change at line 345 | | skipping to change at line 373 |
|---|
| if op.skipped: | | if op.skipped: |
| skipped += 1 | | skipped += 1 |
| elif op.job_type == "install": | | elif op.job_type == "install": |
| installs += 1 | | installs += 1 |
| elif op.job_type == "update": | | elif op.job_type == "update": |
| updates += 1 | | updates += 1 |
| elif op.job_type == "uninstall": | | elif op.job_type == "uninstall": |
| uninstalls += 1 | | uninstalls += 1 |
| | |
| self._io.write_line("") | | self._io.write_line("") |
|
| self._io.write_line( | | self._io.write("Package operations: ") |
| "Package operations: " | | self._io.write(f"<info>{installs}</> install{pluralize(installs)}, " |
| "<info>{}</> install{}, " | | ) |
| "<info>{}</> update{}, " | | self._io.write(f"<info>{updates}</> update{pluralize(updates)}, ") |
| "<info>{}</> removal{}" | | self._io.write(f"<info>{uninstalls}</> removal{pluralize(uninstalls) |
| "{}".format( | | }") |
| installs, | | if skipped and self.is_verbose(): |
| "" if installs == 1 else "s", | | self._io.write(f", <info>{skipped}</> skipped") |
| updates, | | self._io.write_line("") |
| "" if updates == 1 else "s", | | |
| uninstalls, | | |
| "" if uninstalls == 1 else "s", | | |
| ", <info>{}</> skipped".format(skipped) | | |
| if skipped and self.is_verbose() | | |
| else "", | | |
| ) | | |
| ) | | |
| | |
| self._io.write_line("") | | self._io.write_line("") |
| | |
| for op in operations: | | for op in operations: |
| self._execute_operation(op) | | self._execute_operation(op) |
| | |
| return 0 | | return 0 |
| | |
|
| def _execute_operation(self, operation): # type: (Operation) -> None | | def _execute_operation(self, operation: Operation) -> None: |
| """ | | """ |
| Execute a given operation. | | Execute a given operation. |
| """ | | """ |
| method = operation.job_type | | method = operation.job_type |
| | |
|
| getattr(self, "_execute_{}".format(method))(operation) | | getattr(self, f"_execute_{method}")(operation) |
| | |
|
| def _execute_install(self, operation): # type: (Install) -> None | | def _execute_install(self, operation: Install) -> None: |
| | target = operation.package |
| if operation.skipped: | | if operation.skipped: |
| if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): | | if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Skipping <c1>{}</c1> (<c2>{}</c2>) {}".format( | | f" - Skipping <c1>{target.pretty_name}</c1>" |
| operation.package.pretty_name, | | f" (<c2>{target.full_pretty_version}</c2>) {operation.skip_r |
| operation.package.full_pretty_version, | | eason}" |
| operation.skip_reason, | | |
| ) | | |
| ) | | ) |
| | |
| return | | return |
| | |
| if self._execute_operations or self.is_dry_run(): | | if self._execute_operations or self.is_dry_run(): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Installing <c1>{}</c1> (<c2>{}</c2>)".format( | | f" - Installing <c1>{target.pretty_name}</c1>" |
| operation.package.pretty_name, operation.package.full_pretty | | f" (<c2>{target.full_pretty_version}</c2>)" |
| _version | | |
| ) | | |
| ) | | ) |
| | |
| if not self._execute_operations: | | if not self._execute_operations: |
| return | | return |
| | |
| self._installer.install(operation.package) | | self._installer.install(operation.package) |
| | |
|
| def _execute_update(self, operation): # type: (Update) -> None | | def _execute_update(self, operation: Update) -> None: |
| source = operation.initial_package | | source = operation.initial_package |
| target = operation.target_package | | target = operation.target_package |
| | |
| if operation.skipped: | | if operation.skipped: |
| if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): | | if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Skipping <c1>{}</c1> (<c2>{}</c2>) {}".format( | | f" - Skipping <c1>{target.pretty_name}</c1> " |
| target.pretty_name, | | f"(<c2>{target.full_pretty_version}</c2>) {operation.skip_re |
| target.full_pretty_version, | | ason}" |
| operation.skip_reason, | | |
| ) | | |
| ) | | ) |
| | |
| return | | return |
| | |
| if self._execute_operations or self.is_dry_run(): | | if self._execute_operations or self.is_dry_run(): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Updating <c1>{}</c1> (<c2>{}</c2> -> <c2>{}</c2>)".format( | | f" - Updating <c1>{target.pretty_name}</c1>" |
| target.pretty_name, | | f" (<c2>{source.full_pretty_version}</c2> ->" |
| source.full_pretty_version, | | f" <c2>{target.full_pretty_version}</c2>)" |
| target.full_pretty_version, | | |
| ) | | |
| ) | | ) |
| | |
| if not self._execute_operations: | | if not self._execute_operations: |
| return | | return |
| | |
| self._installer.update(source, target) | | self._installer.update(source, target) |
| | |
|
| def _execute_uninstall(self, operation): # type: (Uninstall) -> None | | def _execute_uninstall(self, operation: Uninstall) -> None: |
| | target = operation.package |
| if operation.skipped: | | if operation.skipped: |
| if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): | | if self.is_verbose() and (self._execute_operations or self.is_dry_ru
n()): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Not removing <c1>{}</c1> (<c2>{}</c2>) {}".format( | | f" - Not removing <c1>{target.pretty_name}</c1>" |
| operation.package.pretty_name, | | f" (<c2>{target.pretty_version}</c2>) {operation.skip_reason |
| operation.package.full_pretty_version, | | }" |
| operation.skip_reason, | | |
| ) | | |
| ) | | ) |
| | |
| return | | return |
| | |
| if self._execute_operations or self.is_dry_run(): | | if self._execute_operations or self.is_dry_run(): |
| self._io.write_line( | | self._io.write_line( |
|
| " - Removing <c1>{}</c1> (<c2>{}</c2>)".format( | | f" - Removing <c1>{target.pretty_name}</c1>" |
| operation.package.pretty_name, operation.package.full_pretty | | f" (<c2>{target.pretty_version}</c2>)" |
| _version | | |
| ) | | |
| ) | | ) |
| | |
| if not self._execute_operations: | | if not self._execute_operations: |
| return | | return |
| | |
| self._installer.remove(operation.package) | | self._installer.remove(operation.package) |
| | |
|
| def _populate_local_repo(self, local_repo, ops): | | def _populate_lockfile_repo( |
| | self, repo: LockfileRepository, ops: Sequence[Operation] |
| | ) -> None: |
| for op in ops: | | for op in ops: |
| if isinstance(op, Uninstall): | | if isinstance(op, Uninstall): |
| continue | | continue |
| elif isinstance(op, Update): | | elif isinstance(op, Update): |
| package = op.target_package | | package = op.target_package |
| else: | | else: |
| package = op.package | | package = op.package |
| | |
|
| if not local_repo.has_package(package): | | if not repo.has_package(package): |
| local_repo.add_package(package) | | repo.add_package(package) |
| | |
| def _get_operations_from_lock( | | def _get_operations_from_lock( |
|
| self, locked_repository # type: Repository | | self, locked_repository: Repository |
| ): # type: (...) -> List[Operation] | | ) -> list[Operation]: |
| installed_repo = self._installed_repository | | installed_repo = self._installed_repository |
|
| ops = [] | | ops: list[Operation] = [] |
| | |
| extra_packages = self._get_extra_packages(locked_repository) | | extra_packages = self._get_extra_packages(locked_repository) |
| for locked in locked_repository.packages: | | for locked in locked_repository.packages: |
| is_installed = False | | is_installed = False |
| for installed in installed_repo.packages: | | for installed in installed_repo.packages: |
| if locked.name == installed.name: | | if locked.name == installed.name: |
| is_installed = True | | is_installed = True |
|
| if locked.category == "dev" and not self.is_dev_mode(): | | if locked.optional and locked.name not in extra_packages: |
| ops.append(Uninstall(locked)) | | |
| elif locked.optional and locked.name not in extra_packages: | | |
| # Installed but optional and not requested in extras | | # Installed but optional and not requested in extras |
| ops.append(Uninstall(locked)) | | ops.append(Uninstall(locked)) |
| elif locked.version != installed.version: | | elif locked.version != installed.version: |
| ops.append(Update(installed, locked)) | | ops.append(Update(installed, locked)) |
| | |
| # If it's optional and not in required extras | | # If it's optional and not in required extras |
| # we do not install | | # we do not install |
| if locked.optional and locked.name not in extra_packages: | | if locked.optional and locked.name not in extra_packages: |
| continue | | continue |
| | |
| op = Install(locked) | | op = Install(locked) |
| if is_installed: | | if is_installed: |
| op.skip("Already installed") | | op.skip("Already installed") |
| | |
| ops.append(op) | | ops.append(op) |
| | |
| return ops | | return ops |
| | |
|
| def _filter_operations( | | def _filter_operations(self, ops: Sequence[Operation], repo: Repository) -> |
| self, ops, repo | | None: |
| ): # type: (List[Operation], Repository) -> None | | |
| extra_packages = self._get_extra_packages(repo) | | extra_packages = self._get_extra_packages(repo) |
| for op in ops: | | for op in ops: |
| if isinstance(op, Update): | | if isinstance(op, Update): |
| package = op.target_package | | package = op.target_package |
| else: | | else: |
| package = op.package | | package = op.package |
| | |
| if op.job_type == "uninstall": | | if op.job_type == "uninstall": |
| continue | | continue |
| | |
| if not self._env.is_valid_for_marker(package.marker): | | if not self._env.is_valid_for_marker(package.marker): |
| op.skip("Not needed for the current environment") | | op.skip("Not needed for the current environment") |
| continue | | continue |
| | |
| if self._update: | | if self._update: |
| extras = {} | | extras = {} |
|
| for extra, deps in self._package.extras.items(): | | for extra, dependencies in self._package.extras.items(): |
| extras[extra] = [dep.name for dep in deps] | | extras[extra] = [dependency.name for dependency in dependenc |
| | ies] |
| else: | | else: |
| extras = {} | | extras = {} |
| for extra, deps in self._locker.lock_data.get("extras", {}).item
s(): | | for extra, deps in self._locker.lock_data.get("extras", {}).item
s(): |
| extras[extra] = [dep.lower() for dep in deps] | | extras[extra] = [dep.lower() for dep in deps] |
| | |
| # If a package is optional and not requested | | # If a package is optional and not requested |
| # in any extra we skip it | | # in any extra we skip it |
|
| if package.optional: | | if package.optional and package.name not in extra_packages: |
| if package.name not in extra_packages: | | op.skip("Not required") |
| op.skip("Not required") | | |
| | | |
| # If the package is a dev package and dev packages | | |
| # are not requested, we skip it | | |
| if package.category == "dev" and not self.is_dev_mode(): | | |
| op.skip("Dev dependencies not requested") | | |
| | |
|
| def _get_extra_packages(self, repo): # type: (Repository) -> List[str] | | def _get_extra_packages(self, repo: Repository) -> list[str]: |
| """ | | """ |
| Returns all package names required by extras. | | Returns all package names required by extras. |
| | |
| Maybe we just let the solver handle it? | | Maybe we just let the solver handle it? |
| """ | | """ |
|
| | extras: dict[str, list[str]] |
| if self._update: | | if self._update: |
| extras = {k: [d.name for d in v] for k, v in self._package.extras.it
ems()} | | extras = {k: [d.name for d in v] for k, v in self._package.extras.it
ems()} |
| else: | | else: |
| extras = self._locker.lock_data.get("extras", {}) | | extras = self._locker.lock_data.get("extras", {}) |
| | |
| return list(get_extra_package_names(repo.packages, extras, self._extras)
) | | return list(get_extra_package_names(repo.packages, extras, self._extras)
) |
| | |
|
| def _get_installer(self): # type: () -> BaseInstaller | | def _get_installer(self) -> BaseInstaller: |
| return PipInstaller(self._env, self._io, self._pool) | | return PipInstaller(self._env, self._io, self._pool) |
| | |
|
| def _get_installed(self): # type: () -> InstalledRepository | | def _get_installed(self) -> InstalledRepository: |
| return InstalledRepository.load(self._env) | | return InstalledRepository.load(self._env) |
| | | |
| End of changes. 80 change blocks. |
|---|
| 181 lines changed or deleted | | 188 lines changed or added |
|---|
"Fossies" - the Fresh Open Source Software Archive 