fureunoir/schon
fureunoir/schon
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Features: 1) Add mutations for product management (CreateProduct, UpdateProduct, DeleteProduct) with improved attribute and tag resolution; 2) Introduce enhanced GraphQL inputs for better product handling; 3) Add translation support to payment views and callback handling; 4) Refactor Telegram forwarder to use modern typing annotations (| syntax);

Browse source 
Fixes: 1) Remove redundant `from __future__ import annotations` in multiple files; 2) Correct callback integration to handle missing gateway scenarios gracefully; 3) Fix typos and update class references in tests and views;

Extra: Refactor deprecated mutation definitions and cleanup legacy product mutation references in GraphQL schema.
This commit is contained in:
Egor Pavlovich Gorbunov 2025-11-14 17:07:40 +03:00
parent 71fe47d428
commit aa8d40c781
12 changed files with 219 additions and 129 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
engine/core/graphene/dashboard_mutations/__init__.py Normal file
View file

179
engine/core/graphene/dashboard_mutations/product.py Normal file
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)

72
engine/core/graphene/mutations.py
View file

@ -15,13 +15,12 @@ from engine.core.graphene import BaseMutation
from engine.core.graphene.object_types import (
AddressType,
BulkProductInput,
FeedbackType,
OrderType,
ProductType,
SearchResultsType,
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.caching import web_cache
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
# noinspection PyUnusedLocal
# noinspection PyUnusedLocal,PyTypeChecker
class RemoveOrderProduct(BaseMutation):
class Meta:
description = _("remove a product from the order")
@ -118,7 +117,7 @@ class RemoveOrderProduct(BaseMutation):
order = Field(OrderType)
@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
try:
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))
return AddOrderProduct(order=order)
return RemoveOrderProduct(order=order)
except Order.DoesNotExist as 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
# 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
class CreateAddress(BaseMutation):
class Arguments:

6
engine/core/graphene/schema.py
View file

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

57
engine/payments/views.py
View file

@ -2,6 +2,7 @@ import logging
import traceback
from typing import Any
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema, extend_schema_view
from rest_framework import status
from rest_framework.request import Request
@ -18,21 +19,13 @@ logger = logging.getLogger(__name__)
@extend_schema_view(**DEPOSIT_SCHEMA)
class DepositView(APIView):
"""Handles deposit operations.
This class provides an API endpoint to handle deposit transactions.
It supports the creation of a deposit transaction after validating the
provided data. If the user is not authenticated, an appropriate response
is returned. On successful validation and execution, a response
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.
"""
__doc__ = _( # type: ignore [assignment]
"This class provides an API endpoint to handle deposit transactions.\n"
"It supports the creation of a deposit transaction after validating the "
"provided data. If the user is not authenticated, an appropriate response "
"is returned. On successful validation and execution, a response "
"with the transaction details is provided."
)
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
logger.debug(request.__dict__)
@ -52,33 +45,25 @@ class DepositView(APIView):
@extend_schema(exclude=True)
class CallbackAPIView(APIView):
"""
Handles incoming callback requests to the API.
This class processes and routes incoming HTTP POST requests to the appropriate
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
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.
"""
__doc__ = _( # type: ignore [assignment]
"Handles incoming callback requests to the API.\n"
"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 "
"callback events coming from external systems and provide an appropriate HTTP response "
"indicating success or failure."
)
def post(self, request: Request, *args: list[Any], **kwargs: dict[Any, Any]) -> Response:
try:
transaction = Transaction.objects.get(uuid=str(kwargs.get("uuid")))
if not transaction.gateway:
raise UnknownGatewayError()
gateway = transaction.gateway.get_integration_class_object(raise_exc=True)
gateway.process_callback(request.data)
raise UnknownGatewayError(_(f"Transaction {transaction.uuid} has no gateway"))
gateway_integration = transaction.gateway.get_integration_class_object(raise_exc=True)
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)
except Exception as e:
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()}
)

2
engine/vibes_auth/messaging/auth.py
View file

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

6
engine/vibes_auth/messaging/consumers.py
View file

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

9
engine/vibes_auth/messaging/forwarders/telegram.py
View file

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

2
engine/vibes_auth/messaging/services.py
View file

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

2
engine/vibes_auth/tests/test_drf.py
View file

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

12
engine/vibes_auth/views.py
View file

@ -36,10 +36,10 @@ class TokenObtainPairView(TokenViewBase):
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,
]
authentication_classes: list[str] = []
authentication_classes: list[str] = [] # type: ignore [assignment]
@method_decorator(ratelimit(key="ip", rate="10/h"))
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]
permission_classes: list[Type[BasePermission]] = [
permission_classes: list[Type[BasePermission]] = [ # type: ignore [assignment]
AllowAny,
]
authentication_classes: list[str] = []
authentication_classes: list[str] = [] # type: ignore [assignment]
@method_decorator(ratelimit(key="ip", rate="10/h"))
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]
permission_classes: list[Type[BasePermission]] = [
permission_classes: list[Type[BasePermission]] = [ # type: ignore [assignment]
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:
try:

1
evibes/middleware.py
View file

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