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
    :class:`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 :exc:`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 :attr:`_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 :class:`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.

    Example::

        >>> parser = RequirementsParser()
        >>> reqs = parser.parse_file("requirements/prod.txt")
        >>> [r.name for r in reqs if r.editable]
        ['my-local-package']
    """
    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
    :meth:`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 :attr:`_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 :class:`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']
    """
    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):
            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)
    - ``-e <url-or-path>`` → editable :class:`Requirement`
    - ``pkg==1.0 --hash sha256:...`` → :class:`Requirement` with hashes
    - Standard PEP 508 specs → :class:`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.

    Example::

        >>> parser = RequirementsParser()
        >>> parser.parse_line("requests>=2.25.0", 1)
        <Requirement requests>=2.25.0>
        >>> parser.parse_line("# comment", 2) is None
        True
        >>> parser.parse_line("-r base.txt", 3)  # returns List[Requirement]
    """
    stripped_line = line_text.strip()

    # Blank lines and pure comments are skipped
    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

    # 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] = re.findall(r"--hash[=\s]+(\S+)", requirement_spec)
    if hash_values:
        # Remove all --hash tokens from the spec
        requirement_spec = " ".join(
            token
            for token in requirement_spec.split()
            if not token.startswith(HASH_DIRECTIVE)
        )

    # ── 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 normalised package names to their constraint
        :class:`Requirement` objects.

    Example::

        >>> parser = RequirementsParser()
        >>> parser.parse_file("requirements.txt")  # includes -c constraints.txt
        >>> constraints = parser.get_constraints()
        >>> constraints.get("django")
        <Requirement django==3.2>
    """
    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.

    Example::

        >>> parser = RequirementsParser()
        >>> parser.parse_file("projectA/requirements.txt")
        >>> parser.reset()
        >>> parser.parse_file("projectB/requirements.txt")  # clean slate
    """
    self._included_files_stack = []
    self._constraint_requirements = {}

async def get_package_info(
    self,
    name: str,
    current_version: Optional[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 :meth:`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.

    Returns:
        A :class:`Package` with ``latest_version``,
        ``recommended_version``, and metadata fields populated.

    Raises:
        PyPIError: The package does not exist on PyPI or the API
            returned an unexpected status.

    Example::

        >>> pkg = await checker.get_package_info("requests", current_version="2.25.0")
        >>> pkg.latest_version
        '2.31.0'
        >>> pkg.recommended_version
        '2.31.0'  # Stays in major version 2
    """
    try:
        pkg_data = await self.data_store.get_package_data(name)
    except PyPIError:
        # Package not found or API error — return unavailable stub
        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)

async def check_packages(
    self,
    requirements: List[Requirement],
) -> List[Package]:
    """Check multiple packages concurrently.

    For each requirement, extracts the current version (via
    :meth:`extract_current_version`) and calls :meth:`get_package_info`.
    Errors for individual packages are caught and replaced with
    unavailable stubs so that one bad package does not block the rest.

    Args:
        requirements: Parsed requirements from a requirements file.

    Returns:
        List of :class:`Package` objects, one per requirement.

    Example::

        >>> requirements = parser.parse_file("requirements.txt")
        >>> packages = await checker.check_packages(requirements)
        >>> [p.name for p in packages if p.recommended_version]
        ['flask', 'requests', 'click']
    """
    tasks = [self._create_package_check_task(req) for req in requirements]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return self._process_check_results(requirements, results)

def extract_current_version(
    self,
    req: Requirement,
) -> Optional[str]:
    """Infer a "current" version from a requirement's version specifiers.

    Heuristic:

    1. If the requirement has exactly one specifier and it is ``==``,
       return that version (pinned).
    2. If :attr:`infer_version_from_constraints` is ``False``, stop here.
    3. Otherwise, scan for the first ``>=``, ``>``, or ``~=`` specifier
       and return its version. This treats ``>=2.0`` as "currently on
       2.0" for major-version boundary purposes.

    Args:
        req: A parsed :class:`Requirement`.

    Returns:
        The inferred version string, or ``None`` when inference is not
        possible.

    Example::

        >>> req1 = Requirement(name="flask", specs=[("==", "2.0.0")], ...)
        >>> checker.extract_current_version(req1)
        '2.0.0'

        >>> req2 = Requirement(name="flask", specs=[(">=", "2.0"), ("<", "3")], ...)
        >>> checker.extract_current_version(req2)
        '2.0'

        >>> req3 = Requirement(name="flask", specs=[], ...)
        >>> checker.extract_current_version(req3) is None
        True
    """
    if not req.specs:
        return None

    # Exact pin: treat as the current version
    if len(req.specs) == 1 and req.specs[0][0] == "==":
        return req.specs[0][1]

    if not self.infer_version_from_constraints:
        return None

    # Infer from range lower-bound operators
    for operator, version in req.specs:
        if operator in (">=", ">", "~="):
            return version

    return None

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

    Uses double-checked locking: the first check is lock-free; if the
    package is missing a second check runs *inside* the semaphore so
    that only one coroutine actually performs the HTTP call.

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

    Returns:
        A :class:`PyPIPackageData` populated from the latest PyPI
        JSON response.

    Raises:
        PyPIError: The package does not exist on PyPI or the API
            returned an unexpected status code.

    Example::

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

    # Fast path — already cached (no lock needed)
    if normalized in self._package_data:
        return self._package_data[normalized]

    async with self._semaphore:
        # Second check — another coroutine may have populated while we waited
        if normalized in self._package_data:
            return self._package_data[normalized]

        data = await self._fetch_from_pypi(name)
        pkg_data = self._parse_package_data(name, data)
        self._package_data[normalized] = pkg_data
        return pkg_data

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

    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", "scipy"])
        # subsequent get_package_data calls for these return instantly
    """
    await asyncio.gather(
        *(self.get_package_data(name) for name in names),
        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
       :class:`PyPIPackageData` (``latest_dependencies`` or
       ``dependencies_cache``).
    3. A targeted ``/pypi/{name}/{version}/json`` fetch, guarded by
       the semaphore and a second cache check.

    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 ──────────────────────────
    if cache_key in self._version_deps_cache:
        return self._version_deps_cache[cache_key]

    # ── 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 (double-checked) ────────────────────
    async with self._semaphore:
        if cache_key in self._version_deps_cache:
            return self._version_deps_cache[cache_key]

        deps = await self._fetch_version_dependencies(name, version)
        self._version_deps_cache[cache_key] = deps

        # Back-fill the package-level cache so future reads skip this path
        if pkg_data:
            pkg_data.dependencies_cache[version] = deps

        return deps

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 :class:`PyPIPackageData`, or ``None`` if the
        package has not been fetched yet.

    Example::

        >>> store.get_cached_package("flask")  # after a prior fetch
        PyPIPackageData(name='flask', latest_version='3.0.0', ...)
        >>> store.get_cached_package("unknown")
        None
    """
    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 ``[]``.

    Example::

        >>> store.get_versions("flask")
        ['3.0.0', '2.3.3', '2.3.2', ...]
    """
    pkg = self.get_cached_package(name)
    return pkg.all_versions if pkg else []

async def resolve_and_annotate_conflicts(
    self,
    packages: List[Package],
) -> ResolutionResult:
    """Resolve conflicts while strictly respecting major version boundaries.

    Algorithm outline:

    1. Build an *update set* mapping each package name to its
       proposed version (``recommended_version`` if available,
       otherwise ``current_version``). Recommended versions already
       respect major version boundaries.
    2. Prefetch metadata for every package in one concurrent burst.
    3. Loop up to :data:`_MAX_RESOLUTION_ITERATIONS` times:

       a. Scan for cross-conflicts in the current update set.
       b. If none remain, stop — the set is self-consistent.
       c. Attempt resolution within major version boundaries only:

          - Try to find a compatible source version within its current major
          - If that fails, try to constrain the target within its current major
          - If both fail, revert both packages to their current versions

       d. Break early when no progress is made.

    4. Annotate each :class:`Package` with its final version and any
       unresolved :class:`Conflict` objects.
    5. Return a :class:`ResolutionResult` with complete details.

    Args:
        packages: Mutable list of :class:`Package` objects. Each
            object is updated in place with the resolved version and
            conflict metadata.

    Returns:
        :class:`ResolutionResult` containing the final version for each
        package, conflict details, and resolution statistics.

    Example::

        >>> result = await analyzer.resolve_and_annotate_conflicts(pkgs)
        >>> print(result.summary())
        >>> for pkg_name, info in result.resolved_versions.items():
        ...     if info.was_changed():
        ...         print(f"{pkg_name}: {info.original} → {info.resolved}")
    """
    # ── initialise update set ─────────────────────────────────────
    pkg_lookup: Dict[str, Package] = {pkg.name: pkg for pkg in packages}
    update_set: Dict[str, Optional[str]] = {}
    conflict_tracking: Dict[str, List[Conflict]] = {}

    # Track original proposed versions for comparison
    original_versions: Dict[str, Optional[str]] = {}

    for pkg in packages:
        # Use recommended version (which already respects major boundaries)
        proposed = pkg.recommended_version or pkg.current_version
        update_set[pkg.name] = proposed
        original_versions[pkg.name] = proposed

    # ── warm the cache in one round-trip ──────────────────────────
    await self.data_store.prefetch_packages([pkg.name for pkg in packages])

    # ── iterative conflict resolution ─────────────────────────────
    iterations_used = 0
    converged = False

    for iteration in range(_MAX_RESOLUTION_ITERATIONS):
        iterations_used = iteration + 1
        cross_conflicts = await self._find_cross_conflicts(packages, update_set)

        if not cross_conflicts:
            logger.debug(
                "Update set is conflict-free after %d iteration(s)", iteration
            )
            converged = True
            break

        # Record every conflict for later annotation (with deduplication)
        for conflict in cross_conflicts:
            conflicts_list = conflict_tracking.setdefault(
                conflict.target_package, []
            )

            # Deduplicate using conflict signature
            conflict_key = (
                conflict.source_package,
                conflict.source_version,
                conflict.required_spec,
                conflict.conflicting_version,
            )
            existing_keys = {
                (
                    c.source_package,
                    c.source_version,
                    c.required_spec,
                    c.conflicting_version,
                )
                for c in conflicts_list
            }
            if conflict_key not in existing_keys:
                conflicts_list.append(conflict)

        # Attempt resolution while respecting major version boundaries
        resolved_any = await self._resolve_conflicts_within_major(
            pkg_lookup, update_set, cross_conflicts
        )

        if not resolved_any:
            # No version change was made → further iterations would
            # produce the exact same conflict set; stop early.
            logger.warning(
                "Conflict resolution stalled after %d iteration(s)",
                iteration + 1,
            )
            break
    else:
        # for/else: exhausted all iterations without breaking
        logger.warning(
            "Conflict resolution did not converge within %d iterations",
            _MAX_RESOLUTION_ITERATIONS,
        )

    # ── annotate packages and build resolution map ────────────────
    resolved_versions: Dict[str, PackageResolution] = {}
    packages_with_conflicts = 0

    for pkg in packages:
        original = original_versions.get(pkg.name)
        resolved = update_set.get(pkg.name)
        conflicts = conflict_tracking.get(pkg.name, [])

        # Determine resolution status
        status = self._determine_status(pkg, original, resolved, conflicts)

        # Find compatible alternative if there are conflicts
        # (constrained to current major version only)
        compatible_alt = None
        if conflicts:
            # Get current major version to constrain search
            current_major = _get_major_version(pkg.current_version)

            if current_major is not None:
                # Only look within current major
                available = self.data_store.get_versions(pkg.name)
                available_in_major = [
                    v for v in available if _get_major_version(v) == current_major
                ]

                conflict_set = ConflictSet(pkg.name)
                for c in conflicts:
                    conflict_set.add_conflict(c)

                compatible_alt = self.find_compatible_version(
                    conflict_set, available_in_major, pkg.current_version
                )

            packages_with_conflicts += 1

        # Update the Package object itself
        if resolved and resolved != pkg.recommended_version:
            # Only update recommended_version if resolution changed it
            pkg.recommended_version = (
                resolved if resolved != pkg.current_version else pkg.current_version
            )
        if conflicts:
            pkg.set_conflicts(conflicts, resolved_version=compatible_alt)

        # Store resolution details
        resolved_versions[pkg.name] = PackageResolution(
            name=pkg.name,
            original=original,
            resolved=resolved,
            status=status,
            conflicts=conflicts,
            compatible_alternative=compatible_alt,
        )

    return ResolutionResult(
        resolved_versions=resolved_versions,
        total_packages=len(packages),
        packages_with_conflicts=packages_with_conflicts,
        iterations_used=iterations_used,
        converged=converged,
    )