import glob import importlib.metadata import os import subprocess from typing import Dict, Type from django.conf import settings as s from envipy_plugins import Descriptor, Classifier, Property def is_installed(package_name): try: importlib.metadata.version(package_name) return True except importlib.metadata.PackageNotFoundError: return False def install_wheel(wheel_path): print(f"Installing wheel: {wheel_path}") subprocess.check_call(["uv", "pip", "install", wheel_path]) def extract_package_name_from_wheel(wheel_filename): # Example: my_plugin-0.1.0-py3-none-any.whl -> my_plugin return wheel_filename.split('-')[0] def ensure_plugins_installed(): wheel_files = glob.glob(os.path.join(s.PLUGIN_DIR, '*.whl')) for wheel_path in wheel_files: wheel_filename = os.path.basename(wheel_path) package_name = extract_package_name_from_wheel(wheel_filename) if not is_installed(package_name): install_wheel(wheel_path) else: # print(f"Plugin already installed: {package_name}") pass def discover_plugins(_cls: Type = None) -> Dict[str, Type]: ensure_plugins_installed() plugins = {} for entry_point in importlib.metadata.entry_points(group='enviPy_plugins'): try: plugin_class = entry_point.load() if _cls: if issubclass(plugin_class, _cls): instance = plugin_class() plugins[instance.name()] = instance else: if ( issubclass(plugin_class, Classifier) or issubclass(plugin_class, Descriptor) or issubclass(plugin_class, Property) ): instance = plugin_class() plugins[instance.name()] = instance except Exception as e: print(f"Error loading plugin {entry_point.name}: {e}") return plugins