Python API¶

def parse_file(
    self,
    file_path: Union[str, Path],
    is_constraint_file: bool = False,
    _parent_directory_path: Optional[Path] = None,
) -> List[Requirement]:
    """Parse a requirements file from disk.

    Reads the file at *file_path*, processes all directives (``-r``,
    ``-c``, ``-e``, ``--hash``), and returns a flat list of
    `Requirement` objects.  If *file_path* is relative and
    *_parent_directory_path* is provided (internal use by ``-r``), the
    path is resolved relative to the parent.

    Circular include chains (``A.txt`` includes ``B.txt`` which
    includes ``A.txt``) are detected and raise `ParseError`.

    Args:
        file_path: Path to the requirements file (absolute or relative).
        is_constraint_file: If ``True``, all parsed requirements are
            stored as constraints (via `_constraint_requirements`)
            rather than returned.  Used internally by ``-c`` handlers.
        _parent_directory_path: Internal parameter used when resolving
            ``-r`` includes; the parent's directory is used as the base
            for relative paths.

    Returns:
        List of `Requirement` objects (empty if
        *is_constraint_file* is ``True``).

    Raises:
        FileOperationError: The file does not exist or cannot be read.
        ParseError: A circular include was detected or the file contains
            invalid syntax.
    """
    resolved_path = self._resolve_file_path(
        file_path=Path(file_path),
        parent_directory=_parent_directory_path,
    )

    self.logger.debug(
        "Parsing file: %s%s",
        resolved_path,
        " (constraint file)" if is_constraint_file else "",
    )

    # Detect circular includes before reading
    if resolved_path in self._included_files_stack:
        cycle_path = " -> ".join(
            str(p) for p in self._included_files_stack + [resolved_path]
        )
        self.logger.error("Circular dependency detected: %s", cycle_path)
        raise ParseError(
            f"Circular dependency detected: {cycle_path}",
            file_path=str(resolved_path),
        )

    file_content = safe_read_file(resolved_path)

    self._included_files_stack.append(resolved_path)
    try:
        result = self.parse_string(
            file_content,
            source_file_path=str(resolved_path),
            is_constraint_file=is_constraint_file,
            _current_directory_path=resolved_path,
        )
        self.logger.debug(
            "Parsed %d requirement(s) from %s",
            len(result),
            resolved_path.name,
        )
        return result
    finally:
        self._included_files_stack.pop()

def parse_string(
    self,
    requirements_content: str,
    source_file_path: Optional[str] = None,
    is_constraint_file: bool = False,
    _current_directory_path: Optional[Path] = None,
) -> List[Requirement]:
    """Parse requirements from raw text content.

    Splits *requirements_content* into lines and processes each via
    `parse_line`.  Requirements loaded from ``-r`` includes are
    flattened into the result list.

    Args:
        requirements_content: Multi-line requirements text.
        source_file_path: Optional file path for error messages (purely
            informational; does not affect parsing).
        is_constraint_file: If ``True``, all parsed requirements are
            stored in `_constraint_requirements` instead of being
            returned.
        _current_directory_path: Internal parameter; the directory
            containing the "file" being parsed (used to resolve
            relative ``-r`` / ``-c`` paths).

    Returns:
        List of `Requirement` objects.

    Example::

        >>> content = \"\"\"
        ... flask>=2.0
        ... # A comment
        ... requests>=2.25.0
        ... \"\"\"
        >>> parser = RequirementsParser()
        >>> reqs = parser.parse_string(content)
        >>> [r.name for r in reqs]
        ['flask', 'requests']
    """
    # A byte order mark is a stream-level signature, not part of line 1.
    # It survives non-BOM-aware decoding and is not removed by strip().
    if requirements_content.startswith(BOM_CHARACTER):
        requirements_content = requirements_content[len(BOM_CHARACTER):]

    parsed_requirements: List[Requirement] = []
    total_lines = len(requirements_content.splitlines())
    self.logger.debug(
        "Parsing %d line(s)%s",
        total_lines,
        f" from {source_file_path}" if source_file_path else "",
    )

    for line_number, line_text in enumerate(
        requirements_content.splitlines(), start=1
    ):
        parse_result = self.parse_line(
            line_text,
            line_number,
            source_file_path,
            _current_directory_path=_current_directory_path,
        )

        if parse_result is None:
            # Comment or blank line
            continue

        if isinstance(parse_result, list):
            # Nested requirements from -r directive
            self.logger.debug(
                "Included %d requirement(s) from directive on line %d",
                len(parse_result),
                line_number,
            )
            parsed_requirements.extend(parse_result)
        elif isinstance(parse_result, Requirement):
            # Record provenance so the update writer can rewrite the
            # correct file. Requirements flattened in from -r includes
            # (the list branch above) already carry the included file's
            # path set by the recursive parse_string call.
            parse_result.source_file = source_file_path

            if is_constraint_file:
                # Store in constraint map instead of returning
                self._constraint_requirements[parse_result.name] = parse_result
                self.logger.debug(
                    "Stored constraint: %s %s",
                    parse_result.name,
                    parse_result.specs,
                )
            else:
                parsed_requirements.append(parse_result)

    self.logger.debug(
        "Completed parsing: %d requirement(s)", len(parsed_requirements)
    )
    return parsed_requirements

def parse_line(
    self,
    line_text: str,
    line_number: int,
    source_file_path: Optional[str] = None,
    _current_directory_path: Optional[Path] = None,
) -> Optional[Union[Requirement, List[Requirement]]]:
    """Parse a single line from a requirements file.

    Handles all pip-supported line types:

    - Blank lines and ``#`` comments → ``None``
    - ``-r file.txt`` → ``List[Requirement]`` (nested parse)
    - ``-c file.txt`` → ``None`` (side-effect: populates constraints)
    - Pip global option lines (e.g. ``--index-url ...``) → ``None``
    - ``-e <url-or-path>`` → editable `Requirement`
    - ``pkg==1.0 --hash sha256:...`` → `Requirement` with hashes
    - Standard PEP 508 specs → `Requirement`

    Args:
        line_text: Raw line text (may include leading/trailing whitespace).
        line_number: Line number (1-indexed) for error reporting.
        source_file_path: Optional source file path for error messages.
        _current_directory_path: Internal; directory of the file being
            parsed (used to resolve relative ``-r`` / ``-c`` paths).

    Returns:
        - ``None`` for comments, blank lines, or ``-c`` directives.
        - ``List[Requirement]`` when the line is a ``-r`` include.
        - ``Requirement`` for all other valid package specs.

    Raises:
        ParseError: The line contains invalid syntax or a directive
            that cannot be processed.
    """
    stripped_line = line_text.strip()

    if not stripped_line or stripped_line.startswith("#"):
        return None

    # Extract inline comment (everything after a non-URL '#')
    requirement_spec, inline_comment = self._extract_inline_comment(stripped_line)

    # ── Handle -r / --requirement (include another file) ──────────
    if requirement_spec.startswith((INCLUDE_DIRECTIVE, INCLUDE_DIRECTIVE_LONG)):
        return self._handle_include_directive(
            requirement_spec,
            line_number,
            source_file_path,
            _current_directory_path,
        )

    # ── Handle -c / --constraint (load constraints) ───────────────
    if requirement_spec.startswith(
        (CONSTRAINT_DIRECTIVE, CONSTRAINT_DIRECTIVE_LONG)
    ):
        self._handle_constraint_directive(
            requirement_spec,
            line_number,
            source_file_path,
            _current_directory_path,
        )
        return None  # constraints are stored, not returned

    # Recognized pip global options configure installer behavior and are
    # not package requirements, so they should be ignored by the parser.
    if self._is_supported_global_option_line(requirement_spec):
        self.logger.debug(
            "Line %d: Skipping supported pip global option: %s",
            line_number,
            requirement_spec,
        )
        return None

    # Strip quotes that may wrap the entire spec
    requirement_spec = self._remove_surrounding_quotes(requirement_spec)

    # ── Check for -e / --editable flag ────────────────────────────
    is_editable = requirement_spec.startswith(
        (EDITABLE_DIRECTIVE, EDITABLE_DIRECTIVE_LONG)
    )
    if is_editable:
        # Extract everything after "-e " or "--editable "
        requirement_spec = (
            requirement_spec.split(None, 1)[1] if " " in requirement_spec else ""
        )

    # ── Extract --hash directives ──────────────────────────────────
    hash_values: List[str] = _HASH_DIRECTIVE_PATTERN.findall(requirement_spec)
    if hash_values:
        # Remove the entire ``--hash <digest>`` / ``--hash=<digest>``
        # directive (flag AND digest) using the same pattern that
        # extracted it.  Token-based filtering only dropped the ``--hash``
        # flag and left the space-separated digest behind, which then
        # failed PEP 508 parsing.
        requirement_spec = _HASH_DIRECTIVE_PATTERN.sub(" ", requirement_spec)
        requirement_spec = " ".join(requirement_spec.split())

    # ── Dispatch to appropriate builder ────────────────────────────
    url_components = self._parse_direct_url(requirement_spec)
    if url_components:
        parsed_requirement = self._build_url_based_requirement(
            url_string=requirement_spec,
            url_components=url_components,
            is_editable=is_editable,
            hash_values=hash_values,
            inline_comment=inline_comment,
            original_line=line_text,
            line_number=line_number,
        )

    elif local_path_components := self._parse_local_file_path(requirement_spec):
        parsed_requirement = self._build_local_path_requirement(
            path_components=local_path_components,
            current_directory=_current_directory_path,
            is_editable=is_editable,
            hash_values=hash_values,
            inline_comment=inline_comment,
            original_line=line_text,
            line_number=line_number,
        )

    else:
        # Standard PEP 508 package specifier
        parsed_requirement = self._build_standard_pep508_requirement(
            requirement_spec=requirement_spec,
            is_editable=is_editable,
            hash_values=hash_values,
            inline_comment=inline_comment,
            original_line=line_text,
            line_number=line_number,
            source_file_path=source_file_path,
        )

    # Apply any constraint loaded via -c directive
    return self._apply_constraint_to_requirement(parsed_requirement)

def get_constraints(self) -> Dict[str, Requirement]:
    """Return a copy of all constraint requirements loaded via ``-c``.

    Returns:
        Dictionary mapping normalized package names to their constraint
        `Requirement` objects.
    """
    return self._constraint_requirements.copy()

def reset(self) -> None:
    """Clear all internal state (include stack and constraints).

    Call this before reusing the parser on a new, unrelated set of
    files to prevent cross-contamination.
    """
    self._included_files_stack = []
    self._constraint_requirements = {}

async def get_package_data(self, name: str) -> PyPIPackageData:
    """Fetch (or return cached) metadata for *name*.

    Cached entries are returned without awaiting anything.  Otherwise
    the call is *coalesced*: the first caller starts the fetch and any
    concurrent caller for the same normalized name awaits that same
    fetch, so PyPI sees exactly one request.

    Args:
        name: PyPI package name (any casing / underscore style).

    Returns:
        A `PyPIPackageData` populated from the latest PyPI
        JSON response.  Every caller for the same normalized name
        receives the *same* object.

    Raises:
        PyPIError: The package does not exist on PyPI or the API
            returned an unexpected status code.
        NetworkError: The request failed (timeout, rate limit, 5xx).

    Example::

        >>> data = await store.get_package_data("Requests")
        >>> data.name
        'requests'
        >>> data.latest_version
        '2.31.0'
    """
    normalized = _normalize(name)

    # Fast path — already cached (never suspends)
    cached = self._package_data.get(normalized)
    if cached is not None:
        return cached

    return await self._coalesce(
        self._inflight_packages,
        normalized,
        lambda: self._load_package_data(name, normalized),
    )

async def prefetch_packages(self, names: List[str]) -> None:
    """Concurrently warm the cache for a batch of packages.

    Duplicate names (including different spellings that normalize to
    the same key, e.g. ``Flask`` and ``flask``) are collapsed to a
    single fetch.  Errors for individual packages are silenced so that
    one bad package name does not prevent the rest from being cached.

    Args:
        names: Package names to prefetch.

    Example::

        >>> await store.prefetch_packages(["numpy", "pandas", "numpy"])
        # two fetches, not three; subsequent get_package_data calls
        # for these return instantly
    """
    # dict preserves insertion order → deterministic request ordering,
    # and the first spelling of each name is the one sent to PyPI.
    unique: Dict[str, str] = {}
    for name in names:
        unique.setdefault(_normalize(name), name)

    await asyncio.gather(
        *(self.get_package_data(name) for name in unique.values()),
        return_exceptions=True,  # swallow per-package failures
    )

async def get_version_dependencies(
    self,
    name: str,
    version: str,
) -> List[str]:
    """Return the base dependencies for a specific version of *name*.

    Resolution order (fastest first):

    1. Per-version dependency cache (``_version_deps_cache``).
    2. Already-populated fields inside the cached
       `PyPIPackageData` (``latest_dependencies`` or
       ``dependencies_cache``).
    3. A targeted ``/pypi/{name}/{version}/json`` fetch, coalesced per
       ``name==version`` key and throttled by the semaphore.

    Args:
        name: Package name.
        version: Exact version string, e.g. ``"1.2.3"``.

    Returns:
        List of PEP-508 dependency specifiers with extras and
        environment markers stripped.

    Example::

        >>> deps = await store.get_version_dependencies("flask", "2.3.0")
        >>> deps
        ['Werkzeug>=2.0', 'Jinja2>=3.0', ...]
    """
    normalized = _normalize(name)
    cache_key = f"{normalized}=={version}"

    # ── layer 1: flat version-deps cache ──────────────────────────
    cached_deps = self._version_deps_cache.get(cache_key)
    if cached_deps is not None:
        return cached_deps

    # ── layer 2: already inside PyPIPackageData ────────────────────
    pkg_data = self._package_data.get(normalized)
    if pkg_data:
        if version == pkg_data.latest_version:
            self._version_deps_cache[cache_key] = pkg_data.latest_dependencies
            return pkg_data.latest_dependencies

        if version in pkg_data.dependencies_cache:
            deps = pkg_data.dependencies_cache[version]
            self._version_deps_cache[cache_key] = deps
            return deps

    # ── layer 3: network fetch (coalesced per name==version) ───────
    return await self._coalesce(
        self._inflight_version_deps,
        cache_key,
        lambda: self._load_version_dependencies(name, version, normalized, cache_key),
    )

def get_cached_package(self, name: str) -> Optional[PyPIPackageData]:
    """Return cached data for *name* without triggering a fetch.

    Args:
        name: Package name (any casing / underscore style).

    Returns:
        The cached `PyPIPackageData`, or ``None`` if the
        package has not been fetched yet.
    """
    return self._package_data.get(_normalize(name))

def get_versions(self, name: str) -> List[str]:
    """Return cached stable versions for *name* (newest first).

    Returns an empty list when *name* has not been fetched yet.

    Args:
        name: Package name.

    Returns:
        List of version strings, or ``[]``.
    """
    pkg = self.get_cached_package(name)
    return pkg.all_versions if pkg else []

def is_python_compatible(
    self,
    name: str,
    version: str,
    python_version: str,
) -> bool:
    """Check Python compatibility using only cached metadata.

    Returns ``True`` when the package has not been fetched yet — the
    caller should call `get_package_data` first if a definitive
    answer is needed.

    Args:
        name: Package name.
        version: Package version string.
        python_version: Dot-separated Python version.

    Returns:
        Compatibility flag (see `PyPIPackageData.is_python_compatible`).
    """
    pkg = self.get_cached_package(name)
    return pkg.is_python_compatible(version, python_version) if pkg else True

@staticmethod
def get_current_python_version() -> str:
    """Return the running interpreter's version as ``"major.minor.micro"``.

    Example::

        >>> PyPIDataStore.get_current_python_version()
        '3.11.4'
    """
    return (
        f"{sys.version_info.major}."
        f"{sys.version_info.minor}."
        f"{sys.version_info.micro}"
    )

def get_versions_in_major(self, major: int) -> List[str]:
    """Return stable versions that share a given major number.

    Pre-releases and versions whose ``release`` tuple is empty are
    skipped.

    Args:
        major: The major version number to filter on (e.g. ``2``).

    Returns:
        Version strings in descending order (inherits the sort order
        of `parsed_versions`).
    """
    result: List[str] = []

    for version_str, parsed in self.parsed_versions:
        if parsed.is_prerelease:
            continue
        # `release` may be empty for exotic versions such as an epoch-only
        # tag, so index 0 is not guaranteed to exist.
        if parsed.release and parsed.release[0] == major:
            result.append(version_str)

    return result

def is_python_compatible(
    self,
    version: str,
    python_version: str,
) -> bool:
    """Check whether a package version supports a given Python version.

    Returns ``True`` when the package omits ``requires_python`` or when
    parsing the specifier fails — matching pip's own permissive
    behavior.

    Args:
        version: Package version string, e.g. ``"1.4.2"``.
        python_version: Dot-separated Python version, e.g.
            ``"3.11.2"``.

    Returns:
        ``True`` if *python_version* satisfies the package's
        ``requires_python`` constraint (or if the constraint is absent /
        unparseable).

    Example::

        >>> data.python_requirements["1.4.2"] = ">=3.7"
        >>> data.is_python_compatible("1.4.2", "3.11.2")
        True
        >>> data.is_python_compatible("1.4.2", "2.7.18")
        False
    """
    requires_python = self.python_requirements.get(version)

    # Absent constraint means "any Python", which is how pip reads it.
    if not requires_python:
        return True

    try:
        return python_version in SpecifierSet(requires_python)
    except InvalidSpecifier:
        # Malformed upstream metadata must not exclude an otherwise
        # installable release.
        return True

def get_python_compatible_versions(
    self,
    python_version: str,
    major: Optional[int] = None,
) -> List[str]:
    """Return stable versions compatible with *python_version*.

    Optionally restrict results to a single major version.  Versions
    are returned in descending order.

    Args:
        python_version: Dot-separated Python version to check against,
            e.g. ``"3.10.0"``.
        major: If provided, only versions with this major number are
            included.

    Returns:
        Filtered, descending list of version strings.
    """
    result: List[str] = []

    for version_str, parsed in self.parsed_versions:
        if parsed.is_prerelease:
            continue

        if major is not None:
            if not parsed.release or parsed.release[0] != major:
                continue

        if self.is_python_compatible(version_str, python_version):
            result.append(version_str)

    return result

async def get_package_info(
    self,
    name: str,
    current_version: Optional[str] = None,
    constraints: Optional[Sequence[Tuple[str, str]]] = None,
) -> Package:
    """Fetch metadata and compute a recommended version for *name*.

    **CRITICAL**: Recommendations **NEVER** cross major version boundaries.
    If *current_version* is ``2.x.x``, the recommended version will be
    ``2.y.z`` (never ``3.0.0``), even if ``3.0.0`` is the latest available
    version on PyPI.

    Calls `PyPIDataStore.get_package_data` (which may trigger a
    network fetch or return cached data), then applies the strict
    major-boundary recommendation algorithm to choose the best upgrade
    target.

    Args:
        name: Package name (any casing / separator style).
        current_version: The version currently installed (if known).
            When provided, the recommendation stays within the same
            major version. If no compatible version exists in that
            major, stays on current version rather than crossing
            the boundary.
        constraints: Additional ``(operator, version)`` specifiers the
            recommendation must satisfy — typically the upper bounds and
            exclusions declared by the requirement itself (see
            `retained_specs`).
            Recommending a version that violates them would produce an
            unsatisfiable requirement line.

    Returns:
        A `Package` with ``latest_version``,
        ``recommended_version``, and metadata fields populated. When PyPI
        data cannot be retrieved (missing package, unexpected status,
        timeout, rate limiting) an *unavailable stub* is returned instead
        — see `create_unavailable_package`.
    """
    try:
        pkg_data = await self.data_store.get_package_data(name)
    except NetworkError:
        # 404, unexpected status, timeout or rate limiting (PyPIError is a
        # NetworkError) — return an unavailable stub rather than failing
        # the whole run.
        logger.warning("Package '%s' unavailable; creating stub", name)
        return self.create_unavailable_package(name, current_version)

    return self._build_package_from_data(pkg_data, current_version, constraints)