Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Replace suffix_union with less general function #532

Merged
merged 5 commits into from
Oct 16, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
38 changes: 0 additions & 38 deletions conda_lock/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,6 @@
)


if typing.TYPE_CHECKING:
# Not in the release version of typeshed yet
from _typeshed import SupportsRichComparisonT # type: ignore

T = TypeVar("T")


Expand Down Expand Up @@ -79,40 +75,6 @@ def ordered_union(collections: Iterable[Iterable[T]]) -> List[T]:
return list({k: k for k in chain.from_iterable(collections)}.values())


def suffix_union(
collections: Iterable[Sequence["SupportsRichComparisonT"]],
) -> List["SupportsRichComparisonT"]:
"""Generates the union of sequence ensuring that they have a common suffix.

This is used to unify channels.

>>> suffix_union([[1], [2, 1], [3, 2, 1], [2, 1], [1]])
[3, 2, 1]

>>> suffix_union([[1], [2, 1], [4, 1]])
Traceback (most recent call last)
...
RuntimeError: [4, 1] is not a subset of [2, 1]

"""
from genericpath import commonprefix

result: List["SupportsRichComparisonT"] = []
for seq in collections:
if seq:
rev_priority = list(reversed(seq))
prefix = commonprefix([result, rev_priority]) # type: ignore
if len(result) == 0:
result = rev_priority
elif prefix[: len(rev_priority)] != result[: len(rev_priority)]:
raise ValueError(
f"{list(reversed(rev_priority))} is not a ordered subset of {list(reversed(result))}"
)
elif len(rev_priority) > len(result):
result = rev_priority
return list(reversed(result))


def relative_path(source: pathlib.Path, target: pathlib.Path) -> str:
"""
Get posix representation of the relative path from `source` to `target`.
Expand Down
38 changes: 36 additions & 2 deletions conda_lock/src_parser/aggregation.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,9 @@
from itertools import chain
from typing import Dict, List, Tuple

from conda_lock.common import ordered_union, suffix_union
from conda_lock.common import ordered_union
from conda_lock.errors import ChannelAggregationError
from conda_lock.models.channel import Channel
from conda_lock.models.lock_spec import Dependency, LockSpecification


Expand Down Expand Up @@ -37,7 +38,9 @@ def aggregate_lock_specs(
dependencies[platform] = list(unique_deps.values())

try:
channels = suffix_union(lock_spec.channels for lock_spec in lock_specs)
channels = unify_package_sources(
[lock_spec.channels for lock_spec in lock_specs]
)
except ValueError as e:
raise ChannelAggregationError(*e.args)

Expand All @@ -51,3 +54,34 @@ def aggregate_lock_specs(
lock_spec.allow_pypi_requests for lock_spec in lock_specs
),
)


def unify_package_sources(collections: List[List[Channel]]) -> List[Channel]:
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In #529 I would replace Channel here with:

PackageSource = TypeVar("PackageSource", Channel, PipRepository)

"""Unify the package sources from multiple lock specs.

To be able to merge the lock specs, the package sources must be compatible between
them. This means that between any two lock specs, the package sources must be
identical or one must be an extension of the other.

This allows us to use a superset of all of the package source lists in the
aggregated lock spec.

The following is allowed:

> unify_package_sources([[channel_two, channel_one], [channel_one]])
[channel_two, channel_one]

Whilst the following will fail:

> unify_package_sources([[channel_two, channel_one], [channel_three, channel_one]])

In the failing example, it is not possible to predictably decide which channel
to search first, `channel_two` or `channel_three`, so we error in this case.
"""
if not collections:
return []
result = max(collections, key=len)
for collection in collections:
if collection != result[-len(collection) :]:
jacksmith15 marked this conversation as resolved.
Show resolved Hide resolved
raise ValueError(f"{collection} is not an ordered subset of {result}")
return result