utils: add support for extracting project dist info

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>

moonraker/utils/__init__.py

@@ -21,6 +21,7 @@ import socket
 import enum
 import ipaddress
 import platform
+from .exceptions import ServerError
 from . import source_info
 from . import json_wrapper
 
@@ -49,11 +50,6 @@ try:
 except Exception:
     KERNEL_VERSION = (0, 0)
 
-class ServerError(Exception):
-    def __init__(self, message: str, status_code: int = 400) -> None:
-        Exception.__init__(self, message)
-        self.status_code = status_code
-
 
 class Sentinel(enum.Enum):
     MISSING = object()

moonraker/utils/exceptions.py

@@ -0,0 +1,10 @@
+# Moonraker Exceptions
+#
+# Copyright (C) 2023 Eric Callahan <arksine.code@gmail.com>
+#
+# This file may be distributed under the terms of the GNU GPLv3 license
+
+class ServerError(Exception):
+    def __init__(self, message: str, status_code: int = 400) -> None:
+        Exception.__init__(self, message)
+        self.status_code = status_code

moonraker/utils/source_info.py

@@ -9,10 +9,21 @@ import importlib.resources as ilr
 import pathlib
 import sys
 import site
+import re
+import json
+import logging
+from dataclasses import dataclass
+if sys.version_info < (3, 8):
+    from importlib_metadata import Distribution, PathDistribution, PackageMetadata
+else:
+    from importlib.metadata import Distribution, PathDistribution, PackageMetadata
+from .exceptions import ServerError
 
 # Annotation imports
 from typing import (
     Optional,
+    Dict,
+    Any
 )
 
 def package_path() -> pathlib.Path:
@@ -36,19 +47,33 @@ def find_git_repo(src_path: Optional[pathlib.Path] = None) -> Optional[pathlib.P
             return parent
     return None
 
-def is_dist_package(src_path: Optional[pathlib.Path] = None) -> bool:
-    if src_path is None:
-        # Check Moonraker's source path
-        src_path = source_path()
+def is_dist_package(item_path: Optional[pathlib.Path] = None) -> bool:
+    """
+    Check if the supplied path exists within a python dist installation or
+    site installation.
+    """
+    if item_path is None:
+        # Check Moonraker's package path
+        item_path = package_path()
         if hasattr(site, "getsitepackages"):
             # The site module is present, get site packages for Moonraker's venv.
             # This is more "correct" than the fallback method.
-            site_dirs = site.getsitepackages()
-            return str(src_path) in site_dirs
-    # Make an assumption based on the source path.  If its name is
-    # site-packages or dist-packages then presumably it is an
+            for site_dir in site.getsitepackages():
+                site_path = pathlib.Path(site_dir)
+                try:
+                    if site_path.samefile(item_path.parent):
+                        return True
+                except Exception:
+                    pass
+    # Make an assumption based on the item and/or its parents.  If a folder
+    # is named site-packages or dist-packages then presumably it is an
     # installed package
-    return src_path.name in ["dist-packages", "site-packages"]
+    if item_path.name in ("dist-packages", "site-packages"):
+        return True
+    for parent in item_path.parents:
+        if parent.name in ("dist-packages", "site-packages"):
+            return True
+    return False
 
 def package_version() -> Optional[str]:
     try:
@@ -86,3 +111,86 @@ def get_asset_path() -> Optional[pathlib.Path]:
         # Somehow running in a zipapp.  This is an error.
         return None
     return asset_path
+
+def _load_release_info_json(dist_info: Distribution) -> Optional[Dict[str, Any]]:
+    files = dist_info.files
+    if files is None:
+        return None
+    for dist_file in files:
+        if dist_file.parts[0] in ["..", "/"]:
+            continue
+        if dist_file.name == "release_info":
+            pkg = dist_file.parts[0]
+            logging.info(f"Package {pkg}: Detected release_info json file")
+            try:
+                return json.loads(dist_file.read_text())
+            except Exception:
+                logging.exception(f"Failed to load release_info from {dist_file}")
+    return None
+
+def _load_direct_url_json(dist_info: Distribution) -> Optional[Dict[str, Any]]:
+    ret: Optional[str] = dist_info.read_text("direct_url.json")
+    if ret is None:
+        return None
+    try:
+        direct_url: Dict[str, Any] = json.loads(ret)
+    except json.JSONDecodeError:
+        return None
+    return direct_url
+
+def normalize_project_name(name: str) -> str:
+    return re.sub(r"[-_.]+", "-", name).lower().replace('-', '_')
+
+def load_distribution_info(
+    venv_path: pathlib.Path, project_name: str
+) -> PackageInfo:
+    proj_name_normalized = normalize_project_name(project_name)
+    site_items = venv_path.joinpath("lib").glob("python*/site-packages/")
+    lib_paths = [str(p) for p in site_items if p.is_dir()]
+    for dist_info in Distribution.discover(name=project_name, path=lib_paths):
+        metadata = dist_info.metadata
+        if metadata is None:
+            continue
+        if not isinstance(dist_info, PathDistribution):
+            logging.info(f"Project {dist_info.name} not a PathDistribution")
+            continue
+        metaname = normalize_project_name(metadata["Name"] or "")
+        if metaname != proj_name_normalized:
+            continue
+        release_info = _load_release_info_json(dist_info)
+        install_info = _load_direct_url_json(dist_info)
+        return PackageInfo(
+            dist_info, metadata, release_info, install_info
+        )
+    raise ServerError(f"Failed to find distribution info for project {project_name}")
+
+def is_vitualenv_project(
+    venv_path: Optional[pathlib.Path] = None,
+    pkg_path: Optional[pathlib.Path] = None,
+    project_name: str = "moonraker"
+) -> bool:
+    if venv_path is None:
+        venv_path = pathlib.Path(sys.exec_prefix)
+    if pkg_path is None:
+        pkg_path = package_path()
+    if not pkg_path.exists():
+        return False
+    try:
+        pkg_info = load_distribution_info(venv_path, project_name)
+    except Exception:
+        return False
+    site_path = pathlib.Path(str(pkg_info.dist_info.locate_file("")))
+    for parent in pkg_path.parents:
+        try:
+            if site_path.samefile(parent):
+                return True
+        except Exception:
+            pass
+    return True
+
+@dataclass(frozen=True)
+class PackageInfo:
+    dist_info: Distribution
+    metadata: PackageMetadata
+    release_info: Optional[Dict[str, Any]]
+    direct_url_data: Optional[Dict[str, Any]]