(
self,
*,
sync_db: bool,
force_reload: bool,
previous_packages: Dict[str, str],
previous_aliases: Dict[str, str],
previous_paths: Dict[str, str],
token: int,
)
| 106 | close_old_connections() |
| 107 | |
| 108 | def _discover_plugins_impl( |
| 109 | self, |
| 110 | *, |
| 111 | sync_db: bool, |
| 112 | force_reload: bool, |
| 113 | previous_packages: Dict[str, str], |
| 114 | previous_aliases: Dict[str, str], |
| 115 | previous_paths: Dict[str, str], |
| 116 | token: int, |
| 117 | ) -> Dict[str, LoadedPlugin]: |
| 118 | try: |
| 119 | configs: Optional[Dict[str, PluginConfig]] = None |
| 120 | try: |
| 121 | configs = {c.key: c for c in PluginConfig.objects.all()} |
| 122 | except Exception: |
| 123 | # DB might not be ready; treat all plugins as untrusted |
| 124 | configs = None |
| 125 | |
| 126 | new_registry: Dict[str, LoadedPlugin] = {} |
| 127 | new_packages: Dict[str, str] = {} |
| 128 | new_aliases: Dict[str, str] = {} |
| 129 | for entry in sorted(os.listdir(self.plugins_dir)): |
| 130 | path = os.path.join(self.plugins_dir, entry) |
| 131 | if not os.path.isdir(path): |
| 132 | continue |
| 133 | |
| 134 | has_pkg = os.path.exists(os.path.join(path, "__init__.py")) |
| 135 | has_pluginpy = os.path.exists(os.path.join(path, "plugin.py")) |
| 136 | if not (has_pkg or has_pluginpy): |
| 137 | continue |
| 138 | |
| 139 | plugin_key = entry.replace(" ", "_").lower() |
| 140 | alias_name = self._resolve_alias_name(entry, path) |
| 141 | |
| 142 | if force_reload: |
| 143 | prev_alias = previous_aliases.get(plugin_key) |
| 144 | if prev_alias: |
| 145 | self._unload_alias(prev_alias) |
| 146 | prev_path = previous_paths.get(plugin_key) |
| 147 | if prev_path: |
| 148 | self._unload_path_modules(prev_path) |
| 149 | |
| 150 | cfg = configs.get(plugin_key) if configs else None |
| 151 | enabled = bool(cfg and cfg.enabled) |
| 152 | trusted = bool(cfg and (cfg.ever_enabled or cfg.enabled)) |
| 153 | |
| 154 | manifest, has_manifest = self._read_manifest(path) |
| 155 | legacy = not has_manifest |
| 156 | manifest_name = None |
| 157 | manifest_version = None |
| 158 | manifest_description = None |
| 159 | manifest_author = None |
| 160 | manifest_help_url = None |
| 161 | manifest_fields: List[Dict[str, Any]] = [] |
| 162 | manifest_actions: List[Dict[str, Any]] = [] |
| 163 | if has_manifest and isinstance(manifest, dict): |
| 164 | manifest_name = manifest.get("name") if isinstance(manifest.get("name"), str) else None |
| 165 | manifest_version = manifest.get("version") if isinstance(manifest.get("version"), str) else None |
no test coverage detected