| import logging |
| |
| from pip._internal.models.direct_url import ( |
| DIRECT_URL_METADATA_NAME, |
| ArchiveInfo, |
| DirectUrl, |
| DirectUrlValidationError, |
| DirInfo, |
| VcsInfo, |
| ) |
| from pip._internal.utils.typing import MYPY_CHECK_RUNNING |
| from pip._internal.vcs import vcs |
| |
| try: |
| from json import JSONDecodeError |
| except ImportError: |
| # PY2 |
| JSONDecodeError = ValueError # type: ignore |
| |
| if MYPY_CHECK_RUNNING: |
| from typing import Optional |
| |
| from pip._internal.models.link import Link |
| |
| from pip._vendor.pkg_resources import Distribution |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| def direct_url_as_pep440_direct_reference(direct_url, name): |
| # type: (DirectUrl, 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) |
| # pip should never reach this point for editables, since |
| # pip freeze inspects the editable project location to produce |
| # the requirement string |
| assert not direct_url.info.editable |
| requirement += direct_url.url |
| if direct_url.subdirectory: |
| fragments.append("subdirectory=" + direct_url.subdirectory) |
| if fragments: |
| requirement += "#" + "&".join(fragments) |
| return requirement |
| |
| |
| def direct_url_from_link(link, source_dir=None, link_is_in_wheel_cache=False): |
| # type: (Link, Optional[str], bool) -> 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 = "{}={}".format(hash_name, link.hash) |
| return DirectUrl( |
| url=link.url_without_fragment, |
| info=ArchiveInfo(hash=hash), |
| subdirectory=link.subdirectory_fragment, |
| ) |
| |
| |
| def dist_get_direct_url(dist): |
| # type: (Distribution) -> Optional[DirectUrl] |
| """Obtain a DirectUrl from a pkg_resource.Distribution. |
| |
| Returns None if the distribution has no `direct_url.json` metadata, |
| or if `direct_url.json` is invalid. |
| """ |
| if not dist.has_metadata(DIRECT_URL_METADATA_NAME): |
| return None |
| try: |
| return DirectUrl.from_json(dist.get_metadata(DIRECT_URL_METADATA_NAME)) |
| except ( |
| DirectUrlValidationError, |
| JSONDecodeError, |
| UnicodeDecodeError |
| ) as e: |
| logger.warning( |
| "Error parsing %s for %s: %s", |
| DIRECT_URL_METADATA_NAME, |
| dist.project_name, |
| e, |
| ) |
| return None |