From a101815a47c3ebfc3e11f43ae4408df21076f79f Mon Sep 17 00:00:00 2001 From: Eric Callahan Date: Sat, 4 Mar 2023 11:14:19 -0500 Subject: [PATCH] utils: add support for extracting project dist info Signed-off-by: Eric Callahan --- moonraker/utils/__init__.py | 6 +- moonraker/utils/exceptions.py | 10 +++ moonraker/utils/source_info.py | 126 ++++++++++++++++++++++++++++++--- 3 files changed, 128 insertions(+), 14 deletions(-) create mode 100644 moonraker/utils/exceptions.py diff --git a/moonraker/utils/__init__.py b/moonraker/utils/__init__.py index 5a216a3..0a9c21f 100644 --- a/moonraker/utils/__init__.py +++ b/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() diff --git a/moonraker/utils/exceptions.py b/moonraker/utils/exceptions.py new file mode 100644 index 0000000..8223c0d --- /dev/null +++ b/moonraker/utils/exceptions.py @@ -0,0 +1,10 @@ +# Moonraker Exceptions +# +# Copyright (C) 2023 Eric Callahan +# +# 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 diff --git a/moonraker/utils/source_info.py b/moonraker/utils/source_info.py index d845199..061756d 100644 --- a/moonraker/utils/source_info.py +++ b/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]]