src/hatch/project/sources.py - master - niceSOFT/python3-hatchling

src/hatch/project/sources.py
491 строка17 KB
Cary Hawkins
Add sources to enable other types of local dependencies (#2313)
10 авг 2026, 06:24
10 авг 2026, 06:24a885803
from __future__ import annotations

from functools import cached_property
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from collections.abc import Iterable, Mapping

    from hatch.dep.core import Dependency


_TYPE_KEYS: tuple[str, ...] = ("path", "git", "url", "index", "workspace")


class Source:
    """
    Base class for all dependency sources defined under an environment's `sources` table.

    A source provides an alternative origin for a dependency at install time
    without altering the project's published metadata. Each environment plugin
    is responsible for translating sources into installer-specific arguments.
    """


class PathSource(Source):
    def __init__(self, *, path: str, editable: bool = True, subdirectory: str | None = None) -> None:
        self.path = path
        self.editable = editable
        self.subdirectory = subdirectory


class GitSource(Source):
    def __init__(
        self,
        *,
        git: str,
        rev: str | None = None,
        tag: str | None = None,
        branch: str | None = None,
        subdirectory: str | None = None,
    ) -> None:
        self.git = git
        self.rev = rev
        self.tag = tag
        self.branch = branch
        self.subdirectory = subdirectory

    @cached_property
    def reference(self) -> str | None:
        return self.rev or self.tag or self.branch


class UrlSource(Source):
    def __init__(self, *, url: str, subdirectory: str | None = None) -> None:
        self.url = url
        self.subdirectory = subdirectory


class IndexSource(Source):
    def __init__(self, *, index: str) -> None:
        self.index = index


class WorkspaceSource(Source):
    """
    A member of the current workspace.

    The actual install path is resolved by `tool.hatch.envs.<ENV_NAME>.workspace`.
    """


DEFAULT_ROOT_FIELD = "tool.hatch.sources"


def _check_str(value: Any, field_path: str) -> str:
    if not isinstance(value, str):
        message = f"Field `{field_path}` must be a string"
        raise TypeError(message)

    return value


def _check_optional_str(value: Any, field_path: str) -> str | None:
    if value is None:
        return None

    return _check_str(value, field_path)


def _parse_path_source(field_prefix: str, raw: dict[str, Any]) -> PathSource:
    path = _check_str(raw["path"], f"{field_prefix}.path")
    if not path:
        message = f"Field `{field_prefix}.path` cannot be an empty string"
        raise ValueError(message)

    editable = raw.get("editable", True)
    if not isinstance(editable, bool):
        message = f"Field `{field_prefix}.editable` must be a boolean"
        raise TypeError(message)

    subdirectory = _check_optional_str(raw.get("subdirectory"), f"{field_prefix}.subdirectory")

    return PathSource(path=path, editable=editable, subdirectory=subdirectory)


def _parse_git_source(field_prefix: str, raw: dict[str, Any]) -> GitSource:
    git = _check_str(raw["git"], f"{field_prefix}.git")
    if not git:
        message = f"Field `{field_prefix}.git` cannot be an empty string"
        raise ValueError(message)

    ref_kinds = [k for k in ("rev", "tag", "branch") if k in raw]
    if len(ref_kinds) > 1:
        message = f"Field `{field_prefix}` must define only one of: {', '.join(ref_kinds)}"
        raise ValueError(message)

    rev = _check_optional_str(raw.get("rev"), f"{field_prefix}.rev")
    tag = _check_optional_str(raw.get("tag"), f"{field_prefix}.tag")
    branch = _check_optional_str(raw.get("branch"), f"{field_prefix}.branch")
    subdirectory = _check_optional_str(raw.get("subdirectory"), f"{field_prefix}.subdirectory")

    return GitSource(git=git, rev=rev, tag=tag, branch=branch, subdirectory=subdirectory)


def _parse_url_source(field_prefix: str, raw: dict[str, Any]) -> UrlSource:
    url = _check_str(raw["url"], f"{field_prefix}.url")
    if not url:
        message = f"Field `{field_prefix}.url` cannot be an empty string"
        raise ValueError(message)

    subdirectory = _check_optional_str(raw.get("subdirectory"), f"{field_prefix}.subdirectory")

    return UrlSource(url=url, subdirectory=subdirectory)


def _parse_index_source(field_prefix: str, raw: dict[str, Any]) -> IndexSource:
    index = _check_str(raw["index"], f"{field_prefix}.index")
    if not index:
        message = f"Field `{field_prefix}.index` cannot be an empty string"
        raise ValueError(message)

    return IndexSource(index=index)


def _parse_workspace_source(field_prefix: str, raw: dict[str, Any]) -> WorkspaceSource:
    workspace_value = raw["workspace"]
    if not isinstance(workspace_value, bool):
        message = f"Field `{field_prefix}.workspace` must be a boolean"
        raise TypeError(message)

    if not workspace_value:
        message = f"Field `{field_prefix}.workspace` must be `true`"
        raise ValueError(message)

    return WorkspaceSource()


_SOURCE_PARSERS = {
    "path": _parse_path_source,
    "git": _parse_git_source,
    "url": _parse_url_source,
    "index": _parse_index_source,
    "workspace": _parse_workspace_source,
}


def parse_source(name: str, raw: Any, *, root_field: str = DEFAULT_ROOT_FIELD) -> Source:
    """
    Parse a single entry of a sources table. `root_field` is the location of the
    table being parsed, used in error messages, e.g. `tool.hatch.envs.test.sources`.
    """
    field_prefix = f"{root_field}.{name}"

    if isinstance(raw, str):
        if not raw:
            message = f"Field `{field_prefix}` cannot be an empty string"
            raise ValueError(message)

        return PathSource(path=raw)

    if not isinstance(raw, dict):
        message = f"Field `{field_prefix}` must be a string or a table"
        raise TypeError(message)

    found_types = [k for k in _TYPE_KEYS if k in raw]
    if not found_types:
        message = f"Field `{field_prefix}` must define exactly one of: {', '.join(_TYPE_KEYS)}"
        raise ValueError(message)

    if len(found_types) > 1:
        message = f"Field `{field_prefix}` must define only one of: {', '.join(found_types)}"
        raise ValueError(message)

    return _SOURCE_PARSERS[found_types[0]](field_prefix, raw)


def parse_sources(config: Any, *, root_field: str = DEFAULT_ROOT_FIELD) -> dict[str, Source]:
    """
    Parse an entire sources table like `[tool.hatch.envs.default.sources]`. Returns a mapping keyed by
    [normalized](https://peps.python.org/pep-0503/#normalized-names) project name.
    `root_field` is the location of the table being parsed, used in error messages.
    """
    from collections import defaultdict

    from hatch.utils.metadata import normalize_project_name

    if not isinstance(config, dict):
        message = f"Field `{root_field}` must be a table"
        raise TypeError(message)

    sources: dict[str, Source] = {}
    original_names: dict[str, list[str]] = defaultdict(list)

    for name, raw in config.items():
        if not isinstance(name, str):
            message = f"Source names in `{root_field}` must be strings"
            raise TypeError(message)

        if not name:
            message = f"Source names in `{root_field}` cannot be empty"
            raise ValueError(message)

        normalized = normalize_project_name(name)
        original_names[normalized].append(name)
        sources[normalized] = parse_source(name, raw, root_field=root_field)

    duplicates = [
        f"{normed} ({', '.join(originals)})" for normed, originals in original_names.items() if len(originals) > 1
    ]
    if duplicates:
        message = f"Field `{root_field}` contains duplicate names: {', '.join(duplicates)}"
        raise ValueError(message)

    return sources


def merge_source_tables(base: Mapping[str, Any], overrides: Mapping[str, Any]) -> dict[str, Any]:
    """
    Merge two unparsed sources tables, with entries in `overrides` replacing entries in
    `base` for the same project. Names are compared after
    [normalization](https://peps.python.org/pep-0503/#normalized-names) so that a table
    can override individual entries of another no matter how each spells a project name.
    """
    from hatch.utils.metadata import normalize_project_name

    overridden = {normalize_project_name(name) for name in overrides if isinstance(name, str)}
    merged = {
        name: source
        for name, source in base.items()
        if not isinstance(name, str) or normalize_project_name(name) not in overridden
    }
    merged.update(overrides)

    return merged


def render_path_url(path: str, root: str, *, subdirectory: str | None = None) -> str:
    """
    Render a `file://` URL for a path source. The `path` may be absolute or relative
    to `root`. The result is suitable as the right-hand side of a PEP 508 direct reference.
    """
    import os

    from hatch.utils.fs import Path

    candidate = Path(path) if os.path.isabs(path) else Path(root) / path
    uri = candidate.resolve().as_uri()
    if subdirectory:
        uri = f"{uri}#subdirectory={subdirectory}"

    return uri


def render_git_url(source: GitSource) -> str:
    """
    Render a `git+...` URL suitable for a PEP 508 direct reference.
    """
    url = f"git+{source.git}"
    if source.reference:
        url = f"{url}@{source.reference}"
    if source.subdirectory:
        url = f"{url}#subdirectory={source.subdirectory}"

    return url


def render_url(source: UrlSource) -> str:
    """
    Render a final URL for a URL source, including any `subdirectory` fragment.
    """
    url = source.url
    if source.subdirectory:
        separator = "&" if "#" in url else "#"
        url = f"{url}{separator}subdirectory={source.subdirectory}"

    return url


def lookup(sources: Mapping[str, Source], name: str) -> Source | None:
    """
    Look up a source by project name. Names are matched after PEP 503 normalization.
    """
    from hatch.utils.metadata import normalize_project_name

    return sources.get(normalize_project_name(name))


def apply_source_to_requirement(
    name: str,
    extras: list[str],
    source: Source,
    root: str,
    workspace_members: Mapping[str, str] | None = None,
) -> tuple[str, bool] | None:
    """
    Rewrite a requirement so it points at the given source. Returns a tuple of
    `(requirement_string, editable)` suitable for constructing a new `Dependency`,
    or `None` if the source does not rewrite the requirement (e.g. `IndexSource`,
    or a `WorkspaceSource` that cannot be matched to a member in `workspace_members`).

    `workspace_members` maps [normalized](https://peps.python.org/pep-0503/#normalized-names)
    project names to the local path of the matching workspace member.
    """
    extras_segment = f"[{','.join(extras)}]" if extras else ""

    if isinstance(source, PathSource):
        if source.editable and source.subdirectory:
            # Editable installs pass a bare directory to the installer, where a
            # `#subdirectory` fragment would be treated as part of the filesystem
            # path, so resolve the subdirectory into the path itself.
            import os

            url = render_path_url(os.path.join(source.path, source.subdirectory), root)
            return f"{name}{extras_segment} @ {url}", True

        url = render_path_url(source.path, root, subdirectory=source.subdirectory)
        return f"{name}{extras_segment} @ {url}", source.editable

    if isinstance(source, GitSource):
        url = render_git_url(source)
        return f"{name}{extras_segment} @ {url}", False

    if isinstance(source, UrlSource):
        url = render_url(source)
        return f"{name}{extras_segment} @ {url}", False

    if isinstance(source, WorkspaceSource):
        if workspace_members is None:
            return None

        from hatch.utils.metadata import normalize_project_name

        member_path = workspace_members.get(normalize_project_name(name))
        if member_path is None:
            return None

        # Workspace members are installed editable, matching `local_dependencies_complex`.
        url = render_path_url(member_path, root)
        return f"{name}{extras_segment} @ {url}", True

    return None


def collect_global_install_args(dependencies: Iterable[Dependency], sources: Mapping[str, Source]) -> list[str]:
    """
    Collect installer flags that apply to the entire install command (rather
    than to a single dependency). Currently this surfaces every `IndexSource`
    matching one of the given dependencies as `--extra-index-url`, deduplicated
    and order-preserving so PyPI remains the primary index.
    """
    args: list[str] = []
    seen: set[str] = set()
    for dependency in dependencies:
        # An explicit direct reference always wins over a configured source
        if dependency.url is not None:
            continue

        source = lookup(sources, dependency.name)
        if isinstance(source, IndexSource) and source.index not in seen:
            seen.add(source.index)
            args.extend(["--extra-index-url", source.index])

    return args


def decorate_dependency(
    dependency: Dependency,
    sources: Mapping[str, Source],
    root: str,
    workspace_members: Mapping[str, str] | None = None,
) -> Dependency:
    """
    Apply a matching source from `sources` to `dependency`, returning a new
    [`Dependency`](../utilities.md#hatch.dep.core.Dependency) if a rewrite is
    needed, or the original dependency otherwise.

    Dependencies that already define a PEP 508 direct reference are left alone.

    `workspace_members` maps [normalized](https://peps.python.org/pep-0503/#normalized-names)
    project names to the local path of the matching workspace member, used to resolve
    [`WorkspaceSource`](#WorkspaceSource) entries.
    """
    from hatch.dep.core import Dependency

    if dependency.url is not None:
        return dependency

    source = lookup(sources, dependency.name)
    if source is None:
        return dependency

    rewritten = apply_source_to_requirement(dependency.name, list(dependency.extras), source, root, workspace_members)
    if rewritten is None:
        if isinstance(source, WorkspaceSource):
            # A `WorkspaceSource` that reaches here could not be matched to a member, which
            # is a configuration error. Installing from a default index would silently ignore
            # the declared source, so fail loudly instead.
            message = (
                f"Dependency `{dependency.name}` declares `workspace = true` in its environment's "
                f"`sources` but no matching member was found in `tool.hatch.envs.<ENV_NAME>.workspace.members`"
            )
            raise ValueError(message)

        # `IndexSource` becomes an install-wide flag rather than changing the requirement
        return dependency

    spec, editable = rewritten
    if dependency.marker is not None:
        spec = f"{spec} ; {dependency.marker}"

    return Dependency(spec, editable=editable)


def decorate_dependencies(
    dependencies: Iterable[Dependency],
    sources: Mapping[str, Source],
    root: str,
    workspace_members: Mapping[str, str] | None = None,
) -> list[Dependency]:
    """
    Apply [`decorate_dependency`](#decorate_dependency) to each dependency in turn.
    """
    if not sources:
        return list(dependencies)

    return [decorate_dependency(dep, sources, root, workspace_members) for dep in dependencies]


def source_applied(
    dependency: Dependency, source: Source, root: str, workspace_members: Mapping[str, str] | None = None
) -> bool:
    """
    Whether `source` is what produced `dependency`, as opposed to the dependency carrying
    a direct reference of its own that takes precedence over the source.
    """
    rewritten = apply_source_to_requirement(dependency.name, list(dependency.extras), source, root, workspace_members)
    if rewritten is None:
        # Sources that do not rewrite the requirement apply unless it points elsewhere already
        return dependency.url is None

    from hatch.dep.core import Dependency

    spec, _ = rewritten
    return Dependency(spec).url == dependency.url


def describe_source(source: Source) -> tuple[str, str]:
    """
    Return a `(type, target)` pair describing `source` for display purposes.
    """
    if isinstance(source, PathSource):
        target = source.path
        if source.subdirectory:
            target = f"{target} (subdirectory: {source.subdirectory})"
        if not source.editable:
            target = f"{target} (not editable)"
        return "path", target

    if isinstance(source, GitSource):
        return "git", render_git_url(source)

    if isinstance(source, UrlSource):
        return "url", render_url(source)

    if isinstance(source, IndexSource):
        return "index", source.index

    if isinstance(source, WorkspaceSource):
        return "workspace", "resolved via `workspace.members`"

    return type(source).__name__, ""