1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
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
|
