Merge branch 'main' into storefront-nuxt

This commit is contained in:
Egor Pavlovich Gorbunov 2025-11-14 17:22:00 +03:00
commit 244d94831e
16 changed files with 285 additions and 149 deletions

View file

@ -0,0 +1,179 @@
from contextlib import suppress
from django.core.exceptions import PermissionDenied
from django.utils.translation import gettext_lazy as _
from graphene import UUID, Boolean, Field, InputObjectType, List, NonNull, String
from engine.core.graphene import BaseMutation
from engine.core.graphene.object_types import ProductType
from engine.core.models import (
Attribute,
AttributeGroup,
AttributeValue,
Brand,
Category,
Product,
ProductTag,
)
from engine.core.utils.messages import permission_denied_message
def resolve_attributes(product, attributes):
for attr_input in attributes:
attribute = None
attr_uuid = attr_input.get("attribute_uuid")
if attr_uuid:
with suppress(Attribute.DoesNotExist):
attribute = Attribute.objects.get(uuid=attr_uuid)
if attribute is None:
group_name = attr_input.get("group_name")
attribute_name = attr_input.get("attribute_name")
value_type = attr_input.get("value_type") or "string"
if group_name and attribute_name:
group, _ = AttributeGroup.objects.get_or_create(name=group_name)
attribute, _ = Attribute.objects.get_or_create(
group=group,
name=attribute_name,
defaults={"value_type": value_type},
)
if attribute.value_type != value_type:
attribute.value_type = value_type
attribute.save(update_fields=["value_type"])
if attribute is not None:
AttributeValue.objects.update_or_create(
product=product,
attribute=attribute,
defaults={"value": str(attr_input.get("value", ""))},
)
def resolve_tags(product, tag_uuids):
tags = list(ProductTag.objects.filter(uuid__in=tag_uuids))
if tags:
product.tags.set(tags)
class AttributeInput(InputObjectType): # type: ignore[misc]
attribute_uuid = UUID(required=False, name="attributeUuid")
group_name = String(required=False, name="groupName")
attribute_name = String(required=False, name="attributeName")
value_type = String(required=False, name="valueType")
value = String(required=True)
class ProductInput(InputObjectType): # type: ignore[misc]
name = NonNull(String)
description = String(required=False)
is_digital = Boolean(required=False, name="isDigital")
partnumber = String(required=False)
sku = String(required=False)
category_uuid = NonNull(UUID, name="categoryUuid")
brand_uuid = UUID(required=False, name="brandUuid")
tag_uuids = List(UUID, required=False, name="tagUuids")
attributes = List(NonNull(AttributeInput), required=False)
# noinspection PyUnusedLocal,PyTypeChecker
class CreateProduct(BaseMutation):
class Meta:
description = _("create a product")
class Arguments:
product_data = NonNull(ProductInput, name="productData")
product = Field(ProductType)
@staticmethod
def mutate(parent, info, product_data): # type: ignore [override]
user = info.context.user
if not user.has_perm("core.add_product"):
raise PermissionDenied(permission_denied_message)
category = Category.objects.get(uuid=product_data["category_uuid"]) # type: ignore[index]
brand = None
if product_data.get("brand_uuid"):
with suppress(Brand.DoesNotExist): # type: ignore[name-defined]
brand = Brand.objects.get(uuid=product_data["brand_uuid"]) # type: ignore[index]
product = Product.objects.create(
name=product_data["name"],
description=product_data.get("description"),
is_digital=product_data.get("is_digital") or False,
partnumber=product_data.get("partnumber"),
sku=product_data.get("sku") or None,
category=category,
brand=brand,
)
resolve_tags(product, product_data.get("tag_uuids", []))
resolve_attributes(product, product_data.get("attributes", []))
return CreateProduct(product=product)
# noinspection PyUnusedLocal,PyTypeChecker
class UpdateProduct(BaseMutation):
class Meta:
description = _("create a product")
class Arguments:
product_uuid = UUID(required=True)
product_data = NonNull(ProductInput, name="productData")
product = Field(ProductType)
@staticmethod
def mutate(parent, info, product_uuid, product_data): # type: ignore [override]
user = info.context.user
if not user.has_perm("core.change_product"):
raise PermissionDenied(permission_denied_message)
product = Product.objects.get(uuid=product_uuid)
updates = {}
for field_in, model_field in (
("name", "name"),
("description", "description"),
("is_digital", "is_digital"),
("partnumber", "partnumber"),
("sku", "sku"),
):
if field_in in product_data and product_data[field_in] is not None:
updates[model_field] = product_data[field_in]
if product_data.get("category_uuid"):
product.category = Category.objects.get(uuid=product_data["category_uuid"]) # type: ignore[index]
if product_data.get("brand_uuid") is not None:
if product_data.get("brand_uuid"):
product.brand = Brand.objects.get(uuid=product_data["brand_uuid"]) # type: ignore[index]
else:
product.brand = None
for k, v in updates.items():
setattr(product, k, v)
product.save()
resolve_tags(product, product_data.get("tag_uuids", []))
resolve_attributes(product, product_data.get("attributes"))
return UpdateProduct(product=product)
# noinspection PyUnusedLocal,PyTypeChecker
class DeleteProduct(BaseMutation):
class Meta:
description = _("create a product")
class Arguments:
product_uuid = UUID(required=True)
ok = Boolean()
@staticmethod
def mutate(parent, info, product_uuid): # type: ignore [override]
user = info.context.user
if not user.has_perm("core.delete_product"):
raise PermissionDenied(permission_denied_message)
Product.objects.get(uuid=product_uuid).delete()
return DeleteProduct(ok=True)

View file

@ -15,13 +15,12 @@ from engine.core.graphene import BaseMutation
from engine.core.graphene.object_types import ( from engine.core.graphene.object_types import (
AddressType, AddressType,
BulkProductInput, BulkProductInput,
FeedbackType,
OrderType, OrderType,
ProductType,
SearchResultsType, SearchResultsType,
WishlistType, WishlistType,
FeedbackType,
) )
from engine.core.models import Address, Category, Order, Product, Wishlist, OrderProduct from engine.core.models import Address, Order, OrderProduct, Wishlist
from engine.core.utils import format_attributes, is_url_safe from engine.core.utils import format_attributes, is_url_safe
from engine.core.utils.caching import web_cache from engine.core.utils.caching import web_cache
from engine.core.utils.emailing import contact_us_email from engine.core.utils.emailing import contact_us_email
@ -105,7 +104,7 @@ class AddOrderProduct(BaseMutation):
raise Http404(_(f"order {order_uuid} not found")) from dne raise Http404(_(f"order {order_uuid} not found")) from dne
# noinspection PyUnusedLocal # noinspection PyUnusedLocal,PyTypeChecker
class RemoveOrderProduct(BaseMutation): class RemoveOrderProduct(BaseMutation):
class Meta: class Meta:
description = _("remove a product from the order") description = _("remove a product from the order")
@ -118,7 +117,7 @@ class RemoveOrderProduct(BaseMutation):
order = Field(OrderType) order = Field(OrderType)
@staticmethod @staticmethod
def mutate(parent, info, product_uuid, order_uuid, attributes=None) -> AddOrderProduct | None: # type: ignore [override] def mutate(parent, info, product_uuid, order_uuid, attributes=None): # type: ignore [override]
user = info.context.user user = info.context.user
try: try:
order = Order.objects.get(uuid=order_uuid) order = Order.objects.get(uuid=order_uuid)
@ -127,7 +126,7 @@ class RemoveOrderProduct(BaseMutation):
order = order.remove_product(product_uuid=product_uuid, attributes=format_attributes(attributes)) order = order.remove_product(product_uuid=product_uuid, attributes=format_attributes(attributes))
return AddOrderProduct(order=order) return RemoveOrderProduct(order=order)
except Order.DoesNotExist as dne: except Order.DoesNotExist as dne:
raise Http404(_(f"order {order_uuid} not found")) from dne raise Http404(_(f"order {order_uuid} not found")) from dne
@ -577,67 +576,6 @@ class FeedbackProductAction(BaseMutation):
raise Http404(_(f"order product {order_product_uuid} not found")) from dne raise Http404(_(f"order product {order_product_uuid} not found")) from dne
# noinspection PyUnusedLocal,PyTypeChecker
class CreateProduct(BaseMutation):
class Arguments:
name = String(required=True)
description = String()
category_uuid = UUID(required=True)
product = Field(ProductType)
@staticmethod
def mutate(parent, info, name, category_uuid, description=None): # type: ignore [override]
if not info.context.user.has_perm("core.add_product"):
raise PermissionDenied(permission_denied_message)
category = Category.objects.get(uuid=category_uuid)
product = Product.objects.create(name=name, description=description, category=category)
return CreateProduct(product=product)
# noinspection PyUnusedLocal,PyTypeChecker
class UpdateProduct(BaseMutation):
class Arguments:
uuid = UUID(required=True)
name = String()
description = String()
category_uuid = UUID()
product = Field(ProductType)
@staticmethod
def mutate(parent, info, uuid, name=None, description=None, category_uuid=None): # type: ignore [override]
user = info.context.user
if not user.has_perm("core.change_product"):
raise PermissionDenied(permission_denied_message)
product = Product.objects.get(uuid=uuid)
if name:
product.name = name
if description:
product.description = description
if category_uuid:
product.category = Category.objects.get(uuid=category_uuid)
product.save()
return UpdateProduct(product=product)
# noinspection PyUnusedLocal,PyTypeChecker
class DeleteProduct(BaseMutation):
class Arguments:
uuid = UUID(required=True)
ok = Boolean()
@staticmethod
def mutate(parent, info, uuid): # type: ignore [override]
user = info.context.user
if not user.has_perm("core.delete_product"):
raise PermissionDenied(permission_denied_message)
product = Product.objects.get(uuid=uuid)
product.delete()
return DeleteProduct(ok=True)
# noinspection PyUnusedLocal,PyTypeChecker # noinspection PyUnusedLocal,PyTypeChecker
class CreateAddress(BaseMutation): class CreateAddress(BaseMutation):
class Arguments: class Arguments:

View file

@ -31,9 +31,7 @@ from engine.core.graphene.mutations import (
CacheOperator, CacheOperator,
ContactUs, ContactUs,
CreateAddress, CreateAddress,
CreateProduct,
DeleteAddress, DeleteAddress,
DeleteProduct,
FeedbackProductAction, FeedbackProductAction,
RemoveAllOrderProducts, RemoveAllOrderProducts,
RemoveAllWishlistProducts, RemoveAllWishlistProducts,
@ -42,7 +40,6 @@ from engine.core.graphene.mutations import (
RemoveWishlistProduct, RemoveWishlistProduct,
RequestCursedURL, RequestCursedURL,
Search, Search,
UpdateProduct,
) )
from engine.core.graphene.object_types import ( from engine.core.graphene.object_types import (
AddressType, AddressType,
@ -369,9 +366,6 @@ class Mutation(ObjectType):
reset_password = ResetPassword.Field() reset_password = ResetPassword.Field()
confirm_reset_password = ConfirmResetPassword.Field() confirm_reset_password = ConfirmResetPassword.Field()
buy_product = BuyProduct.Field() buy_product = BuyProduct.Field()
create_product = CreateProduct.Field()
update_product = UpdateProduct.Field()
delete_product = DeleteProduct.Field()
create_address = CreateAddress.Field() create_address = CreateAddress.Field()
delete_address = DeleteAddress.Field() delete_address = DeleteAddress.Field()
autocomplete_address = AutocompleteAddress.Field() autocomplete_address = AutocompleteAddress.Field()

View file

@ -1,8 +1,33 @@
from django.test import TestCase from django.test import TestCase
from rest_framework.test import APIClient from rest_framework.test import APIClient
from engine.vibes_auth.models import User
from engine.vibes_auth.serializers import TokenObtainPairSerializer
class DRFCoreViewsTests(TestCase): class DRFCoreViewsTests(TestCase):
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.client = APIClient() self.client = APIClient()
self.superuser_password = "Str0ngPass!word1"
self.superuser = User.objects.create(
email="test-superuser@email.com",
password=self.superuser_password,
is_active=True,
is_verified=True,
is_superuser=True,
is_staff=True,
)
self.user_password = "Str0ngPass!word2"
self.user = User.objects.create(
email="test-superuser@email.com", password=self.user_password, is_active=True, is_verified=True
)
def _get_authorization_token(self, user):
serializer = TokenObtainPairSerializer(
data={"email": user.email, "password": self.superuser_password if user.is_superuser else self.user_password}
)
serializer.is_valid(raise_exception=True)
return serializer.validated_data["access_token"]
# TODO: create tests for every possible HTTP method in core module with DRF stack

View file

@ -13,3 +13,6 @@ class GraphQLCoreTests(TestCase):
response = self.client.post(url, data=payload, content_type="application/json") response = self.client.post(url, data=payload, content_type="application/json")
self.assertEqual(response.status_code, 200, response.json()) self.assertEqual(response.status_code, 200, response.json())
return response.json() return response.json()
# TODO: create tests for every possible HTTP method in core module with Graphene stack

View file

@ -2,6 +2,7 @@ import logging
import traceback import traceback
from typing import Any from typing import Any
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema, extend_schema_view from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import status from rest_framework import status
from rest_framework.request import Request from rest_framework.request import Request
@ -18,21 +19,13 @@ logger = logging.getLogger(__name__)
@extend_schema_view(**DEPOSIT_SCHEMA) @extend_schema_view(**DEPOSIT_SCHEMA)
class DepositView(APIView): class DepositView(APIView):
"""Handles deposit operations. __doc__ = _( # type: ignore [assignment]
"This class provides an API endpoint to handle deposit transactions.\n"
This class provides an API endpoint to handle deposit transactions. "It supports the creation of a deposit transaction after validating the "
It supports the creation of a deposit transaction after validating the "provided data. If the user is not authenticated, an appropriate response "
provided data. If the user is not authenticated, an appropriate response "is returned. On successful validation and execution, a response "
is returned. On successful validation and execution, a response "with the transaction details is provided."
with the transaction details is provided. )
Attributes:
No attributes are declared at the class-level for this view.
Methods:
post: Processes the deposit request, validates the request data, ensures
user authentication, and creates a transaction.
"""
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response: def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
logger.debug(request.__dict__) logger.debug(request.__dict__)
@ -52,33 +45,25 @@ class DepositView(APIView):
@extend_schema(exclude=True) @extend_schema(exclude=True)
class CallbackAPIView(APIView): class CallbackAPIView(APIView):
""" __doc__ = _( # type: ignore [assignment]
Handles incoming callback requests to the API. "Handles incoming callback requests to the API.\n"
"This class processes and routes incoming HTTP POST requests to the appropriate "
This class processes and routes incoming HTTP POST requests to the appropriate "pgateway handler based on the provided gateway parameter. It is designed to handle "
gateway handler based on the provided gateway parameter. It is designed to handle "callback events coming from external systems and provide an appropriate HTTP response "
callback events coming from external systems and provide an appropriate HTTP response "indicating success or failure."
indicating success or failure. )
Attributes:
No additional attributes are defined for this class beyond what is
inherited from APIView.
Methods:
post(request, *args, **kwargs): Processes POST requests and routes them
based on the specified gateway. Handles exceptions gracefully by returning
a server error response if an unknown gateway or other issues occur.
"""
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response: def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
try: try:
transaction = Transaction.objects.get(uuid=str(kwargs.get("uuid"))) transaction = Transaction.objects.get(uuid=str(kwargs.get("uuid")))
if not transaction.gateway: if not transaction.gateway:
raise UnknownGatewayError() raise UnknownGatewayError(_(f"Transaction {transaction.uuid} has no gateway"))
gateway = transaction.gateway.get_integration_class_object(raise_exc=True) gateway_integration = transaction.gateway.get_integration_class_object(raise_exc=True)
gateway.process_callback(request.data) if not gateway_integration:
raise UnknownGatewayError(_(f"Gateway {transaction.gateway} has no integration"))
gateway_integration.process_callback(request.data)
return Response(status=status.HTTP_202_ACCEPTED) return Response(status=status.HTTP_202_ACCEPTED)
except Exception as e: except Exception as e:
return Response( return Response(
status=status.HTTP_500_INTERNAL_SERVER_ERROR, data={"error": f"{e}; {traceback.format_exc()}"} status=status.HTTP_500_INTERNAL_SERVER_ERROR, data={"error": str(e), "detail": traceback.format_exc()}
) )

View file

@ -204,9 +204,8 @@ class ObtainJSONWebToken(BaseMutation):
serializer = TokenObtainPairSerializer(data={"email": email, "password": password}) serializer = TokenObtainPairSerializer(data={"email": email, "password": password})
try: try:
serializer.is_valid(raise_exception=True) serializer.is_valid(raise_exception=True)
user = User.objects.get(email=email)
return ObtainJSONWebToken( return ObtainJSONWebToken(
user=user, user=serializer.validated_data["user"],
refresh_token=serializer.validated_data["refresh"], refresh_token=serializer.validated_data["refresh"],
access_token=serializer.validated_data["access"], access_token=serializer.validated_data["access"],
) )
@ -227,9 +226,9 @@ class RefreshJSONWebToken(BaseMutation):
try: try:
serializer.is_valid(raise_exception=True) serializer.is_valid(raise_exception=True)
return RefreshJSONWebToken( return RefreshJSONWebToken(
user=serializer.validated_data["user"],
access_token=serializer.validated_data["access"], access_token=serializer.validated_data["access"],
refresh_token=serializer.validated_data["refresh"], refresh_token=serializer.validated_data["refresh"],
user=User.objects.get(uuid=serializer.validated_data["user"]["uuid"]),
) )
except Exception as e: except Exception as e:
raise PermissionDenied(f"invalid refresh token provided: {e!s}") from e raise PermissionDenied(f"invalid refresh token provided: {e!s}") from e
@ -247,10 +246,8 @@ class VerifyJSONWebToken(BaseMutation):
serializer = TokenVerifySerializer(data={"token": token}) serializer = TokenVerifySerializer(data={"token": token})
with suppress(Exception): with suppress(Exception):
serializer.is_valid(raise_exception=True) serializer.is_valid(raise_exception=True)
user_uuid = serializer.validated_data["user"]["uuid"]
user = User.objects.get(pk=user_uuid)
# noinspection PyTypeChecker # noinspection PyTypeChecker
return VerifyJSONWebToken(token_is_valid=True, user=user) return VerifyJSONWebToken(token_is_valid=True, user=serializer.validated_data["user"])
detail = traceback.format_exc() if settings.DEBUG else "" detail = traceback.format_exc() if settings.DEBUG else ""
# noinspection PyTypeChecker # noinspection PyTypeChecker
return VerifyJSONWebToken(token_is_valid=False, user=None, detail=detail) return VerifyJSONWebToken(token_is_valid=False, user=None, detail=detail)

View file

@ -1,5 +1,3 @@
from __future__ import annotations
from contextlib import suppress from contextlib import suppress
from typing import Iterable from typing import Iterable

View file

@ -1,14 +1,13 @@
from __future__ import annotations
from typing import Any from typing import Any
from asgiref.sync import sync_to_async from asgiref.sync import sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer from channels.generic.websocket import AsyncJsonWebsocketConsumer
from drf_spectacular_websocket.decorators import extend_ws_schema
from django.conf import settings from django.conf import settings
from django.core.cache import cache from django.core.cache import cache
from django.utils import timezone from django.utils import timezone
from drf_spectacular_websocket.decorators import extend_ws_schema
from engine.vibes_auth.choices import SenderType, ThreadStatus
from engine.vibes_auth.docs.drf.messaging import ( from engine.vibes_auth.docs.drf.messaging import (
STAFF_INBOX_CONSUMER_SCHEMA, STAFF_INBOX_CONSUMER_SCHEMA,
THREAD_CONSUMER_SCHEMA, THREAD_CONSUMER_SCHEMA,
@ -23,7 +22,6 @@ from engine.vibes_auth.messaging.services import (
send_message, send_message,
) )
from engine.vibes_auth.models import ChatThread, User from engine.vibes_auth.models import ChatThread, User
from engine.vibes_auth.choices import SenderType, ThreadStatus
MAX_MESSAGE_LENGTH = 1028 MAX_MESSAGE_LENGTH = 1028
USER_SUPPORT_GROUP_NAME = "User Support" USER_SUPPORT_GROUP_NAME = "User Support"

View file

@ -1,9 +1,6 @@
from __future__ import annotations
import asyncio import asyncio
import logging import logging
from contextlib import suppress from contextlib import suppress
from typing import Optional
from aiogram import Bot, Dispatcher, Router, types from aiogram import Bot, Dispatcher, Router, types
from aiogram.enums import ParseMode from aiogram.enums import ParseMode
@ -25,14 +22,14 @@ def is_telegram_enabled() -> bool:
return bool(settings.TELEGRAM_TOKEN) return bool(settings.TELEGRAM_TOKEN)
def _get_bot() -> Optional["Bot"]: def _get_bot() -> Bot | None:
if not is_telegram_enabled(): if not is_telegram_enabled():
logger.warning("Telegram forwarder disabled: missing aiogram or TELEGRAM_TOKEN") logger.warning("Telegram forwarder disabled: missing aiogram or TELEGRAM_TOKEN")
return None return None
return Bot(token=settings.TELEGRAM_TOKEN, parse_mode=ParseMode.HTML) # type: ignore[arg-type] return Bot(token=settings.TELEGRAM_TOKEN, parse_mode=ParseMode.HTML) # type: ignore[arg-type]
def build_router() -> Optional["Router"]: def build_router() -> Router | None:
if not is_telegram_enabled(): if not is_telegram_enabled():
return None return None
@ -160,7 +157,7 @@ async def forward_thread_message_to_assigned_staff(thread_uuid: str, text: str)
if not is_telegram_enabled(): if not is_telegram_enabled():
return return
def _resolve_chat_and_chat_id() -> tuple[Optional[int], Optional[str]]: def _resolve_chat_and_chat_id() -> tuple[int | None, str | None]:
try: try:
t = ChatThread.objects.select_related("assigned_to").get(uuid=thread_uuid) t = ChatThread.objects.select_related("assigned_to").get(uuid=thread_uuid)
except ChatThread.DoesNotExist: except ChatThread.DoesNotExist:

View file

@ -1,5 +1,3 @@
from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from asgiref.sync import async_to_sync from asgiref.sync import async_to_sync

View file

@ -124,6 +124,7 @@ class DRFAuthViewsTests(TestCase):
stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True) stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True)
access = str(RefreshToken.for_user(stranger).access_token) access = str(RefreshToken.for_user(stranger).access_token)
# noinspection PyUnresolvedReferences
self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}")
url = reverse("vibes_auth:users-upload-avatar", kwargs={"pk": owner.pk}) url = reverse("vibes_auth:users-upload-avatar", kwargs={"pk": owner.pk})
@ -137,6 +138,7 @@ class DRFAuthViewsTests(TestCase):
stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True) stranger = User.objects.create_user(email="stranger@example.com", password="Str0ngPass!word", is_active=True)
access = str(RefreshToken.for_user(stranger).access_token) access = str(RefreshToken.for_user(stranger).access_token)
# noinspection PyUnresolvedReferences
self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}") self.client.credentials(HTTP_X_EVIBES_AUTH=f"Bearer {access}")
url = reverse("vibes_auth:users-merge-recently-viewed", kwargs={"pk": owner.pk}) url = reverse("vibes_auth:users-merge-recently-viewed", kwargs={"pk": owner.pk})

View file

@ -36,10 +36,10 @@ class TokenObtainPairView(TokenViewBase):
serializer_class = TokenObtainPairSerializer # type: ignore [assignment] serializer_class = TokenObtainPairSerializer # type: ignore [assignment]
_serializer_class = TokenObtainPairSerializer # type: ignore [assignment] _serializer_class = TokenObtainPairSerializer # type: ignore [assignment]
permission_classes: list[Type[BasePermission]] = [ permission_classes: list[Type[BasePermission]] = [ # type: ignore [assignment]
AllowAny, AllowAny,
] ]
authentication_classes: list[str] = [] authentication_classes: list[str] = [] # type: ignore [assignment]
@method_decorator(ratelimit(key="ip", rate="10/h")) @method_decorator(ratelimit(key="ip", rate="10/h"))
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response: def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
@ -59,10 +59,10 @@ class TokenRefreshView(TokenViewBase):
serializer_class = TokenRefreshSerializer # type: ignore [assignment] serializer_class = TokenRefreshSerializer # type: ignore [assignment]
_serializer_class = TokenRefreshSerializer # type: ignore [assignment] _serializer_class = TokenRefreshSerializer # type: ignore [assignment]
permission_classes: list[Type[BasePermission]] = [ permission_classes: list[Type[BasePermission]] = [ # type: ignore [assignment]
AllowAny, AllowAny,
] ]
authentication_classes: list[str] = [] authentication_classes: list[str] = [] # type: ignore [assignment]
@method_decorator(ratelimit(key="ip", rate="10/h")) @method_decorator(ratelimit(key="ip", rate="10/h"))
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response: def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
@ -77,10 +77,10 @@ class TokenVerifyView(TokenViewBase):
serializer_class = TokenVerifySerializer # type: ignore [assignment] serializer_class = TokenVerifySerializer # type: ignore [assignment]
_serializer_class = TokenVerifySerializer # type: ignore [assignment] _serializer_class = TokenVerifySerializer # type: ignore [assignment]
permission_classes: list[Type[BasePermission]] = [ permission_classes: list[Type[BasePermission]] = [ # type: ignore [assignment]
AllowAny, AllowAny,
] ]
authentication_classes: list[str] = [] authentication_classes: list[str] = [] # type: ignore [assignment]
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response: def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
try: try:

View file

@ -115,6 +115,7 @@ class RateLimitMiddleware:
def __call__(self, request): def __call__(self, request):
return self.get_response(request) return self.get_response(request)
# noinspection PyUnusedLocal
def process_exception(self, request, exception): def process_exception(self, request, exception):
if isinstance(exception, RatelimitedError): if isinstance(exception, RatelimitedError):
return JsonResponse( return JsonResponse(

View file

@ -1,6 +1,6 @@
from django.utils.translation import gettext_lazy as _ from django.utils.translation import gettext_lazy as _
from evibes.settings.base import EVIBES_VERSION, BASE_DOMAIN, STOREFRONT_DOMAIN from evibes.settings.base import EVIBES_VERSION, STOREFRONT_DOMAIN
from evibes.settings.constance import CONSTANCE_CONFIG from evibes.settings.constance import CONSTANCE_CONFIG
JAZZMIN_SETTINGS = { JAZZMIN_SETTINGS = {
@ -12,30 +12,51 @@ JAZZMIN_SETTINGS = {
"login_logo_dark": "logo.png", "login_logo_dark": "logo.png",
"site_logo_classes": "", "site_logo_classes": "",
"site_icon": "favicon.ico", "site_icon": "favicon.ico",
"welcome_sign": "Whoa! Only admins allowed here!", "welcome_sign": _("Whoa! Only admins allowed here!"),
"copyright": f"eVibes {EVIBES_VERSION} by Wiseless", "copyright": f"eVibes {EVIBES_VERSION} by Wiseless",
"search_model": None, "search_model": None,
"user_avatar": "avatar", "user_avatar": "avatar",
"topmenu_links": [ "topmenu_links": [
{"name": _("Home"), "url": "admin:index"},
{"name": _("Storefront"), "url": f"https://{STOREFRONT_DOMAIN}", "new_window": True}, # type: ignore [index]
{ {
"name": "GraphQL Docs", "name": _("Home"),
"url": f"https://api.{BASE_DOMAIN}/graphql", # type: ignore [index] "url": "admin:index",
"new_window": True, "new_window": False,
}, },
{ {
"name": "REST Docs", "name": _("Storefront"),
"url": f"https://api.{BASE_DOMAIN}/docs/swagger", # type: ignore [index] "url": f"https://{STOREFRONT_DOMAIN}",
"new_window": True, "new_window": False,
},
{
"name": "GraphQL Docs",
"url": "graphql-platform",
"new_window": False,
},
{
"name": "Swagger",
"url": "swagger-ui-platform",
"new_window": False,
},
{
"name": "Redoc",
"url": "redoc-ui-platform",
"new_window": False,
}, },
{ {
"name": _("Taskboard"), "name": _("Taskboard"),
"url": "https://plane.wiseless.xyz/spaces/issues/dd33cb0ab9b04ef08a10f7eefae6d90c/?board=kanban", "url": "https://plane.wiseless.xyz/spaces/issues/dd33cb0ab9b04ef08a10f7eefae6d90c/?board=kanban",
"new_window": True, "new_window": True,
}, },
{"name": "GitLab", "url": "https://gitlab.com/wiseless/evibes", "new_window": True}, {
{"name": _("Support"), "url": "https://t.me/fureunoir", "new_window": True}, "name": "GitLab",
"url": "https://gitlab.com/wiseless/evibes",
"new_window": True,
},
{
"name": _("Support"),
"url": "https://t.me/fureunoir",
"new_window": True,
},
], ],
"usermenu_links": [], "usermenu_links": [],
"show_sidebar": True, "show_sidebar": True,
@ -58,6 +79,6 @@ JAZZMIN_SETTINGS = {
} }
JAZZMIN_UI_TWEAKS = { JAZZMIN_UI_TWEAKS = {
"theme": "cyborg", "theme": "flatly",
"dark_mode_theme": "cyborg", "dark_mode_theme": "darkly",
} }