Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ classifiers = [
requires-python = ">=3.10"
dependencies=[
"httpx",
"packaging >=23",
"packaging >=26", # TODO 26.1, hopefully
"typer >=0.12.1",
"click >= 8.2",
# installers
Expand Down
49 changes: 35 additions & 14 deletions src/pip_deepfreeze/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from .pip import Installer, InstallerFlavor
from .pyproject_toml import load_pyproject_toml
from .sanity import check_env
from .sync import sync as sync_operation
from .sync import LockFormat, lock_and_sync, sync as sync_operation
from .tree import tree as tree_operation
from .utils import comma_split, increase_verbosity, log_debug, log_error, log_warning

Expand Down Expand Up @@ -61,6 +61,10 @@ def sync(
),
),
] = None,
lock_format: Annotated[
LockFormat,
typer.Option(),
] = LockFormat.requirements_txt,
uninstall_unneeded: Annotated[
bool | None,
typer.Option(
Expand Down Expand Up @@ -116,26 +120,43 @@ def sync(

Install/reinstall the project. Install/update dependencies to the
latest allowed version according to pinned dependencies in
requirements.txt or constraints in constraints.txt/requirements.txt.in. On demand
requirements.txt/pylock.toml or constraints in
constraints.txt/requirements.txt.in. On demand
update of dependencies to to the latest version that matches
constraints. Optionally uninstall unneeded dependencies.
"""
if build_contraints:
log_warning(
"--build-contraints is deprecated, use --build-constraints instead."
)
sync_operation(
Installer.create(flavor=installer, python=ctx.obj.python),
ctx.obj.python,
upgrade_all,
comma_split(to_upgrade),
extras=[canonicalize_name(extra) for extra in comma_split(extras)],
uninstall_unneeded=uninstall_unneeded,
project_root=ctx.obj.project_root,
pre_sync_commands=pre_sync_commands or [],
post_sync_commands=post_sync_commands or [],
build_constraints=build_constraints or build_contraints,
)
if lock_format == LockFormat.requirements_txt:
sync_operation(
Installer.create(flavor=installer, python=ctx.obj.python),
ctx.obj.python,
upgrade_all,
comma_split(to_upgrade),
extras=[canonicalize_name(extra) for extra in comma_split(extras)],
uninstall_unneeded=uninstall_unneeded,
project_root=ctx.obj.project_root,
pre_sync_commands=pre_sync_commands or [],
post_sync_commands=post_sync_commands or [],
build_constraints=build_constraints or build_contraints,
)
elif lock_format == LockFormat.pylock_toml:
lock_and_sync(
Installer.create(flavor=installer, python=ctx.obj.python),
ctx.obj.python,
upgrade_all,
comma_split(to_upgrade),
extras=[canonicalize_name(extra) for extra in comma_split(extras)],
uninstall_unneeded=uninstall_unneeded,
project_root=ctx.obj.project_root,
pre_sync_commands=pre_sync_commands or [],
post_sync_commands=post_sync_commands or [],
build_constraints=build_constraints or build_contraints,
)
else:
raise NotImplementedError


@app.command()
Expand Down
1 change: 1 addition & 0 deletions src/pip_deepfreeze/env-info-json.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@

try:
warnings.filterwarnings(action="ignore", category=DeprecationWarning)
warnings.filterwarnings(action="ignore", category=UserWarning)
import pkg_resources # noqa
warnings.resetwarnings()
except ImportError:
Expand Down
107 changes: 107 additions & 0 deletions src/pip_deepfreeze/pip.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,25 @@ def has_metadata_cache(self) -> bool:
"""Whether the installer caches metadata preparation results."""
...

@abstractmethod
def lock(
self,
*,
python: str,
project_root: Path,
constraints: Path | None = None,
build_constraints: Path | None = None,
) -> None: ...

@abstractmethod
def sync(
self,
*,
python: str,
project_root: Path,
extras: Sequence[NormalizedName] | None,
) -> None: ...

@classmethod
def create(cls, flavor: InstallerFlavor, python: str) -> "Installer":
if flavor == InstallerFlavor.pip:
Expand Down Expand Up @@ -125,6 +144,27 @@ def freeze_cmd(self, python: str) -> list[str]:
def has_metadata_cache(self) -> bool:
return False

def lock(
self,
*,
python: str,
project_root: Path,
constraints: Path | None = None,
build_constraints: Path | None = None,
) -> None:
# TODO use pip lock -e .[extras] -c constraints --build-constraints
# build_constraints
raise NotImplementedError

def sync(
self,
*,
python: str,
project_root: Path,
extras: Sequence[NormalizedName] | None,
) -> None:
raise NotImplementedError


class UvpipInstaller(Installer):
def install_cmd(
Expand All @@ -148,6 +188,7 @@ def editable_install_cmd(
)
# https://github.com/astral-sh/uv/issues/5484
cmd.append(f"--refresh-package={project_name}")
cmd.append("--strict")
return cmd

def uninstall_cmd(self, python: str) -> list[str]:
Expand All @@ -159,6 +200,72 @@ def freeze_cmd(self, python: str) -> list[str]:
def has_metadata_cache(self) -> bool:
return True

def lock(
self,
*,
python: str,
project_root: Path,
constraints: Path | None = None,
build_constraints: Path | None = None,
) -> None:
"""Lock project to pylock.toml."""
pylock_tmp = get_temp_path_in_dir(project_root, "pylock.", suffix=".df.toml")
pylock_tmp.unlink() # because it's empty and uv will try to parse it
cmd = [
*get_uv_cmd(),
"pip",
"compile",
"--python",
python,
"--format",
"pylock.toml",
"--output-file",
str(pylock_tmp),
"--custom-compile-command",
"pip-deepfreeze sync",
"--all-extras",
# XXX --all-groups
]
if constraints:
cmd.extend(["--constraints", str(constraints)])
if build_constraints:
cmd.extend(["--build-constraints", str(build_constraints)])
cmd.append("pyproject.toml")
log_debug(f"Running {shlex.join(cmd)}")
check_output(
cmd, cwd=project_root
) # use check_output because https://github.com/astral-sh/uv/issues/15309
pylock_tmp.rename(project_root / "pylock.toml")

def sync(
self,
*,
python: str,
project_root: Path,
extras: Sequence[NormalizedName] | None,
) -> None:
project_name = get_project_name(python, project_root)
sync_cmd = [
*get_uv_cmd(),
"--preview-feature=pylock",
"pip",
"sync",
"--python",
python,
"pylock.toml",
]
if extras:
for extra in extras:
sync_cmd.extend(("--extra", extra))
log_debug(f"Running {shlex.join(sync_cmd)}")
check_call(sync_cmd, cwd=project_root)
editable_install_cmd = [
*self.editable_install_cmd(python, project_root, project_name, extras),
"--exact", # --uninstall-unneeded always on
]
log_debug(f"Running {shlex.join(editable_install_cmd)}")
check_call(editable_install_cmd, cwd=project_root)


def pip_upgrade_project(
installer: Installer,
Expand Down
89 changes: 89 additions & 0 deletions src/pip_deepfreeze/pylock.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
from collections.abc import Iterable, Iterator
from pathlib import Path

from packaging.pylock import (
Package,
PackageArchive,
PackageDirectory,
PackageSdist,
PackageVcs,
PackageWheel,
Pylock,
)

from .compat import tomllib
from .python_env import get_python_environment_and_tags


def _dist_url(dist: PackageArchive | PackageVcs) -> str:
if dist.path:
return Path(dist.path).absolute().as_uri()
elif dist.url:
return dist.url
else:
raise ValueError("Distribution must have either a path or URL.")


def _package_vcs_url(dist: PackageVcs) -> str:
url = f"{dist.type}+{_dist_url(dist)}@{dist.commit_id}"
if dist.subdirectory:
url += f"#subdirectory={dist.subdirectory}"
return url


def _pylock_packages_to_requirements(
packages: Iterable[
tuple[
Package,
PackageArchive
| PackageSdist
| PackageWheel
| PackageVcs
| PackageDirectory,
]
],
) -> Iterator[str]:
for package, package_dist in packages:
if isinstance(package_dist, PackageWheel):
yield f"{package.name}=={package.version}"
elif isinstance(package_dist, PackageSdist):
yield f"{package.name}=={package.version}"
elif isinstance(package_dist, PackageVcs):
yield f"{package.name} @ {_package_vcs_url(package_dist)}"
elif isinstance(package_dist, PackageArchive):
yield f"{package.name} @ {_dist_url(package_dist)}"
elif isinstance(package_dist, PackageDirectory):
yield f"{'-e ' if package_dist.editable else ''}{package_dist.path}"
else:
raise NotImplementedError(
f"Unsupported package distribution type: {type(package_dist)}"
)


def pylock_to_requirements_txt(
python: str,
pylock_path: Path,
requirements_txt_path: Path,
) -> None:
"""Convert a pylock.toml file to a requirements.txt string.

This selects for all extras and dependency groups.
"""
if not pylock_path.is_file():
requirements = ""
else:
environment, tags = get_python_environment_and_tags(python)
pylock = Pylock.from_dict(
tomllib.loads(pylock_path.read_text(encoding="utf-8"))
)
requirements = "\n".join(
_pylock_packages_to_requirements(
pylock.select(
environment=environment,
tags=tags,
extras=pylock.extras,
dependency_groups=pylock.dependency_groups,
)
)
)
requirements_txt_path.write_text(requirements)
54 changes: 54 additions & 0 deletions src/pip_deepfreeze/python_env.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
import json
import os
from functools import lru_cache
from tempfile import TemporaryDirectory
from typing import cast

from packaging.markers import Environment
from packaging.tags import Tag, parse_tag

from .sanity import get_pip_command
from .utils import check_call, check_output

_SCRIPT = """\
import json
import sys

from packaging import markers, tags

json.dump(
{
"environment": markers.default_environment(),
"tags": [str(tag) for tag in tags.sys_tags()],
},
sys.stdout,
)
"""


@lru_cache
def get_python_environment_and_tags(python: str) -> tuple[Environment, list[Tag]]:
"""Get target python Environment and tags.

Run in a subprocess where packaging is installed.
"""
with TemporaryDirectory() as packaging_install_dir:
# first install packaging
check_call(
[
*get_pip_command(python),
"-q",
"install",
"--target",
packaging_install_dir,
"packaging",
]
)
res = check_output(
[python, "-c", _SCRIPT],
env=dict(os.environ, PYTHONPATH=packaging_install_dir),
)
res = json.loads(res)
return cast("Environment", res["environment"]), [
next(iter(parse_tag(tag))) for tag in res["tags"]
]
Loading
Loading