(self, request)
| 388 | |
| 389 | class PluginImportAPIView(PluginAuthMixin, APIView): |
| 390 | def post(self, request): |
| 391 | file: UploadedFile = request.FILES.get("file") |
| 392 | if not file: |
| 393 | return Response({"success": False, "error": "Missing 'file' upload"}, status=status.HTTP_400_BAD_REQUEST) |
| 394 | |
| 395 | # Manual imports default to non-overwrite; require explicit flag to replace existing plugins |
| 396 | overwrite_flag = bool(request.data.get("overwrite")) |
| 397 | |
| 398 | pm = PluginManager.get() |
| 399 | result = _install_plugin_from_zip( |
| 400 | file, pm.plugins_dir, |
| 401 | file_name=getattr(file, "name", "plugin.zip"), |
| 402 | allow_overwrite=overwrite_flag, |
| 403 | ) |
| 404 | if not result["success"]: |
| 405 | return Response( |
| 406 | {"success": False, "error": result["error"]}, |
| 407 | status=status.HTTP_400_BAD_REQUEST, |
| 408 | ) |
| 409 | |
| 410 | plugin_key = result["plugin_key"] |
| 411 | |
| 412 | # Ensure DB config exists (untrusted plugins are registered without loading) |
| 413 | was_managed = False |
| 414 | try: |
| 415 | cfg, _ = PluginConfig.objects.get_or_create( |
| 416 | key=plugin_key, |
| 417 | defaults={ |
| 418 | "name": plugin_key, |
| 419 | "version": "", |
| 420 | "description": "", |
| 421 | "settings": {}, |
| 422 | }, |
| 423 | ) |
| 424 | # Manual install always breaks the managed relationship |
| 425 | if cfg and cfg.source_repo_id: |
| 426 | was_managed = True |
| 427 | cfg.source_repo = None |
| 428 | cfg.slug = "" |
| 429 | cfg.save(update_fields=["source_repo", "slug", "updated_at"]) |
| 430 | logger.info("Plugin '%s' manually replaced - cleared managed repo link", plugin_key) |
| 431 | except Exception: |
| 432 | cfg = None |
| 433 | |
| 434 | # Reload discovery to register the plugin (trusted plugins will load) |
| 435 | pm.discover_plugins(force_reload=True) |
| 436 | plugin_entry = None |
| 437 | try: |
| 438 | plugin_entry = next((p for p in pm.list_plugins() if p.get("key") == plugin_key), None) |
| 439 | except Exception: |
| 440 | plugin_entry = None |
| 441 | |
| 442 | if not plugin_entry: |
| 443 | logo_path = os.path.join(pm.plugins_dir, plugin_key, "logo.png") |
| 444 | logo_url = f"/api/plugins/plugins/{plugin_key}/logo/" if os.path.isfile(logo_path) else None |
| 445 | legacy = not os.path.isfile(os.path.join(pm.plugins_dir, plugin_key, "plugin.json")) |
| 446 | plugin_entry = { |
| 447 | "key": plugin_key, |
nothing calls this directly
no test coverage detected