| Viewing file:  direct_url_helpers.py (3.13 KB)      -rw-r--r-- Select action/file-type:
 
  (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) | 
 
from typing import Optional
 from pip._internal.models.direct_url import ArchiveInfo, DirectUrl, DirInfo, VcsInfo
 from pip._internal.models.link import Link
 from pip._internal.utils.urls import path_to_url
 from pip._internal.vcs import vcs
 
 
 def direct_url_as_pep440_direct_reference(direct_url: DirectUrl, name: str) -> str:
 """Convert a DirectUrl to a pip requirement string."""
 direct_url.validate()  # if invalid, this is a pip bug
 requirement = name + " @ "
 fragments = []
 if isinstance(direct_url.info, VcsInfo):
 requirement += "{}+{}@{}".format(
 direct_url.info.vcs, direct_url.url, direct_url.info.commit_id
 )
 elif isinstance(direct_url.info, ArchiveInfo):
 requirement += direct_url.url
 if direct_url.info.hash:
 fragments.append(direct_url.info.hash)
 else:
 assert isinstance(direct_url.info, DirInfo)
 requirement += direct_url.url
 if direct_url.subdirectory:
 fragments.append("subdirectory=" + direct_url.subdirectory)
 if fragments:
 requirement += "#" + "&".join(fragments)
 return requirement
 
 
 def direct_url_for_editable(source_dir: str) -> DirectUrl:
 return DirectUrl(
 url=path_to_url(source_dir),
 info=DirInfo(editable=True),
 )
 
 
 def direct_url_from_link(
 link: Link, source_dir: Optional[str] = None, link_is_in_wheel_cache: bool = False
 ) -> DirectUrl:
 if link.is_vcs:
 vcs_backend = vcs.get_backend_for_scheme(link.scheme)
 assert vcs_backend
 url, requested_revision, _ = vcs_backend.get_url_rev_and_auth(
 link.url_without_fragment
 )
 # For VCS links, we need to find out and add commit_id.
 if link_is_in_wheel_cache:
 # If the requested VCS link corresponds to a cached
 # wheel, it means the requested revision was an
 # immutable commit hash, otherwise it would not have
 # been cached. In that case we don't have a source_dir
 # with the VCS checkout.
 assert requested_revision
 commit_id = requested_revision
 else:
 # If the wheel was not in cache, it means we have
 # had to checkout from VCS to build and we have a source_dir
 # which we can inspect to find out the commit id.
 assert source_dir
 commit_id = vcs_backend.get_revision(source_dir)
 return DirectUrl(
 url=url,
 info=VcsInfo(
 vcs=vcs_backend.name,
 commit_id=commit_id,
 requested_revision=requested_revision,
 ),
 subdirectory=link.subdirectory_fragment,
 )
 elif link.is_existing_dir():
 return DirectUrl(
 url=link.url_without_fragment,
 info=DirInfo(),
 subdirectory=link.subdirectory_fragment,
 )
 else:
 hash = None
 hash_name = link.hash_name
 if hash_name:
 hash = f"{hash_name}={link.hash}"
 return DirectUrl(
 url=link.url_without_fragment,
 info=ArchiveInfo(hash=hash),
 subdirectory=link.subdirectory_fragment,
 )
 
 |