PK oL-vz z poetry/__init__.pyfrom poetry.console import Application
from .poetry import Poetry
__version__ = Poetry.VERSION
console = Application()
PK ftL@ poetry/__version__.py__version__ = '0.6.3-beta.4'
PK +nLV poetry/config.pyfrom pathlib import Path
from typing import Any
from .locations import CONFIG_DIR
from .utils.toml_file import TomlFile
from .utils.toml_file import TOMLFile
class Config:
def __init__(self, file: TomlFile):
self._file = file
if not self._file.exists():
self._raw_content = {}
self._content = TOMLFile([])
else:
self._raw_content = file.read(raw=True)
self._content = file.read()
@property
def name(self):
return str(self._file.path)
@property
def file(self):
return self._file
@property
def raw_content(self):
return self._raw_content
@property
def content(self):
return self._content
def setting(self, setting_name: str) -> Any:
"""
Retrieve a setting value.
"""
keys = setting_name.split('.')
config = self._raw_content
for key in keys:
if key not in config:
return None
config = config[key]
return config
def add_property(self, key, value):
keys = key.split('.')
config = self._content
for i, key in enumerate(keys):
if key not in config and i < len(keys) - 1:
config[key] = {}
if i == len(keys) - 1:
config[key] = value
break
config = config[key]
self.dump()
def remove_property(self, key):
keys = key.split('.')
config = self._content
for i, key in enumerate(keys):
if key not in config:
return
if i == len(keys) - 1:
del config[key]
break
config = config[key]
self.dump()
def dump(self):
self._file.write(self._content)
@classmethod
def create(cls, file, base_dir=None) -> 'Config':
if base_dir is None:
base_dir = CONFIG_DIR
file = TomlFile(Path(base_dir) / file)
return cls(file)
PK eVLj
% % poetry/console/__init__.pyfrom .application import Application
PK |ipL=2
poetry/console/application.pyimport os
from cleo import Application as BaseApplication
from cleo.inputs import ArgvInput
from cleo.outputs import ConsoleOutput
from poetry.io.raw_argv_input import RawArgvInput
from poetry.poetry import Poetry
from .commands import AboutCommand
from .commands import AddCommand
from .commands import BuildCommand
from .commands import CheckCommand
from .commands import ConfigCommand
from .commands import InstallCommand
from .commands import LockCommand
from .commands import NewCommand
from .commands import PublishCommand
from .commands import RemoveCommand
from .commands import RunCommand
from .commands import ShowCommand
from .commands import UpdateCommand
from .commands.debug import DebugResolveCommand
class Application(BaseApplication):
def __init__(self):
super().__init__('Poetry', Poetry.VERSION)
self._poetry = None
self._skip_io_configuration = False
@property
def poetry(self) -> Poetry:
if self._poetry is not None:
return self._poetry
self._poetry = Poetry.create(os.getcwd())
return self._poetry
def reset_poetry(self) -> None:
self._poetry = None
def run(self, i=None, o=None) -> int:
if i is None:
i = ArgvInput()
if o is None:
o = ConsoleOutput()
name = i.get_first_argument()
if name == 'run':
self._skip_io_configuration = True
i = RawArgvInput()
return super().run(i, o)
def do_run(self, i, o):
name = self.get_command_name(i)
if name != 'run':
return super().do_run(i, o)
command = self.find(name)
self._running_command = command
status_code = command.run(i, o)
self._running_command = None
return status_code
def configure_io(self, i, o):
if self._skip_io_configuration:
return
super().configure_io(i, o)
def get_default_commands(self) -> list:
commands = super(Application, self).get_default_commands()
commands += [
AboutCommand(),
AddCommand(),
BuildCommand(),
CheckCommand(),
ConfigCommand(),
InstallCommand(),
LockCommand(),
NewCommand(),
PublishCommand(),
RemoveCommand(),
RunCommand(),
ShowCommand(),
UpdateCommand(),
]
# Debug commands
commands += [
DebugResolveCommand(),
]
return commands
PK sipL>\c # poetry/console/commands/__init__.pyfrom .about import AboutCommand
from .add import AddCommand
from .build import BuildCommand
from .check import CheckCommand
from .config import ConfigCommand
from .install import InstallCommand
from .lock import LockCommand
from .new import NewCommand
from .publish import PublishCommand
from .remove import RemoveCommand
from .run import RunCommand
from .show import ShowCommand
from .update import UpdateCommand
PK oL0 poetry/console/commands/about.pyfrom .command import Command
class AboutCommand(Command):
"""
Short information about Poetry.
about
"""
def handle(self):
self.line("""Poetry - Package Management for Python
Poetry is a dependency manager tracking local dependencies of your projects and libraries.
See https://github.com/sdispater/poetry> for more information.
""")
PK
RpL-7 7 poetry/console/commands/add.pyimport re
from typing import List
from typing import Tuple
from poetry.installation import Installer
from poetry.semver.version_parser import VersionParser
from poetry.version.version_selector import VersionSelector
from .venv_command import VenvCommand
class AddCommand(VenvCommand):
"""
Add a new dependency to pyproject.toml>.
add
{ name* : Packages to add. }
{--D|dev : Add package as development dependency. }
{--optional : Add as an optional dependency. }
{--dry-run : Outputs the operations but will not execute anything
(implicitly enables --verbose). }
"""
help = """The add command adds required packages to your pyproject.toml> and installs them.
If you do not specify a version constraint, poetry will choose a suitable one based on the available package versions.
"""
def handle(self):
packages = self.argument('name')
is_dev = self.option('dev')
section = 'dependencies'
if is_dev:
section = 'dev-dependencies'
original_content = self.poetry.file.read()
content = self.poetry.file.read()
poetry_content = content['tool']['poetry']
for name in packages:
for key in poetry_content[section]:
if key.lower() == name.lower():
raise ValueError(f'Package {name} is already present')
requirements = self._determine_requirements(packages)
requirements = self._format_requirements(requirements)
# validate requirements format
parser = VersionParser()
for constraint in requirements.values():
parser.parse_constraints(constraint)
for name, constraint in requirements.items():
poetry_content[section][name] = constraint
# Write new content
self.poetry.file.write(content)
# Cosmetic new line
self.line('')
# Update packages
self.reset_poetry()
installer = Installer(
self.output,
self.venv,
self.poetry.package,
self.poetry.locker,
self.poetry.pool
)
installer.dry_run(self.option('dry-run'))
installer.update(True)
installer.whitelist(requirements)
try:
status = installer.run()
except Exception:
self.poetry.file.write(original_content)
raise
if status != 0 or self.option('dry-run'):
# Revert changes
if not self.option('dry-run'):
self.error(
'\n'
'Addition failed, reverting pyproject.toml '
'to its original content.'
)
self.poetry.file.write(original_content)
return status
def _determine_requirements(self, requires: List[str]) -> List[str]:
if not requires:
return []
requires = self._parse_name_version_pairs(requires)
result = []
for requirement in requires:
if 'version' not in requirement:
# determine the best version automatically
name, version = self._find_best_version_for_package(
requirement['name']
)
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']
)
requirement['name'] = name
result.append(f'{requirement["name"]} {requirement["version"]}')
return result
def _find_best_version_for_package(self,
name,
required_version=None
) -> Tuple[str, str]:
selector = VersionSelector(self.poetry.pool)
package = selector.find_best_candidate(name, required_version)
if not package:
# TODO: find similar
raise ValueError(
f'Could not find a matching version of package {name}'
)
return (
package.pretty_name,
selector.find_recommended_require_version(package)
)
def _parse_name_version_pairs(self, pairs: list) -> list:
result = []
for i in range(len(pairs)):
pair = re.sub('^([^=: ]+)[=: ](.*)$', '\\1 \\2', pairs[i].strip())
pair = pair.strip()
if ' ' in pair:
name, version = pair.split(' ', 2)
result.append({
'name': name,
'version': version
})
else:
result.append({
'name': pair
})
return result
def _format_requirements(self, requirements: List[str]) -> dict:
requires = {}
requirements = self._parse_name_version_pairs(requirements)
for requirement in requirements:
requires[requirement['name']] = requirement['version']
return requires
PK oLv-s s poetry/console/commands/build.pyfrom .venv_command import VenvCommand
from poetry.masonry import Builder
class BuildCommand(VenvCommand):
"""
Builds a package, as a tarball and a wheel by default.
build
{ --f|format= : Limit the format to either wheel or sdist. }
"""
def handle(self):
fmt = 'all'
if self.option('format'):
fmt = self.option('format')
package = self.poetry.package
self.line(f'Building {package.pretty_name}> '
f'({package.version}>)')
builder = Builder(self.poetry, self.venv, self.output)
builder.build(fmt)
PK }pL0 0 poetry/console/commands/check.pyimport jsonschema
from .command import Command
class CheckCommand(Command):
"""
Checks the validity of the pyproject.toml file.
check
"""
def handle(self):
# Load poetry and display errors, if any
_ = self.poetry
self.info('All set!')
PK oLک) ) " poetry/console/commands/command.pyfrom cleo import Command as BaseCommand
from cleo.inputs import ListInput
from poetry.poetry import Poetry
from ..styles.poetry import PoetryStyle
class Command(BaseCommand):
@property
def poetry(self) -> Poetry:
return self.get_application().poetry
def reset_poetry(self) -> None:
self.get_application().reset_poetry()
def run(self, i, o) -> int:
"""
Initialize command.
"""
self.input = i
self.output = PoetryStyle(i, o)
return super(BaseCommand, self).run(i, o)
PK ݎoLpJi ! poetry/console/commands/config.pyimport json
import re
from poetry.config import Config
from .command import Command
TEMPLATE = """[settings]
[repositories]
"""
AUTH_TEMPLATE = """[http-basic]
"""
class ConfigCommand(Command):
"""
Sets/Gets config options.
config
{ key : Setting key. }
{ value?* : Setting value. }
{ --list : List configuration settings }
{ --unset : Unset configuration setting }
"""
help = """This command allows you to edit the poetry config settings and repositories..
To add a repository:
poetry repositories.foo https://bar.com/simple/
To remove a repository (repo is a short alias for repositories):
poetry --unset repo.foo
"""
def __init__(self):
super().__init__()
self._config = Config.create('config.toml')
self._auth_config = Config.create('auth.toml')
def initialize(self, i, o):
super().initialize(i, o)
# Create config file if it does not exist
if not self._config.file.exists():
self._config.file.parent.mkdir(parents=True, exist_ok=True)
self._config.file.write_text(TEMPLATE)
if not self._auth_config.file.exists():
self._auth_config.file.parent.mkdir(parents=True, exist_ok=True)
self._auth_config.file.write_text(AUTH_TEMPLATE)
def handle(self):
if self.option('list'):
self._list_configuration(self._config.raw_content)
return 0
setting_key = self.argument('key')
if not setting_key:
return 0
if self.argument('value') and self.option('unset'):
raise RuntimeError('You can not combine a setting value with --unset')
# show the value if no value is provided
if not self.argument('value') and not self.option('unset'):
m = re.match('^repos?(?:itories)?(?:\.(.+))?', self.argument('key'))
if m:
if not m.group(1):
value = {}
if self._config.setting('repositories') is not None:
value = self._config.setting('repositories')
else:
repo = self._config.setting(f'repositories.{m.group(1)}')
if repo is None:
raise ValueError(
f'There is no {m.group(1)} repository defined'
)
value = repo
self.line(str(value))
return 0
values = self.argument('value')
boolean_validator = lambda val: val in {'true', 'false', '1', '0'}
boolean_normalizer = lambda val: True if val in ['true', '1'] else False
unique_config_values = {
'settings.virtualenvs.create': (boolean_validator, boolean_normalizer)
}
if setting_key in unique_config_values:
if self.option('unset'):
return self._remove_single_value(setting_key)
return self._handle_single_value(
setting_key,
unique_config_values[setting_key],
values
)
# handle repositories
m = re.match('^repos?(?:itories)?(?:\.(.+))?', self.argument('key'))
if m:
if not m.group(1):
raise ValueError('You cannot remove the [repositories] section')
if self.option('unset'):
repo = self._config.setting(f'repositories.{m.group(1)}')
if repo is None:
raise ValueError(f'There is no {m.group(1)} repository defined')
self._config.remove_property(f'repositories.{m.group(1)}')
return 0
if len(values) == 1:
url = values[0]
self._config.add_property(f'repositories.{m.group(1)}.url', url)
return 0
raise ValueError(
'You must pass the url. '
'Example: poetry config repositories.foo https://bar.com'
)
# handle auth
m = re.match('^(http-basic)\.(.+)', self.argument('key'))
if m:
if self.option('unset'):
if not self._auth_config.setting(f'{m.group(1)}.{m.group(2)}'):
raise ValueError(
f'There is no {m.group(2)} {m.group(1)} defined'
)
self._auth_config.remove_property(f'{m.group(1)}.{m.group(2)}')
return 0
if m.group(1) == 'http-basic':
if len(values) == 1:
username = values[0]
# Only username, so we prompt for password
password = self.secret('Password:')
elif len(values) != 2:
raise ValueError(f'Expected one or two arguments '
f'(username, password), got {len(values)}')
else:
username = values[0]
password = values[1]
self._auth_config.add_property(
f'{m.group(1)}.{m.group(2)}', {
'username': username,
'password': password
}
)
return 0
raise ValueError(f'Setting {self.argument("key")} does not exist')
def _handle_single_value(self, key, callbacks, values):
validator, normalizer = callbacks
if len(values) > 1:
raise RuntimeError('You can only pass one value.')
value = values[0]
if not validator(value):
raise RuntimeError(
f'"{value}" is an invalid value for {key}'
)
self._config.add_property(key, normalizer(value))
return 0
def _remove_single_value(self, key):
self._config.remove_property(key)
return 0
def _list_configuration(self, contents, k=None):
orig_k = k
for key, value in contents.items():
if k is None and key not in ['config', 'repositories']:
continue
if isinstance(value, dict) or key == 'repositories' and k is None:
if k is None:
k = ''
k += re.sub('^config\.', '', key + '.')
self._list_configuration(value, k=k)
k = orig_k
continue
if isinstance(value, list):
value = [
json.dumps(val) if isinstance(val, list) else val
for val in value
]
value = f'[{", ".join(value)}]'
value = json.dumps(value)
self.line(f'[{(k or "") + key}] {value}')
PK *MpLt) ) ) poetry/console/commands/debug/__init__.pyfrom .resolve import DebugResolveCommand
PK !OpL3W:T5 5 ( poetry/console/commands/debug/resolve.pyimport re
from typing import List
from poetry.packages import Dependency
from poetry.puzzle import Solver
from poetry.repositories.repository import Repository
from poetry.semver.version_parser import VersionParser
from ..command import Command
class DebugResolveCommand(Command):
"""
Debugs dependency resolution.
debug:resolve
{ package?* : packages to resolve. }
"""
def handle(self):
packages = self.argument('package')
if not packages:
package = self.poetry.package
dependencies = package.requires + package.dev_requires
else:
requirements = self._determine_requirements(packages)
requirements = self._format_requirements(requirements)
# validate requirements format
parser = VersionParser()
for constraint in requirements.values():
parser.parse_constraints(constraint)
dependencies = []
for name, constraint in requirements.items():
dependencies.append(
Dependency(name, constraint)
)
solver = Solver(
self.poetry.package,
self.poetry.pool,
Repository(),
self.output
)
ops = solver.solve(dependencies)
self.line('')
self.line('Resolution results:')
self.line('')
for op in ops:
package = op.package
self.line(f' - {package.name} '
f'({package.version})')
def _determine_requirements(self, requires: List[str]) -> List[str]:
if not requires:
return []
requires = self._parse_name_version_pairs(requires)
result = []
for requirement in requires:
if 'version' not in requirement:
requirement['version'] = '*'
result.append(f'{requirement["name"]} {requirement["version"]}')
return result
def _parse_name_version_pairs(self, pairs: list) -> list:
result = []
for i in range(len(pairs)):
pair = re.sub('^([^=: ]+)[=: ](.*)$', '\\1 \\2', pairs[i].strip())
pair = pair.strip()
if ' ' in pair:
name, version = pair.split(' ', 2)
result.append({
'name': name,
'version': version
})
else:
result.append({
'name': pair
})
return result
def _format_requirements(self, requirements: List[str]) -> dict:
requires = {}
requirements = self._parse_name_version_pairs(requirements)
for requirement in requirements:
requires[requirement['name']] = requirement['version']
return requires
PK "RpL&ʗ " poetry/console/commands/install.pyfrom poetry.installation import Installer
from .venv_command import VenvCommand
class InstallCommand(VenvCommand):
"""
Installs the project dependencies.
install
{ --no-dev : Do not install dev dependencies. }
{ --dry-run : Outputs the operations but will not execute anything
(implicitly enables --verbose). }
{ --E|extras=* : Extra sets of dependencies to install
(multiple values allowed). }
"""
help = """The install command reads the pyproject.toml> file from
the current directory, processes it, and downloads and installs all the
libraries and dependencies outlined in that file. If the file does not
exist it will look for pyproject.toml> and do the same.
poetry install
"""
def handle(self):
installer = Installer(
self.output,
self.venv,
self.poetry.package,
self.poetry.locker,
self.poetry.pool
)
installer.extras(self.option('extras'))
installer.dev_mode(not self.option('no-dev'))
installer.dry_run(self.option('dry-run'))
installer.verbose(self.option('verbose'))
return installer.run()
PK 'RpL+3ʯ poetry/console/commands/lock.pyfrom poetry.installation import Installer
from .venv_command import VenvCommand
class LockCommand(VenvCommand):
"""
Locks the project dependencies.
lock
"""
help = """The lock command reads the pyproject.toml> file from
the current directory, processes it, and locks the depdencies in the pyproject.lock> file.
poetry lock
"""
def handle(self):
installer = Installer(
self.output,
self.venv,
self.poetry.package,
self.poetry.locker,
self.poetry.pool
)
installer.update(True)
installer.execute_operations(False)
return installer.run()
PK QwsLvS S poetry/console/commands/new.pyfrom pathlib import Path
from poetry.layouts import layout
from .command import Command
class NewCommand(Command):
"""
Creates a new Python project at
new
{ path : The path to create the project at. }
{ --name : Set the resulting package name. }
"""
def handle(self):
layout_ = layout('standard')
path = Path.cwd() / Path(self.argument('path'))
name = self.option('name')
if not name:
name = path.name
if path.exists():
if list(path.glob('*')):
# Directory is not empty. Aborting.
raise RuntimeError(
'Destination {}> '
'exists and is not empty'.format(
path
)
)
readme_format = 'rst'
layout_ = layout_(name, '0.1.0', readme_format=readme_format)
layout_.create(path)
self.line(
'Created package {}> in {}>'
.format(name, path.relative_to(Path.cwd()))
)
PK oLp*U " poetry/console/commands/publish.pyfrom poetry.masonry.publishing.publisher import Publisher
from .command import Command
class PublishCommand(Command):
"""
Publishes a package to a remote repository.
publish
{ --r|repository= : The repository to publish the package to. }
{ --no-build : Do not build the package before publishing. }
"""
help = """The publish command builds and uploads the package to a remote repository.
By default, it will upload to PyPI but if you pass the --repository option it will
upload to it instead.
The --repository option should match the name of a configured repository using
the config command.
"""
def handle(self):
# Building package first, unless told otherwise
if not self.option('no-build'):
self.call('build')
self.line('')
publisher = Publisher(self.poetry, self.output)
publisher.publish(self.option('repository'))
PK RpLɻp p ! poetry/console/commands/remove.pyfrom poetry.installation import Installer
from .venv_command import VenvCommand
class RemoveCommand(VenvCommand):
"""
Removes a package from the project dependencies.
remove
{ packages* : Packages that should be removed. }
{--D|dev : Removes a package from the development dependencies. }
{--dry-run : Outputs the operations but will not execute anything
(implicitly enables --verbose). }
"""
help = """The remove command removes a package from the current
list of installed packages
poetry remove"""
def handle(self):
packages = self.argument('packages')
is_dev = self.option('dev')
original_content = self.poetry.file.read()
content = self.poetry.file.read()
poetry_content = content['tool']['poetry']
section = 'dependencies'
if is_dev:
section = 'dev-dependencies'
# Deleting entries
requirements = {}
for name in packages:
found = False
for key in poetry_content[section]:
if key.lower() == name.lower():
found = True
requirements[name] = poetry_content[section][name]
break
if not found:
raise ValueError(f'Package {name} not found')
for key in requirements:
del poetry_content[section][key]
# Write the new content back
self.poetry.file.write(content)
# Update packages
self.reset_poetry()
installer = Installer(
self.output,
self.venv,
self.poetry.package,
self.poetry.locker,
self.poetry.pool
)
installer.dry_run(self.option('dry-run'))
installer.update(True)
installer.whitelist(requirements)
try:
status = installer.run()
except Exception:
self.poetry.file.write(original_content)
raise
if status != 0 or self.option('dry-run'):
# Revert changes
if not self.option('dry-run'):
self.error(
'\n'
'Removal failed, reverting pyproject.toml '
'to its original content.'
)
self.poetry.file.write(original_content)
return status
PK m~oLIYE poetry/console/commands/run.pyfrom .venv_command import VenvCommand
class RunCommand(VenvCommand):
"""
Runs a command in the appropriate environment.
run
{ args* : The command and arguments/options to run. }
"""
def handle(self):
args = self.argument('args')
venv = self.venv
return venv.exec(*args)
def merge_application_definition(self, merge_args=True):
if self._application is None \
or (self._application_definition_merged
and (self._application_definition_merged_with_args or not merge_args)):
return
if merge_args:
current_arguments = self._definition.get_arguments()
self._definition.set_arguments(self._application.get_definition().get_arguments())
self._definition.add_arguments(current_arguments)
self._application_definition_merged = True
if merge_args:
self._application_definition_merged_with_args = True
PK [oLK7p! p! poetry/console/commands/show.pyfrom poetry.semver import statisfies
from poetry.version.version_selector import VersionSelector
from .venv_command import VenvCommand
class ShowCommand(VenvCommand):
"""
Shows information about packages.
show
{ package? : Package to inspect. }
{ --t|tree : List the dependencies as a tree. }
{ --l|latest : Show the latest version. }
{ --o|outdated : Show the latest version
but only for packages that are outdated. }
"""
help = """The show command displays detailed information about a package, or
lists all packages available."""
colors = [
'green',
'yellow',
'cyan',
'magenta',
'blue',
]
def handle(self):
package = self.argument('package')
if self.option('tree'):
self.init_styles()
if self.option('outdated'):
self.input.set_option('latest', True)
installed_repo = self.poetry.locker.locked_repository(True)
# Show tree view if requested
if self.option('tree') and not package:
requires = self.poetry.package.requires + self.poetry.package.dev_requires
packages = installed_repo.packages
for package in packages:
for require in requires:
if package.name == require.name:
self.display_package_tree(package, installed_repo)
break
return 0
table = self.table(style='compact')
table.get_style().set_vertical_border_char('')
locked_packages = installed_repo.packages
if package:
pkg = None
for locked in locked_packages:
if package.lower() == locked.name:
pkg = locked
break
if not pkg:
raise ValueError(f'Package {package} not found')
if self.option('tree'):
self.display_package_tree(pkg, installed_repo)
return 0
rows = [
['name>', f' : {pkg.pretty_name}>'],
['version>', f' : {pkg.pretty_version}>'],
['description>', f' : {pkg.description}'],
]
table.add_rows(rows)
table.render()
if pkg.requires:
self.line('')
self.line('dependencies')
for dependency in pkg.requires:
self.line(f' - {dependency.pretty_name} '
f'{dependency.pretty_constraint}>')
return 0
show_latest = self.option('latest')
terminal = self.get_application().terminal
width = terminal.width
name_length = version_length = latest_length = 0
latest_packages = {}
# Computing widths
for locked in locked_packages:
name_length = max(name_length, len(locked.pretty_name))
version_length = max(version_length, len(locked.full_pretty_version))
if show_latest:
latest = self.find_latest_package(locked)
if not latest:
latest = locked
latest_packages[locked.pretty_name] = latest
latest_length = max(latest_length, len(latest.full_pretty_version))
write_version = name_length + version_length + 3 <= width
write_latest = name_length + version_length + latest_length + 3 <= width
write_description = name_length + version_length + latest_length + 24 <= width
for locked in locked_packages:
line = f'{locked.pretty_name:{name_length}}>'
if write_version:
line += f' {locked.full_pretty_version:{version_length}}'
if show_latest and write_latest:
latest = latest_packages[locked.pretty_name]
update_status = self.get_update_status(latest, locked)
color = 'green'
if update_status == 'semver-safe-update':
color = 'red'
elif update_status == 'update-possible':
color = 'yellow'
line += f' {latest.version:{latest_length}}>'
if self.option('outdated') and update_status == 'up-to-date':
continue
if write_description:
description = locked.description
remaining = width - name_length - version_length - 4
if show_latest:
remaining -= latest_length
if len(locked.description) > remaining:
description = description[:remaining-3] + '...'
line += ' ' + description
self.line(line)
def display_package_tree(self, package, installed_repo):
self.write(f'{package.pretty_name}')
self.line(f' {package.pretty_version} {package.description}')
dependencies = package.requires
dependencies = sorted(dependencies, key=lambda x: x.name)
tree_bar = '├'
j = 0
total = len(dependencies)
for dependency in dependencies:
j += 1
if j == total:
tree_bar = '└'
level = 1
color = self.colors[level]
info = f'{tree_bar}── <{color}>{dependency.name}{color}> ' \
f'{dependency.pretty_constraint}'
self._write_tree_line(info)
tree_bar = tree_bar.replace('└', ' ')
packages_in_tree = [package.name, dependency.name]
self._display_tree(
dependency, installed_repo, packages_in_tree,
tree_bar, level + 1
)
def _display_tree(self,
dependency, installed_repo, packages_in_tree,
previous_tree_bar='├', level=1):
previous_tree_bar = previous_tree_bar.replace('├', '│')
dependencies = []
for package in installed_repo.packages:
if package.name == dependency.name:
dependencies = package.requires
break
dependencies = sorted(dependencies, key=lambda x: x.name)
tree_bar = previous_tree_bar + ' ├'
i = 0
total = len(dependencies)
for dependency in dependencies:
i += 1
current_tree = packages_in_tree
if i == total:
tree_bar = previous_tree_bar + ' └'
color_ident = level % len(self.colors)
color = self.colors[color_ident]
circular_warn = ''
if dependency.name in current_tree:
circular_warn = '(circular dependency aborted here)'
info = f'{tree_bar}── <{color}>{dependency.name}{color}> ' \
f'{dependency.pretty_constraint} {circular_warn}'
self._write_tree_line(info)
tree_bar = tree_bar.replace('└', ' ')
if dependency.name not in current_tree:
current_tree.append(dependency.name)
self._display_tree(
dependency, installed_repo, current_tree,
tree_bar, level + 1
)
def _write_tree_line(self, line):
if not self.output.is_decorated():
line = line.replace('└', '`-')
line = line.replace('├', '|-')
line = line.replace('──', '-')
line = line.replace('│', '|')
self.line(line)
def init_styles(self):
for color in self.colors:
self.set_style(color, color)
def find_latest_package(self, package):
# find the latest version allowed in this pool
if package.source_type == 'git':
return
name = package.name
selector = VersionSelector(self.poetry.pool)
return selector.find_best_candidate(name, f'>={package.version}')
def get_update_status(self, latest, package):
if latest.full_pretty_version == package.full_pretty_version:
return 'up-to-date'
constraint = '^' + package.version
if latest.version and statisfies(latest.version, constraint):
# It needs an immediate semver-compliant upgrade
return 'semver-safe-update'
# it needs an upgrade but has potential BC breaks so is not urgent
return 'update-possible'
PK QpLV ! poetry/console/commands/update.pyfrom poetry.installation import Installer
from .venv_command import VenvCommand
class UpdateCommand(VenvCommand):
"""
Update dependencies as according to the pyproject.toml> file.
update
{ packages?* : The packages to update. }
{ --no-dev : Do not install dev dependencies. }
{ --dry-run : Outputs the operations but will not execute anything
(implicitly enables --verbose). }
"""
def handle(self):
packages = self.argument('packages')
installer = Installer(
self.output,
self.venv,
self.poetry.package,
self.poetry.locker,
self.poetry.pool
)
if packages:
installer.whitelist({name: '*' for name in packages})
installer.dev_mode(not self.option('no-dev'))
installer.dry_run(self.option('dry-run'))
# Force update
installer.update(True)
return installer.run()
PK nL$- ' poetry/console/commands/venv_command.pyfrom poetry.utils.venv import Venv
from .command import Command
class VenvCommand(Command):
def __init__(self, name=None):
self._venv = None
super().__init__(name)
def initialize(self, i, o):
super().initialize(i, o)
self._venv = Venv.create(o, self.poetry.package.name)
if self._venv.is_venv() and o.is_verbose():
o.writeln(f'Using virtualenv: {self._venv.venv}>')
@property
def venv(self):
return self._venv
PK ȺZL ! poetry/console/styles/__init__.pyPK )nL=%.- - poetry/console/styles/poetry.pyfrom cleo.styles import CleoStyle
from cleo.styles import OutputStyle
class PoetryStyle(CleoStyle):
def __init__(self, i, o):
super().__init__(i, o)
self.output.get_formatter().add_style('warning', 'black', 'yellow')
self.output.get_formatter().add_style('question', 'blue')
def writeln(self, messages,
type=OutputStyle.OUTPUT_NORMAL,
verbosity=OutputStyle.VERBOSITY_NORMAL):
if self.output.verbosity >= verbosity:
super().writeln(messages, type=type)
def write(self, messages,
newline=False,
type=OutputStyle.OUTPUT_NORMAL,
verbosity=OutputStyle.VERBOSITY_NORMAL):
if self.output.verbosity >= verbosity:
super().write(messages, newline=newline, type=type)
PK }pL=N2c c poetry/exceptions.pyclass PoetryException(Exception):
pass
class InvalidProjectFile(PoetryException):
pass
PK ZL! ! poetry/installation/__init__.pyfrom .installer import Installer
PK b\L, % poetry/installation/base_installer.pyclass BaseInstaller:
def install(self, package):
raise NotImplementedError
def update(self, source, target):
raise NotImplementedError
def remove(self, package):
raise NotImplementedError
PK tsLr> > poetry/installation/installer.pyimport sys
from typing import List
from poetry.packages import Dependency
from poetry.packages import Locker
from poetry.packages import Package
from poetry.packages.constraints.platform_constraint import PlatformConstraint
from poetry.puzzle import Solver
from poetry.puzzle.operations import Install
from poetry.puzzle.operations import Uninstall
from poetry.puzzle.operations import Update
from poetry.puzzle.operations.operation import Operation
from poetry.repositories import Pool
from poetry.repositories import Repository
from poetry.repositories.installed_repository import InstalledRepository
from poetry.semver.constraints import Constraint
from poetry.semver.version_parser import VersionParser
from .base_installer import BaseInstaller
from .pip_installer import PipInstaller
class Installer:
def __init__(self,
io,
venv,
package: Package,
locker: Locker,
pool: Pool):
self._io = io
self._venv = venv
self._package = package
self._locker = locker
self._pool = pool
self._dry_run = False
self._update = False
self._verbose = False
self._write_lock = True
self._dev_mode = True
self._execute_operations = True
self._whitelist = {}
self._extras = []
self._installer = self._get_installer()
@property
def installer(self):
return self._installer
def run(self):
# Force update if there is no lock file present
if not self._update and not self._locker.is_locked():
self._update = True
if self.is_dry_run():
self.verbose(True)
self._write_lock = False
self._execute_operations = False
local_repo = Repository()
self._do_install(local_repo)
return 0
def dry_run(self, dry_run=True) -> 'Installer':
self._dry_run = dry_run
return self
def is_dry_run(self) -> bool:
return self._dry_run
def verbose(self, verbose=True) -> 'Installer':
self._verbose = verbose
return self
def is_verbose(self) -> bool:
return self._verbose
def dev_mode(self, dev_mode=True) -> 'Installer':
self._dev_mode = dev_mode
return self
def is_dev_mode(self) -> bool:
return self._dev_mode
def update(self, update=True) -> 'Installer':
self._update = update
return self
def is_updating(self) -> bool:
return self._update
def execute_operations(self, execute=True) -> 'Installer':
self._execute_operations = execute
return self
def whitelist(self, packages: dict) -> 'Installer':
self._whitelist = packages
return self
def extras(self, extras: list) -> 'Installer':
self._extras = extras
return self
def _do_install(self, local_repo):
locked_repository = Repository()
# initialize locked repo if we are installing from lock
if not self._update or (self._update and self._locker.is_locked()):
locked_repository = self._locker.locked_repository(True)
if self._update:
# Checking extras
for extra in self._extras:
if extra not in self._package.extras:
raise ValueError(f'Extra [{extra}] is not specified.')
self._io.writeln('Updating dependencies>')
fixed = []
# If the whitelist is enabled, packages not in it are fixed
# to the version specified in the lock
if self._whitelist:
# collect packages to fixate from root requirements
candidates = []
for package in locked_repository.packages:
candidates.append(package)
# fix them to the version in lock if they are not updateable
for candidate in candidates:
to_fix = True
for require in self._whitelist.keys():
if require == candidate.name:
to_fix = False
if to_fix:
fixed.append(
Dependency(candidate.name, candidate.version)
)
solver = Solver(
self._package,
self._pool,
locked_repository,
self._io
)
request = self._package.requires
request += self._package.dev_requires
ops = solver.solve(request, fixed=fixed)
else:
self._io.writeln('Installing dependencies from lock file>')
if not self._locker.is_fresh():
self._io.writeln(
''
'Warning: The lock file is not up to date with '
'the latest changes in pyproject.toml. '
'You may be getting outdated dependencies. '
'Run update to update them.'
''
)
for extra in self._extras:
if extra not in self._locker.lock_data.get('extras', {}):
raise ValueError(f'Extra [{extra}] is not specified.')
# If we are installing from lock
# Filter the operations by comparing it with what is
# currently installed
ops = self._get_operations_from_lock(locked_repository)
self._populate_local_repo(local_repo, ops, locked_repository)
# We need to filter operations so that packages
# not compatible with the current system,
# or optional and not requested, are dropped
self._filter_operations(ops, local_repo)
self._io.new_line()
# Execute operations
if not ops and (self._execute_operations or self._dry_run):
self._io.writeln('Nothing to install or update')
if ops and (self._execute_operations or self._dry_run):
installs = []
updates = []
uninstalls = []
skipped = []
for op in ops:
if op.skipped:
skipped.append(op)
continue
if op.job_type == 'install':
installs.append(
f'{op.package.pretty_name}'
f':{op.package.full_pretty_version}'
)
elif op.job_type == 'update':
updates.append(
f'{op.target_package.pretty_name}'
f':{op.target_package.full_pretty_version}'
)
elif op.job_type == 'uninstall':
uninstalls.append(
f'{op.package.pretty_name}'
)
self._io.new_line()
self._io.writeln(
'Package operations: '
f'{len(installs)}> install{"" if len(installs) == 1 else "s"}, '
f'{len(updates)}> update{"" if len(updates) == 1 else "s"}, '
f'{len(uninstalls)}> removal{"" if len(uninstalls) == 1 else "s"}'
f'{", {}> skipped".format(len(skipped)) if skipped and self.is_verbose() else ""}'
)
self._io.new_line()
# Writing lock before installing
if self._update and self._write_lock:
updated_lock = self._locker.set_lock_data(
self._package,
local_repo.packages
)
if updated_lock:
self._io.writeln('Writing lock file>')
self._io.writeln('')
for op in ops:
self._execute(op)
def _execute(self, operation: Operation) -> None:
"""
Execute a given operation.
"""
method = operation.job_type
getattr(self, f'_execute_{method}')(operation)
def _execute_install(self, operation: Install) -> None:
if operation.skipped:
if self.is_verbose() and (self._execute_operations or self.is_dry_run()):
self._io.writeln(
f' - Skipping {operation.package.pretty_name}> '
f'({operation.package.full_pretty_version}>) '
f'{operation.skip_reason}')
return
if self._execute_operations or self.is_dry_run():
self._io.writeln(
f' - Installing {operation.package.pretty_name}> '
f'({operation.package.full_pretty_version}>)'
)
if not self._execute_operations:
return
self._installer.install(operation.package)
def _execute_update(self, operation: Update) -> None:
source = operation.initial_package
target = operation.target_package
if operation.skipped:
if self.is_verbose() and (self._execute_operations or self.is_dry_run()):
self._io.writeln(
f' - Skipping {target.pretty_name}> '
f'({target.full_pretty_version}>) '
f'{operation.skip_reason}')
return
if self._execute_operations or self.is_dry_run():
self._io.writeln(
f' - Updating {target.pretty_name}> '
f'({source.pretty_version}>'
f' -> {target.pretty_version}>)'
)
if not self._execute_operations:
return
self._installer.update(source, target)
def _execute_uninstall(self, operation: Uninstall) -> None:
if self._execute_operations or self.is_dry_run():
self._io.writeln(
f' - Removing {operation.package.pretty_name}> '
f'({operation.package.full_pretty_version}>)'
)
if not self._execute_operations:
return
self._installer.remove(operation.package)
def _populate_local_repo(self, local_repo, ops, locked_repository):
# Add all locked packages from the lock and go from there
for package in locked_repository.packages:
local_repo.add_package(package)
# Now, walk through all operations and add/remove/update accordingly
for op in ops:
if isinstance(op, Update):
package = op.target_package
else:
package = op.package
acted_on = False
for pkg in local_repo.packages:
if pkg.name == package.name:
# The package we operate on is in the local repo
if op.job_type == 'update':
if pkg.version == package.version:
break
local_repo.remove_package(pkg)
local_repo.add_package(op.target_package)
elif op.job_type == 'uninstall':
local_repo.remove_package(op.package)
acted_on = True
if not acted_on:
local_repo.add_package(package)
def _get_operations_from_lock(self,
locked_repository: Repository
) -> List[Operation]:
installed_repo = InstalledRepository.load(self._venv)
ops = []
extra_packages = [
p.name
for p in self._get_extra_packages(locked_repository)
]
for locked in locked_repository.packages:
is_installed = False
for installed in installed_repo.packages:
if locked.name == installed.name:
is_installed = True
if locked.category == 'dev' and not self.is_dev_mode():
ops.append(Uninstall(locked))
elif locked.optional and locked.name not in extra_packages:
# Installed but optional and not requested in extras
ops.append(Uninstall(locked))
elif locked.version != installed.version:
ops.append(Update(
installed, locked
))
if not is_installed:
# If it's optional and not in required extras
# we do not install
if locked.optional and locked.name not in extra_packages:
continue
ops.append(Install(locked))
return ops
def _filter_operations(self, ops: List[Operation], repo: Repository) -> None:
extra_packages = [p.name for p in
self._get_extra_packages(repo)]
for op in ops:
if isinstance(op, Update):
package = op.target_package
else:
package = op.package
if op.job_type == 'uninstall':
continue
parser = VersionParser()
python = '.'.join([str(i) for i in self._venv.version_info[:3]])
if 'python' in package.requirements:
python_constraint = parser.parse_constraints(
package.requirements['python']
)
if not python_constraint.matches(Constraint('=', python)):
# Incompatible python versions
op.skip('Not needed for the current python version')
continue
if 'platform' in package.requirements:
platform_constraint = PlatformConstraint.parse(
package.requirements['platform']
)
if not platform_constraint.matches(
PlatformConstraint('=', sys.platform)
):
# Incompatible systems
op.skip('Not needed for the current platform')
continue
if self._update:
extras = {}
for extra, deps in self._package.extras.items():
extras[extra] = [dep.name for dep in deps]
else:
extras = {}
for extra, deps in self._locker.lock_data.get('extras', {}).items():
extras[extra] = [dep.lower() for dep in deps]
# If a package is optional and not requested
# in any extra we skip it
if package.optional:
if package.name not in extra_packages:
op.skip('Not required')
def _get_extra_packages(self, repo):
"""
Returns all packages required by extras.
Maybe we just let the solver handle it?
"""
if self._update:
extras = {
k: [d.name for d in v]
for k, v in self._package.extras.items()
}
else:
extras = self._locker.lock_data.get('extras', {})
extra_packages = []
for extra_name, packages in extras.items():
if extra_name not in self._extras:
continue
extra_packages += [Dependency(p, '*') for p in packages]
def _extra_packages(packages):
pkgs = []
for package in packages:
for pkg in repo.packages:
if pkg.name == package.name:
pkgs.append(package)
pkgs += _extra_packages(pkg.requires)
break
return pkgs
return _extra_packages(extra_packages)
def _get_installer(self) -> BaseInstaller:
return PipInstaller(self._venv, self._io)
PK _fL" % poetry/installation/noop_installer.pyfrom .base_installer import BaseInstaller
class NoopInstaller(BaseInstaller):
def __init__(self):
self._installs = []
self._updates = []
self._removals = []
@property
def installs(self):
return self._installs
@property
def updates(self):
return self._updates
@property
def removals(self):
return self._removals
def install(self, package) -> None:
self._installs.append(package)
def update(self, source, target) -> None:
self._updates.append((source, target))
def remove(self, package) -> None:
self._removals.append(package)
PK pnLӺ= = $ poetry/installation/pip_installer.pyimport os
import tempfile
from subprocess import CalledProcessError
from poetry.utils.venv import Venv
from .base_installer import BaseInstaller
class PipInstaller(BaseInstaller):
def __init__(self, venv: Venv, io):
self._venv = venv
self._io = io
def install(self, package, update=False):
args = ['install', '--no-deps']
if package.source_type == 'legacy' and package.source_url:
args += ['--index-url', package.source_url]
if update:
args.append('-U')
if package.hashes and not package.source_type:
# Format as a requirements.txt
# We need to create a requirements.txt file
# for each package in order to check hashes.
# This is far from optimal but we do not have any
# other choice since this is the only way for pip
# to verify hashes.
req = self.create_temporary_requirement(package)
args += ['-r', req]
try:
self.run(*args)
finally:
os.unlink(req)
else:
args.append(self.requirement(package))
self.run(*args)
def update(self, _, target):
self.install(target, update=True)
def remove(self, package):
try:
self.run('uninstall', package.name, '-y')
except CalledProcessError as e:
if 'not installed' in str(e):
return
raise
def run(self, *args, **kwargs) -> str:
return self._venv.run('pip', *args, **kwargs)
def requirement(self, package, formatted=False) -> str:
if formatted and not package.source_type == 'git':
req = f'{package.name}=={package.version}'
for h in package.hashes:
req += f' --hash sha256:{h}'
req += '\n'
return req
if package.source_type == 'git':
return f'git+{package.source_url}@{package.source_reference}' \
f'#egg={package.name}'
return f'{package.name}=={package.version}'
def create_temporary_requirement(self, package):
fd, name = tempfile.mkstemp('reqs.txt', f'{package.name}-{package.version}')
with open(fd, 'w') as f:
f.write(self.requirement(package, formatted=True))
return name
PK \LY poetry/io/__init__.pyfrom .null_io import NullIO
PK
nL|j j poetry/io/null_io.pyfrom cleo.inputs import ListInput
from cleo.outputs import NullOutput
from poetry.console.styles.poetry import PoetryStyle
class NullIO(PoetryStyle):
def __init__(self):
super().__init__(ListInput([]), NullOutput())
def is_quiet(self) -> bool:
return False
def is_verbose(self) -> bool:
return False
def is_very_verbose(self) -> bool:
return False
def is_debug(self) -> bool:
return False
def writeln(self, *args, **kwargs):
pass
def write(self, *args, **kwargs):
pass
def new_line(self, *args, **kwargs):
pass
PK loLLa7 7 poetry/io/raw_argv_input.pyimport sys
from cleo.inputs import ArgvInput
class RawArgvInput(ArgvInput):
def parse(self):
self._parsed = self._tokens
while True:
try:
token = self._parsed.pop(0)
except IndexError:
break
self.parse_argument(token)
PK fpL poetry/json/__init__.pyPK ]~pLrO3 &