# pylint: disable=W0718
"""PluginMeta views."""
from http import HTTPStatus
from typing import Optional
from urllib.parse import urljoin
import yaml
from django.core.exceptions import ValidationError
from django.core.handlers.asgi import ASGIRequest
from django.http import HttpResponseRedirect, JsonResponse
from django.shortcuts import redirect
from rest_framework import status
from rest_framework.parsers import FileUploadParser
from rest_framework.response import Response
from smarter.apps.account.models import User, UserProfile, get_resolved_user
from smarter.apps.plugin.contract.models import PluginApiPluginMeta
from smarter.apps.plugin.manifest.controller import (
PLUGIN_MAP,
SAM_MAP,
PluginController,
)
from smarter.apps.plugin.models import PluginDataValueError, PluginMeta
from smarter.apps.plugin.plugin.base import PluginBase
from smarter.apps.plugin.serializers import PluginMetaSerializer
from smarter.apps.plugin.utils import add_example_plugins
from smarter.common.conf import smarter_settings
from smarter.common.exceptions import SmarterValueError
from smarter.lib import json, logging
from smarter.lib.cache import cache_results
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.drf.views.contract import ContractListMixin
from smarter.lib.drf.views.token_authentication_helpers import (
SmarterAuthenticatedAPIView,
SmarterAuthenticatedListAPIView,
)
from smarter.lib.typing import SmarterDictType
logger = logging.getLogger(__name__)
[docs]
class PluginView(SmarterAuthenticatedAPIView):
"""Plugin view for smarter api."""
[docs]
def get(self, request: ASGIRequest, plugin_id):
if not waffle.switch_is_active(SmarterWaffleSwitches.ALLOW_API_GET):
logger.error(
"%s.get() is not allowed because %s switch is inactive.",
self.formatted_class_name,
SmarterWaffleSwitches.ALLOW_API_GET,
)
return JsonResponse(
{"detail": "GET method is not allowed for this endpoint."}, status=HTTPStatus.METHOD_NOT_ALLOWED
)
return get_plugin(request, plugin_id)
[docs]
def put(self, request: ASGIRequest, plugin_id: Optional[int] = None):
if plugin_id is not None:
return update_plugin(request, plugin_id)
return create_plugin(request)
[docs]
def post(self, request: ASGIRequest, plugin_id: Optional[int] = None):
if plugin_id is not None:
return update_plugin(request, plugin_id)
return create_plugin(request)
[docs]
def patch(self, request: ASGIRequest, plugin_id: Optional[int] = None):
return update_plugin(request, plugin_id)
[docs]
def delete(self, request: ASGIRequest, plugin_id):
return delete_plugin(request, plugin_id)
[docs]
class PluginCloneView(SmarterAuthenticatedAPIView):
"""Plugin clone view for smarter api."""
[docs]
def post(self, request: ASGIRequest, plugin_id, new_name):
user = get_resolved_user(request.user)
if not user:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.UNAUTHORIZED)
user_profile = UserProfile.get_cached_object(user=user) # type: ignore
plugin_controller = PluginController(
user_profile=user_profile,
plugin_meta=PluginMeta.get_cached_object(pk=plugin_id), # type: ignore[attr-defined]
)
if not plugin_controller or not plugin_controller.plugin:
return JsonResponse(
{
"error": f"PluginController could not be created for plugin_id: {plugin_id}, user_profile: {user_profile}"
},
status=HTTPStatus.BAD_REQUEST,
)
plugin = plugin_controller.plugin
new_id = plugin.clone(new_name)
return redirect("/plugins/" + str(new_id) + "/")
[docs]
class PluginListView(ContractListMixin, SmarterAuthenticatedListAPIView):
"""Plugins list view for smarter api."""
serializer_class = PluginMetaSerializer
contract_model = PluginApiPluginMeta
[docs]
def item(self, row):
"""A plugin of the response, whose ``canDelete`` is the requesting user's."""
return PluginApiPluginMeta.from_model(row, self.request.user).model_dump(mode="json")
[docs]
def get_queryset(self):
plugins = PluginMeta.objects.with_ownership_permission_for(user=self.user_profile.user).order_by("-created_at") # type: ignore
return plugins
[docs]
class AddPluginExamplesView(SmarterAuthenticatedAPIView):
"""Add example plugins to a user profile."""
[docs]
def post(self, request: ASGIRequest, user_id=None):
@cache_results()
def cached_user_by_id(user_id: int) -> Optional[User]:
"""Retrieve User by ID with caching."""
try:
return User.objects.get(id=user_id)
except User.DoesNotExist:
return None
try:
user = cached_user_by_id(user_id) if user_id else request.user
user_profile = UserProfile.get_cached_object(user=user) # type: ignore
except User.DoesNotExist:
return Response({"error": "User not found"}, status=status.HTTP_400_BAD_REQUEST)
try:
add_example_plugins(user_profile=user_profile)
except Exception as e:
return Response(
{"error": "Internal error", "exception": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
return HttpResponseRedirect("/v1/plugins/")
[docs]
class PluginUploadView(SmarterAuthenticatedAPIView):
"""Plugin view for smarter api."""
parser_class = (FileUploadParser,)
[docs]
@staticmethod
def parse_yaml_file(data):
if type(data) in [dict, list]:
return data
if isinstance(data, str):
data = data.encode("utf-8")
try:
return yaml.safe_load(data)
except yaml.YAMLError:
pass
try:
return json.loads(data)
except json.JSONDecodeError:
pass
raise SmarterValueError("Invalid data format: expected JSON or YAML.")
def _create(self, request: ASGIRequest):
data = self.parse_yaml_file(data=request.body.decode("utf-8"))
return create_plugin(request=request, data=data)
[docs]
def put(self, request: ASGIRequest):
return self._create(request)
[docs]
def post(self, request: ASGIRequest):
return self._create(request)
# -----------------------------------------------------------------------
# handlers for plugins
# -----------------------------------------------------------------------
[docs]
def get_plugin(request, plugin_id):
"""
Get a plugin's manifest, by id, if the user may read the plugin.
A plugin that the user may not read is not found, so that the plugins of other accounts are not
disclosed. Errors are Django REST framework's ``{"detail": ...}``.
"""
plugin: Optional[PluginBase] = None
try:
user_profile = UserProfile.get_cached_object(user=request.user)
except UserProfile.DoesNotExist:
return JsonResponse({"detail": "User not found"}, status=HTTPStatus.NOT_FOUND)
readable = PluginMeta.objects.with_read_permission_for(user=request.user) # type: ignore[attr-defined]
if not readable.filter(pk=plugin_id).exists():
return JsonResponse({"detail": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
try:
plugin_controller = PluginController(
user_profile=user_profile,
plugin_meta=PluginMeta.get_cached_object(pk=plugin_id), # type: ignore[attr-defined]
)
if not plugin_controller or not plugin_controller.plugin:
raise PluginDataValueError(
f"PluginController could not be created for plugin_id: {plugin_id}, user_profile: {user_profile}"
)
plugin = plugin_controller.plugin
except PluginMeta.DoesNotExist:
return JsonResponse({"detail": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
except ValidationError as e:
return JsonResponse({"detail": e.message}, status=HTTPStatus.BAD_REQUEST)
except Exception as e:
return JsonResponse({"detail": f"Internal error: {e}"}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
if plugin.ready:
return JsonResponse(plugin.to_json(), status=HTTPStatus.OK)
return JsonResponse({"detail": "Internal plugin error."}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
[docs]
def create_plugin(request, data: Optional[SmarterDictType] = None):
"""Create a plugin from a json representation in the body of the request."""
try:
user_profile = UserProfile.get_cached_object(user=request.user)
except UserProfile.DoesNotExist:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.UNAUTHORIZED)
if not data:
try:
data = request.data
if not isinstance(data, dict):
return JsonResponse(
{"error": f"Invalid request data. Expected a JSON dict in request body but received {type(data)}"},
status=HTTPStatus.BAD_REQUEST,
)
except Exception as e:
return JsonResponse({"error": "Invalid request data", "exception": str(e)}, status=HTTPStatus.BAD_REQUEST)
if "user_profile" not in data:
data["user_profile"] = user_profile
try:
plugin_controller = PluginController(
user_profile=user_profile,
manifest=data, # type: ignore[arg-type]
)
if not plugin_controller or not plugin_controller.plugin:
raise PluginDataValueError(
f"PluginController could not be created for data: {data}, user_profile: {user_profile}"
)
plugin = plugin_controller.plugin
except ValidationError as e:
return JsonResponse({"error": e.message}, status=HTTPStatus.BAD_REQUEST)
except Exception as e:
return JsonResponse({"error": "Internal error", "exception": str(e)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
base_url = f"{smarter_settings.api_schema}://{request.get_host()}/"
plugins_api_url = urljoin(base_url, "/api/v1/plugins/")
return HttpResponseRedirect(plugins_api_url + str(plugin.id) + "/")
[docs]
def update_plugin(request: ASGIRequest, plugin_id: Optional[int] = None):
"""
Update a plugin from a json representation in the body of the request.
:param plugin_id: The id of the plugin to update, from the url. If provided, it must
be the plugin that the manifest names.
"""
user = get_resolved_user(request.user)
if not user:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.UNAUTHORIZED)
try:
user_profile = UserProfile.get_cached_object(user=user) # type: ignore
except UserProfile.DoesNotExist:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.UNAUTHORIZED)
if not user_profile:
return JsonResponse({"error": "User profile not found"}, status=HTTPStatus.UNAUTHORIZED)
try:
# a DRF request has already parsed its body; a plain django request has not.
data = getattr(request, "data", None)
if data is None:
data = PluginUploadView.parse_yaml_file(request.body.decode("utf-8"))
except Exception as e:
return JsonResponse({"error": "Invalid request data", "exception": str(e)}, status=HTTPStatus.BAD_REQUEST)
if not isinstance(data, dict) or not data:
return JsonResponse(
{"error": f"Invalid request data. Expected a JSON dict in request body but received {type(data)}"},
status=HTTPStatus.BAD_REQUEST,
)
data = dict(data)
kind = data.get("kind")
if kind not in PLUGIN_MAP:
return JsonResponse(
{"error": f"Invalid manifest kind {kind}. Expected one of: {list(PLUGIN_MAP)}"},
status=HTTPStatus.BAD_REQUEST,
)
manifest_name = (data.get("metadata") or {}).get("name")
if plugin_id is not None:
try:
plugin_meta = PluginMeta.get_cached_object(pk=plugin_id) # type: ignore[attr-defined]
except PluginMeta.DoesNotExist:
plugin_meta = None
if not plugin_meta:
return JsonResponse({"error": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
if manifest_name != plugin_meta.name:
return JsonResponse(
{"error": f"Manifest name {manifest_name} does not match plugin {plugin_id}: {plugin_meta.name}"},
status=HTTPStatus.BAD_REQUEST,
)
elif not PluginMeta.objects.filter(name=manifest_name, user_profile=user_profile).exists():
return JsonResponse({"error": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
try:
manifest = SAM_MAP[kind](**data) # type: ignore[arg-type]
# a PluginController prefers the stored PluginMeta to the manifest, so
# the plugin is instantiated from the manifest, which updates it.
plugin = PLUGIN_MAP[kind](manifest=manifest, user_profile=user_profile) # type: ignore[call-arg]
if not plugin or not plugin.ready:
raise PluginDataValueError(f"Plugin {manifest_name} is not ready after update.")
except ValidationError as e:
return JsonResponse({"error": e.message}, status=HTTPStatus.BAD_REQUEST)
except Exception as e:
return JsonResponse({"error": "Internal error", "exception": str(e)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
return HttpResponseRedirect(request.path_info)
[docs]
def delete_plugin(request, plugin_id):
"""Delete a plugin by id."""
try:
user_profile = UserProfile.get_cached_object(user=request.user)
except UserProfile.DoesNotExist:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.UNAUTHORIZED)
try:
plugin_controller = PluginController(
user_profile=user_profile,
plugin_meta=PluginMeta.get_cached_object(pk=plugin_id), # type: ignore[attr-defined]
)
if not plugin_controller or not plugin_controller.plugin:
raise PluginDataValueError(
f"PluginController could not be created for plugin_id: {plugin_id}, user_profile: {user_profile}"
)
plugin = plugin_controller.plugin
except PluginMeta.DoesNotExist:
return JsonResponse({"error": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
except Exception as e:
return JsonResponse({"error": "Internal error", "exception": str(e)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
try:
plugin.delete()
except Exception as e:
return JsonResponse({"error": "Internal error", "exception": str(e)}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
plugins_path = request.path_info.rsplit("/", 2)[0]
return HttpResponseRedirect(plugins_path)