python/private/pypi/pep508_deps.bzl

# Copyright 2025 The Bazel Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""This module is for implementing PEP508 compliant METADATA deps parsing.
"""

load("//python/private:normalize_name.bzl", "normalize_name")
load(":pep508_env.bzl", "create_env")
load(":pep508_evaluate.bzl", "evaluate")
load(":pep508_requirement.bzl", "requirement")

def deps(
        name,
        *,
        requires_dist,
        extras = [],
        excludes = [],
        include = []):
    """Parse the RequiresDist from wheel METADATA

    Args:
        name: {type}`str` the name of the wheel.
        requires_dist: {type}`list[str]` the list of RequiresDist lines from the
            METADATA file.
        excludes: {type}`list[str]` what packages should we exclude.
        include: {type}`list[str]` what packages should we exclude. If it is not
            specified, then we will include all deps from `requires_dist`.
        extras: {type}`list[str]` the requested extras to generate targets for.

    Returns:
        A struct with attributes:
        * deps: {type}`list[str]` dependencies to include unconditionally.
        * deps_select: {type}`dict[str, list[str]]` dependencies to include on particular
              subset of target platforms.
    """
    reqs = sorted(
        [requirement(r) for r in requires_dist],
        key = lambda x: "{}:{}:".format(x.name, sorted(x.extras), x.marker),
    )
    deps = {}
    deps_select = {}
    name = normalize_name(name)
    want_extras = _resolve_extras(name, reqs, extras)
    include = [normalize_name(n) for n in include]

    # drop self edges
    excludes = [name] + [normalize_name(x) for x in excludes]

    reqs_by_name = {}

    for req in reqs:
        if req.name_ in excludes:
            continue

        if include and req.name_ not in include:
            continue

        reqs_by_name.setdefault(req.name, []).append(req)

    for name, reqs in reqs_by_name.items():
        _add_reqs(
            deps,
            deps_select,
            normalize_name(name),
            reqs,
            extras = want_extras,
        )

    return struct(
        deps = sorted(deps),
        deps_select = {
            d: markers
            for d, markers in sorted(deps_select.items())
        },
    )

def _add(deps, deps_select, dep, markers = None):
    dep = normalize_name(dep)

    if not markers:
        deps[dep] = True
    elif len(markers) == 1:
        deps_select[dep] = markers[0]
    else:
        deps_select[dep] = "({})".format(") or (".join(sorted(markers)))

def _resolve_extras(self_name, reqs, extras):
    """Resolve extras which are due to depending on self[some_other_extra].

    Some packages may have cyclic dependencies resulting from extras being used, one example is
    `etils`, where we have one set of extras as aliases for other extras
    and we have an extra called 'all' that includes all other extras.

    Example: github.com/google/etils/blob/a0b71032095db14acf6b33516bca6d885fe09e35/pyproject.toml#L32.

    When the `requirements.txt` is generated by `pip-tools`, then it is likely that
    this step is not needed, but for other `requirements.txt` files this may be useful.

    NOTE @aignas 2023-12-08: the extra resolution is not platform dependent,
    but in order for it to become platform dependent we would have to have
    separate targets for each extra in extras.
    """

    # Resolve any extra extras due to self-edges, empty string means no
    # extras The empty string in the set is just a way to make the handling
    # of no extras and a single extra easier and having a set of {"", "foo"}
    # is equivalent to having {"foo"}.
    #
    # Use a dict as a set here to simplify operations.
    extras = {x: None for x in (extras or [""])}

    self_reqs = []
    for req in reqs:
        if req.name != self_name:
            continue

        if req.marker == None:
            # I am pretty sure we cannot reach this code as it does not
            # make sense to specify packages in this way, but since it is
            # easy to handle, lets do it.
            #
            # TODO @aignas 2023-12-08: add a test
            extras = extras | {x: None for x in req.extras}
        else:
            # process these in a separate loop
            self_reqs.append(req)

    for _ in range(10000):
        # handles packages with up to 10000 recursive extras
        new_extras = {}
        for req in self_reqs:
            if _evaluate_any(req, extras):
                new_extras.update({x: None for x in req.extras})
            else:
                continue

        num_extras_before = len(extras)
        extras = extras | new_extras
        num_extras_after = len(new_extras)

        if num_extras_before == num_extras_after:
            break

    # Poor mans set
    return sorted(extras)

def _evaluate_any(req, extras):
    env = create_env()
    for extra in extras:
        if evaluate(req.marker, env = env | {"extra": extra}):
            return True

    return False

def _add_reqs(deps, deps_select, dep, reqs, *, extras):
    for req in reqs:
        if not req.marker:
            _add(deps, deps_select, dep)
            return

    env = create_env()
    markers = {}
    found_unconditional = False
    for req in reqs:
        for x in extras:
            m = evaluate(req.marker, env = env | {"extra": x}, strict = False)
            if m == False:
                continue
            elif m == True:
                _add(deps, deps_select, dep)
                found_unconditional = True
                break
            else:
                markers[m] = None
                continue
        if found_unconditional:
            break

    if markers and not found_unconditional:
        _add(deps, deps_select, dep, sorted(markers))