import uuid import orjson from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpResponse, HttpResponseBadRequest from django.shortcuts import render from django.views import View from rest_framework.parsers import FormParser, JSONParser from rest_framework.views import APIView from serde import ValidationError from core.forms import HookForm from core.lib.serde import drakdoo from core.models import Callback, Hook from core.util import logs log = logs.get_logger(__name__) def get_hooks(user): hooks = Hook.objects.filter(user=user) return hooks class HookAPI(APIView): parser_classes = [JSONParser] def post(self, request, hook_name): log.debug(f"HookAPI POST: {request.data}") # Try loading the JSON try: loaded_json = orjson.loads(request.data) except orjson.JSONDecodeError: return HttpResponseBadRequest("Invalid JSON") # Try validating the JSON try: hook_resp = drakdoo.BaseDrakdoo.from_dict(loaded_json) except ValidationError as e: log.error(f"HookAPI POST: {e}") return HttpResponseBadRequest(e) data = { "title": hook_resp.title, "message": hook_resp.message, "period": hook_resp.period, "market": hook_resp.market, "timestamp_sent": hook_resp.timestamp.sent, "timestamp_trade": hook_resp.timestamp.trade, "market_exchange": hook_resp.market.exchange, "market_item": hook_resp.market.item, "market_currency": hook_resp.market.currency, "market_contract": hook_resp.market.contract, } log.debug("HookAPI callback: data: %s", data) # Try getting the hook try: hook = Hook.objects.get(hook=hook_name) except Hook.DoesNotExist: return HttpResponseBadRequest("Hook does not exist.") # Create the callback object callback = Callback.objects.create(hook=hook, **data) callback.save() # Bump received count hook.received = hook.received + 1 hook.save() return HttpResponse("OK") def get(self, request, hook_name): hook = Hook.objects.get(name=hook_name) return_data = {"name": hook.name, "hook": hook.hook, "hook_id": hook.id} return HttpResponse(orjson.dumps(return_data), content_type="application/json") class Hooks(LoginRequiredMixin, View): allowed_types = ["modal", "widget", "window", "page"] window_content = "window-content/hooks.html" async def get(self, request, type): if type not in self.allowed_types: return HttpResponseBadRequest template_name = f"wm/{type}.html" unique = str(uuid.uuid4())[:8] hooks = get_hooks(request.user) context = { "title": f"Hooks ({type})", "unique": unique, "window_content": self.window_content, "items": hooks, "type": type, } return render(request, template_name, context) class HookAction(LoginRequiredMixin, APIView): allowed_types = ["modal", "widget", "window", "page"] window_content = "window-content/add-hook.html" parser_classes = [FormParser] def get(self, request, type, hook_id=None): """ Get the form for adding or editing a hook. :param hook_id: The id of the hook to edit. Optional. """ if type not in self.allowed_types: return HttpResponseBadRequest template_name = f"wm/{type}.html" unique = str(uuid.uuid4())[:8] if hook_id: try: hook = Hook.objects.get(id=hook_id, user=request.user) form = HookForm(instance=hook) except Hook.DoesNotExist: message = "Hook does not exist" message_class = "danger" context = { "message": message, "message_class": message_class, "window_content": self.window_content, } return render(request, template_name, context) else: form = HookForm() context = { "form": form, "hook_id": hook_id, "type": type, "unique": unique, "window_content": self.window_content, } return render(request, template_name, context) def put(self, request, type, hook_id=None): """ Add or edit a hook. :param hook_id: The id of the hook to edit. Optional. """ message = None message_class = "success" if hook_id: try: form = HookForm(request.data, instance=Hook.objects.get(id=hook_id)) except Hook.DoesNotExist: message = "Hook does not exist" message_class = "danger" context = { "message": message, "class": message_class, } return render(request, self.template_name, context) else: form = HookForm(request.data) if form.is_valid(): hook = form.save(commit=False) hook.user = request.user hook.save() if hook_id: message = f"Hook {hook_id} edited successfully" else: message = f"Hook {hook.id} added successfully" else: message = "Error adding hook" message_class = "danger" hooks = get_hooks(request.user) context = { "items": hooks, "type": type, } if message: context["message"] = message context["class"] = message_class template_name = "partials/hook-list.html" return render(request, template_name, context) def delete(self, request, type, hook_id): """ Delete a hook. :param hook_id: The id of the hook to delete. """ message = None message_class = "success" try: hook = Hook.objects.get(id=hook_id, user=request.user) hook.delete() message = "Hook deleted successfully" except Hook.DoesNotExist: message = "Error deleting hook" message_class = "danger" hooks = get_hooks(request.user) context = { "items": hooks, "type": type, } if message: context["message"] = message context["class"] = message_class template_name = "partials/hook-list.html" return render(request, template_name, context)