Add helper to get DirectUrl metadata from Distrribution

src/pip/_internal/utils/direct_url_helpers.py

+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
@@ -87,3 +103,28 @@ def direct_url_from_link(link, source_dir=None, link_is_in_wheel_cache=False):
             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

tests/unit/test_direct_url_helpers.py

 from functools import partial
 
-from mock import patch
+from mock import MagicMock, patch
 
 from pip._internal.models.direct_url import (
+    DIRECT_URL_METADATA_NAME,
     ArchiveInfo,
     DirectUrl,
     DirInfo,
@@ -12,6 +13,7 @@ from pip._internal.models.link import Link
 from pip._internal.utils.direct_url_helpers import (
     direct_url_as_pep440_direct_reference,
     direct_url_from_link,
+    dist_get_direct_url,
 )
 from pip._internal.utils.urls import path_to_url
 
@@ -165,3 +167,30 @@ def test_from_link_hide_user_password():
         link_is_in_wheel_cache=True,
     )
     assert direct_url.to_dict()["url"] == "ssh://git@g.c/u/p.git"
+
+
+def test_dist_get_direct_url_no_metadata():
+    dist = MagicMock()
+    dist.has_metadata.return_value = False
+    assert dist_get_direct_url(dist) is None
+    dist.has_metadata.assert_called()
+
+
+def test_dist_get_direct_url_bad_metadata():
+    dist = MagicMock()
+    dist.has_metadata.return_value = True
+    dist.get_metadata.return_value = "{}"  # invalid direct_url.json
+    assert dist_get_direct_url(dist) is None
+    dist.get_metadata.assert_called_with(DIRECT_URL_METADATA_NAME)
+
+
+def test_dist_get_direct_url_valid_metadata():
+    dist = MagicMock()
+    dist.has_metadata.return_value = True
+    dist.get_metadata.return_value = (
+        '{"url": "https://e.c/p.tgz", "archive_info": {}}'
+    )
+    direct_url = dist_get_direct_url(dist)
+    dist.get_metadata.assert_called_with(DIRECT_URL_METADATA_NAME)
+    assert direct_url.url == "https://e.c/p.tgz"
+    assert isinstance(direct_url.info, ArchiveInfo)