# pylint: disable=W0718
"""PluginMeta views."""
import logging
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.wsgi import WSGIRequest
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.account.utils import (
valid_resource_owners_for_user,
)
from smarter.apps.plugin.manifest.controller import PluginController
from smarter.apps.plugin.manifest.models.common.plugin.model import SAMPluginCommon
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
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.token_authentication_helpers import (
SmarterAuthenticatedAPIView,
SmarterAuthenticatedListAPIView,
)
logger = logging.getLogger(__name__)
[docs]
class PluginView(SmarterAuthenticatedAPIView):
"""Plugin view for smarter api."""
[docs]
def get(self, request: WSGIRequest, 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(
{"error": "GET method is not allowed for this endpoint."}, status=HTTPStatus.METHOD_NOT_ALLOWED
)
return get_plugin(request, plugin_id)
[docs]
def put(self, request: WSGIRequest):
return create_plugin(request)
[docs]
def post(self, request: WSGIRequest):
return create_plugin(request)
[docs]
def patch(self, request: WSGIRequest):
return update_plugin(request)
[docs]
def delete(self, request: WSGIRequest, plugin_id):
return delete_plugin(request, plugin_id)
[docs]
class PluginCloneView(SmarterAuthenticatedAPIView):
"""Plugin clone view for smarter api."""
[docs]
def post(self, request: WSGIRequest, 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,
account=user_profile.cached_account, # type: ignore[arg-type]
user=user_profile.cached_user, # type: ignore[arg-type]
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(SmarterAuthenticatedListAPIView):
"""Plugins list view for smarter api."""
serializer_class = PluginMetaSerializer
[docs]
def get_queryset(self):
plugins = PluginMeta.objects.filter(user_profile__account=self.account).order_by("-created_at")
valid_owners = valid_resource_owners_for_user(user_profile=self.user_profile)
plugins = plugins.filter(user_profile__in=valid_owners)
return plugins
[docs]
class AddPluginExamplesView(SmarterAuthenticatedAPIView):
"""Add example plugins to a user profile."""
[docs]
def post(self, request: WSGIRequest, 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: WSGIRequest):
data = self.parse_yaml_file(data=request.body.decode("utf-8"))
return create_plugin(request=request, data=data)
[docs]
def put(self, request: WSGIRequest):
return self._create(request)
[docs]
def post(self, request: WSGIRequest):
return self._create(request)
# -----------------------------------------------------------------------
# handlers for plugins
# -----------------------------------------------------------------------
[docs]
def get_plugin(request, plugin_id):
"""Get a plugin json representation by id."""
plugin: Optional[PluginBase] = None
try:
user_profile = UserProfile.get_cached_object(user=request.user)
except UserProfile.DoesNotExist:
return JsonResponse({"error": "User not found"}, status=HTTPStatus.NOT_FOUND)
try:
plugin_controller = PluginController(
user_profile=user_profile,
account=user_profile.cached_account, # type: ignore[arg-type]
user=user_profile.cached_user, # type: ignore[arg-type]
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 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)
if plugin.ready:
return JsonResponse(plugin.to_json(), status=HTTPStatus.OK)
return JsonResponse({"error": "Internal plugin error."}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
[docs]
def create_plugin(request, data: Optional[dict] = 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,
account=user_profile.cached_account, # type: ignore[arg-type]
user=user_profile.cached_user, # type: ignore[arg-type]
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: WSGIRequest):
"""update a plugin from a json representation in the body of the request."""
user = get_resolved_user(request.user)
data: str
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)
try:
data = request.body.decode("utf-8")
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,
)
data["user_profile"] = user_profile
except Exception as e:
return JsonResponse({"error": "Invalid request data", "exception": str(e)}, status=HTTPStatus.BAD_REQUEST)
try:
data = json.loads(data)
except json.JSONDecodeError:
return JsonResponse({"error": "Invalid JSON format"}, status=HTTPStatus.BAD_REQUEST)
if not user_profile:
return JsonResponse({"error": "User profile not found"}, status=HTTPStatus.UNAUTHORIZED)
try:
plugin_controller = PluginController(
user_profile=user_profile,
account=user_profile.cached_account, # type: ignore[arg-type]
user=user_profile.cached_user, # type: ignore[arg-type]
manifest=SAMPluginCommon(**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
if not plugin:
return JsonResponse({"error": "Plugin not found"}, status=HTTPStatus.NOT_FOUND)
if not data:
return JsonResponse({"error": "No data provided for update"}, status=HTTPStatus.BAD_REQUEST)
plugin.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,
account=user_profile.cached_account, # type: ignore[arg-type]
user=user_profile.cached_user, # type: ignore[arg-type]
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)