Egor fureunoir Gorbunov
eef774c3a3
Replace WYSIWYG editor with Markdown editor across all relevant models and admin fields. Add utilities for rendering and stripping markdown. Adjust serializers, views, and templates to support markdown content. Introduce `PastedImage` model and upload endpoint for handling inline image uploads in markdown. This change simplifies content formatting while enhancing flexibility with markdown support.
1277 lines
47 KiB
Python
1277 lines
47 KiB
Python
import logging
|
|
import uuid
|
|
from typing import Any, Type
|
|
from uuid import UUID
|
|
|
|
from django.conf import settings
|
|
from django.db.models import Exists, OuterRef, Prefetch, Q
|
|
from django.http import Http404
|
|
from django.shortcuts import get_object_or_404
|
|
from django.utils.decorators import method_decorator
|
|
from django.utils.translation import gettext_lazy as _
|
|
from django_filters.rest_framework import DjangoFilterBackend
|
|
from django_ratelimit.decorators import ratelimit
|
|
from drf_spectacular.utils import extend_schema_view
|
|
from rest_framework import status
|
|
from rest_framework.decorators import action
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.permissions import AllowAny
|
|
from rest_framework.request import Request
|
|
from rest_framework.response import Response
|
|
from rest_framework.serializers import Serializer
|
|
from rest_framework.viewsets import ModelViewSet
|
|
|
|
from engine.core.docs.drf.viewsets import (
|
|
ADDRESS_SCHEMA,
|
|
ATTRIBUTE_GROUP_SCHEMA,
|
|
ATTRIBUTE_SCHEMA,
|
|
ATTRIBUTE_VALUE_SCHEMA,
|
|
BRAND_SCHEMA,
|
|
CATEGORY_SCHEMA,
|
|
FEEDBACK_SCHEMA,
|
|
ORDER_PRODUCT_SCHEMA,
|
|
ORDER_SCHEMA,
|
|
PRODUCT_IMAGE_SCHEMA,
|
|
PRODUCT_SCHEMA,
|
|
PRODUCT_TAG_SCHEMA,
|
|
PROMOCODE_SCHEMA,
|
|
PROMOTION_SCHEMA,
|
|
STOCK_SCHEMA,
|
|
VENDOR_SCHEMA,
|
|
WISHLIST_SCHEMA,
|
|
)
|
|
from engine.core.filters import (
|
|
AddressFilter,
|
|
BrandFilter,
|
|
CategoryFilter,
|
|
FeedbackFilter,
|
|
OrderFilter,
|
|
ProductFilter,
|
|
)
|
|
from engine.core.models import (
|
|
Address,
|
|
Attribute,
|
|
AttributeGroup,
|
|
AttributeValue,
|
|
Brand,
|
|
Category,
|
|
Feedback,
|
|
Order,
|
|
OrderProduct,
|
|
Product,
|
|
ProductImage,
|
|
ProductTag,
|
|
PromoCode,
|
|
Promotion,
|
|
Stock,
|
|
Vendor,
|
|
Wishlist,
|
|
)
|
|
from engine.core.permissions import SchonPermission
|
|
from engine.core.serializers import (
|
|
AddOrderProductSerializer,
|
|
AddressAutocompleteInputSerializer,
|
|
AddressCreateSerializer,
|
|
AddressSerializer,
|
|
AddressSuggestionSerializer,
|
|
AddWishlistProductSerializer,
|
|
AttributeDetailSerializer,
|
|
AttributeGroupDetailSerializer,
|
|
AttributeGroupSimpleSerializer,
|
|
AttributeSimpleSerializer,
|
|
AttributeValueDetailSerializer,
|
|
AttributeValueSimpleSerializer,
|
|
BrandDetailSerializer,
|
|
BrandSimpleSerializer,
|
|
BulkAddOrderProductsSerializer,
|
|
BulkAddWishlistProductSerializer,
|
|
BulkRemoveOrderProductsSerializer,
|
|
BulkRemoveWishlistProductSerializer,
|
|
BuyOrderSerializer,
|
|
BuyUnregisteredOrderSerializer,
|
|
CategoryDetailSerializer,
|
|
CategorySimpleSerializer,
|
|
DoFeedbackSerializer,
|
|
FeedbackDetailSerializer,
|
|
FeedbackSimpleSerializer,
|
|
OrderDetailSerializer,
|
|
OrderProductSimpleSerializer,
|
|
OrderSimpleSerializer,
|
|
ProductDetailSerializer,
|
|
ProductImageDetailSerializer,
|
|
ProductImageSimpleSerializer,
|
|
ProductSimpleSerializer,
|
|
ProductTagDetailSerializer,
|
|
ProductTagSimpleSerializer,
|
|
PromoCodeDetailSerializer,
|
|
PromoCodeSimpleSerializer,
|
|
PromotionDetailSerializer,
|
|
PromotionSimpleSerializer,
|
|
RemoveOrderProductSerializer,
|
|
RemoveWishlistProductSerializer,
|
|
StockDetailSerializer,
|
|
StockSimpleSerializer,
|
|
VendorDetailSerializer,
|
|
VendorSimpleSerializer,
|
|
WishlistDetailSerializer,
|
|
WishlistSimpleSerializer,
|
|
)
|
|
from engine.core.serializers.seo import SeoSnapshotSerializer
|
|
from engine.core.utils import format_attributes
|
|
from engine.core.utils.messages import permission_denied_message
|
|
from engine.core.utils.nominatim import fetch_address_suggestions
|
|
from engine.core.utils.seo_builders import (
|
|
brand_schema,
|
|
breadcrumb_schema,
|
|
category_schema,
|
|
item_list_schema,
|
|
org_schema,
|
|
product_schema,
|
|
website_schema,
|
|
)
|
|
from engine.payments.serializers import TransactionProcessSerializer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SchonViewSet(ModelViewSet):
|
|
__doc__ = _(
|
|
"Defines a viewset for managing Schon-related operations. "
|
|
"The SchonViewSet class inherits from ModelViewSet and provides functionality "
|
|
"for handling actions and operations on Schon entities. It includes support "
|
|
"for dynamic serializer classes based on the current action, customizable "
|
|
"permissions, and rendering formats."
|
|
)
|
|
|
|
action_serializer_classes: dict[str, Type[Serializer]] = {}
|
|
additional: dict[str, str] = {}
|
|
permission_classes = [SchonPermission]
|
|
|
|
def get_serializer_class(self) -> Type[Any]:
|
|
# noinspection PyTypeChecker
|
|
return self.action_serializer_classes.get(
|
|
self.action, super().get_serializer_class()
|
|
)
|
|
|
|
|
|
@extend_schema_view(**ATTRIBUTE_GROUP_SCHEMA)
|
|
class AttributeGroupViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Represents a viewset for managing AttributeGroup objects. "
|
|
"Handles operations related to AttributeGroup, including filtering, "
|
|
"serialization, and retrieval of data. This class is part of the "
|
|
"application's API layer and provides a standardized way to process "
|
|
"requests and responses for AttributeGroup data."
|
|
)
|
|
|
|
queryset = AttributeGroup.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["is_active"]
|
|
serializer_class = AttributeGroupDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": AttributeGroupSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**ATTRIBUTE_SCHEMA)
|
|
class AttributeViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Handles operations related to Attribute objects within the application. "
|
|
"Provides a set of API endpoints to interact with Attribute data. This class "
|
|
"manages querying, filtering, and serialization of Attribute objects, allowing "
|
|
"dynamic control over the data returned, such as filtering by specific fields "
|
|
"or retrieving detailed versus simplified information depending on the request."
|
|
)
|
|
|
|
queryset = Attribute.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["group", "value_type", "is_active"]
|
|
serializer_class = AttributeDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": AttributeSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**ATTRIBUTE_VALUE_SCHEMA)
|
|
class AttributeValueViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"A viewset for managing AttributeValue objects. "
|
|
"This viewset provides functionality for listing, retrieving, creating, updating, and deleting "
|
|
"AttributeValue objects. It integrates with Django REST Framework's viewset mechanisms and uses "
|
|
"appropriate serializers for different actions. Filtering capabilities are provided through the "
|
|
"DjangoFilterBackend."
|
|
)
|
|
|
|
queryset = AttributeValue.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["attribute", "is_active"]
|
|
serializer_class = AttributeValueDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": AttributeValueSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**CATEGORY_SCHEMA)
|
|
class CategoryViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Manages views for Category-related operations. "
|
|
"The CategoryViewSet class is responsible for handling operations related to "
|
|
"the Category model in the system. It supports retrieving, filtering, and "
|
|
"serializing category data. The viewset also enforces permissions to ensure "
|
|
"that only authorized users can access specific data."
|
|
)
|
|
|
|
queryset = Category.objects.all().prefetch_related("parent", "children", "tags")
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = CategoryFilter
|
|
serializer_class = CategoryDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": CategorySimpleSerializer,
|
|
}
|
|
lookup_field = "uuid"
|
|
lookup_url_kwarg = "lookup_value"
|
|
additional = {"seo_meta": "ALLOW"}
|
|
|
|
def get_object(self):
|
|
queryset = self.filter_queryset(self.get_queryset())
|
|
lookup_value = self.kwargs[self.lookup_url_kwarg]
|
|
|
|
obj = None
|
|
|
|
try:
|
|
uuid_obj = UUID(lookup_value)
|
|
obj = queryset.filter(uuid=uuid_obj).first()
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if not obj:
|
|
obj = queryset.filter(slug=lookup_value).first()
|
|
|
|
if not obj:
|
|
name = "Category"
|
|
raise Http404(f"{name} does not exist: {lookup_value}")
|
|
|
|
self.check_object_permissions(self.request, obj)
|
|
return obj
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
if self.request.user.has_perm("core.view_category"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
return qs.filter(is_active=True)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(
|
|
detail=True,
|
|
methods=("GET",),
|
|
url_path="meta",
|
|
permission_classes=[
|
|
AllowAny,
|
|
],
|
|
)
|
|
@method_decorator(ratelimit(key="ip", rate="4/s" if not settings.DEBUG else "44/s"))
|
|
def seo_meta(self, request: Request, *args, **kwargs) -> Response:
|
|
category = self.get_object()
|
|
|
|
title = f"{category.name} | {settings.PROJECT_NAME}"
|
|
description = category.seo_description
|
|
canonical = f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/catalog/{category.slug}"
|
|
og_image = (
|
|
request.build_absolute_uri(category.image.url)
|
|
if getattr(category, "image", None)
|
|
else ""
|
|
)
|
|
|
|
og = {
|
|
"title": title,
|
|
"description": description,
|
|
"type": "website",
|
|
"url": canonical,
|
|
"image": og_image,
|
|
}
|
|
tw = {"card": "summary_large_image", "title": title, "description": description}
|
|
|
|
crumbs = [("Home", f"https://{settings.BASE_DOMAIN}/")]
|
|
if category.get_ancestors().exists():
|
|
for c in category.get_ancestors():
|
|
crumbs.append(
|
|
(
|
|
c.name,
|
|
f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/catalog/{c.slug}",
|
|
)
|
|
)
|
|
crumbs.append((category.name, canonical))
|
|
|
|
json_ld = [
|
|
org_schema(),
|
|
website_schema(),
|
|
breadcrumb_schema(crumbs),
|
|
category_schema(category, canonical),
|
|
]
|
|
|
|
product_urls = []
|
|
qs = (
|
|
Product.objects.filter(
|
|
is_active=True,
|
|
category=category,
|
|
brand__is_active=True,
|
|
stocks__vendor__is_active=True,
|
|
)
|
|
.only("slug")
|
|
.distinct()[:24]
|
|
)
|
|
for p in qs:
|
|
product_urls.append(
|
|
f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/product/{p.slug}"
|
|
)
|
|
if product_urls:
|
|
json_ld.append(item_list_schema(product_urls))
|
|
|
|
payload = {
|
|
"title": title,
|
|
"description": description,
|
|
"canonical": canonical,
|
|
"robots": "index,follow",
|
|
"hreflang": request.LANGUAGE_CODE,
|
|
"open_graph": og,
|
|
"twitter": tw,
|
|
"json_ld": json_ld,
|
|
}
|
|
return Response(SeoSnapshotSerializer(payload).data)
|
|
|
|
|
|
@extend_schema_view(**BRAND_SCHEMA)
|
|
class BrandViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Represents a viewset for managing Brand instances. "
|
|
"This class provides functionality for querying, filtering, and "
|
|
"serializing Brand objects. It uses Django's ViewSet framework "
|
|
"to simplify the implementation of API endpoints for Brand objects."
|
|
)
|
|
|
|
queryset = Brand.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = BrandFilter
|
|
serializer_class = BrandDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": BrandSimpleSerializer,
|
|
}
|
|
lookup_field = "uuid"
|
|
lookup_url_kwarg = "lookup_value"
|
|
additional = {"seo_meta": "ALLOW"}
|
|
|
|
def get_object(self):
|
|
queryset = self.filter_queryset(self.get_queryset())
|
|
lookup_value = self.kwargs[self.lookup_url_kwarg]
|
|
|
|
obj = None
|
|
|
|
try:
|
|
uuid_obj = UUID(lookup_value)
|
|
obj = queryset.filter(uuid=uuid_obj).first()
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if not obj:
|
|
obj = queryset.filter(slug=lookup_value).first()
|
|
|
|
if not obj:
|
|
name = "Brand"
|
|
raise Http404(f"{name} does not exist: {lookup_value}")
|
|
|
|
self.check_object_permissions(self.request, obj)
|
|
return obj
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
if self.request.user.has_perm("core.view_brand"): # ty:ignore[possibly-missing-attribute]
|
|
if self.request.user.has_perm("core.view_brand"): # ty:ignore[possibly-missing-attribute]
|
|
return qs.prefetch_related("categories")
|
|
return qs.prefetch_related(
|
|
Prefetch("categories", queryset=Category.objects.filter(is_active=True))
|
|
)
|
|
if self.request.user.has_perm("core.view_category"): # ty:ignore[possibly-missing-attribute]
|
|
return qs.filter(is_active=True).prefetch_related("categories")
|
|
return qs.filter(is_active=True).prefetch_related(
|
|
Prefetch("categories", queryset=Category.objects.filter(is_active=True))
|
|
)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(
|
|
detail=True,
|
|
methods=("GET",),
|
|
url_path="meta",
|
|
permission_classes=[
|
|
AllowAny,
|
|
],
|
|
)
|
|
def seo_meta(self, request: Request, *args, **kwargs) -> Response:
|
|
brand = self.get_object()
|
|
|
|
title = f"{brand.name} | {settings.PROJECT_NAME}"
|
|
description = brand.seo_description
|
|
canonical = f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/brand/{brand.slug}"
|
|
|
|
logo_url = (
|
|
request.build_absolute_uri(brand.big_logo.url)
|
|
if getattr(brand, "big_logo", None)
|
|
else request.build_absolute_uri(brand.small_logo.url)
|
|
if getattr(brand, "small_logo", None)
|
|
else ""
|
|
)
|
|
|
|
og = {
|
|
"title": title,
|
|
"description": description,
|
|
"type": "website",
|
|
"url": canonical,
|
|
"image": logo_url,
|
|
}
|
|
tw = {"card": "summary_large_image", "title": title, "description": description}
|
|
|
|
crumbs = [
|
|
("Home", f"https://{settings.BASE_DOMAIN}/"),
|
|
(brand.name, canonical),
|
|
]
|
|
|
|
json_ld = [
|
|
org_schema(),
|
|
website_schema(),
|
|
breadcrumb_schema(crumbs),
|
|
brand_schema(brand, canonical, logo_url=logo_url),
|
|
]
|
|
|
|
payload = {
|
|
"title": title,
|
|
"description": description,
|
|
"canonical": canonical,
|
|
"robots": "index,follow",
|
|
"hreflang": request.LANGUAGE_CODE,
|
|
"open_graph": og,
|
|
"twitter": tw,
|
|
"json_ld": json_ld,
|
|
}
|
|
return Response(SeoSnapshotSerializer(payload).data)
|
|
|
|
|
|
@extend_schema_view(**PRODUCT_SCHEMA)
|
|
class ProductViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Manages operations related to the `Product` model in the system. "
|
|
"This class provides a viewset for managing products, including their filtering, serialization, "
|
|
"and operations on specific instances. It extends from `SchonViewSet` to use common "
|
|
"functionality and integrates with the Django REST framework for RESTful API operations. "
|
|
"Includes methods for retrieving product details, applying permissions, and accessing "
|
|
"related feedback of a product."
|
|
)
|
|
|
|
queryset = Product.objects.prefetch_related(
|
|
"tags", "attributes", "stocks", "images"
|
|
).all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = ProductFilter
|
|
serializer_class = ProductDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": ProductSimpleSerializer,
|
|
"feedbacks": FeedbackSimpleSerializer,
|
|
}
|
|
lookup_field = "lookup_value"
|
|
lookup_url_kwarg = "lookup_value"
|
|
additional = {
|
|
"seo_meta": "ALLOW",
|
|
"feedbacks": "ALLOW",
|
|
}
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
|
|
qs = qs.select_related("brand", "category")
|
|
|
|
if self.request.user.has_perm("core.view_product"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
|
|
active_stocks = Stock.objects.filter(
|
|
product_id=OuterRef("pk"), vendor__is_active=True
|
|
)
|
|
|
|
return (
|
|
qs.filter(
|
|
is_active=True,
|
|
brand__is_active=True,
|
|
category__is_active=True,
|
|
)
|
|
.annotate(has_active_stocks=Exists(active_stocks))
|
|
.filter(has_active_stocks=True)
|
|
.distinct()
|
|
)
|
|
|
|
def get_object(self):
|
|
queryset = self.filter_queryset(self.get_queryset())
|
|
lookup_value = self.kwargs[self.lookup_url_kwarg]
|
|
|
|
obj = None
|
|
|
|
try:
|
|
uuid_obj = UUID(lookup_value)
|
|
obj = queryset.filter(uuid=uuid_obj).first()
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if not obj:
|
|
obj = queryset.filter(slug=lookup_value).first()
|
|
|
|
if not obj:
|
|
obj = queryset.filter(sku=lookup_value).first()
|
|
|
|
if not obj:
|
|
name = "Product"
|
|
raise Http404(f"{name} does not exist: {lookup_value}")
|
|
|
|
self.check_object_permissions(self.request, obj)
|
|
return obj
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=True, methods=("GET",), url_path="feedbacks")
|
|
@method_decorator(ratelimit(key="ip", rate="2/s" if not settings.DEBUG else "44/s"))
|
|
def feedbacks(self, request: Request, *args, **kwargs) -> Response:
|
|
product = self.get_object()
|
|
qs = Feedback.objects.filter(order_product__product=product)
|
|
if not request.user.has_perm("core.view_feedback"): # ty:ignore[possibly-missing-attribute]
|
|
qs = qs.filter(is_active=True)
|
|
return Response(data=FeedbackSimpleSerializer(qs, many=True).data)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(
|
|
detail=True,
|
|
methods=("GET",),
|
|
url_path="meta",
|
|
permission_classes=[
|
|
AllowAny,
|
|
],
|
|
)
|
|
@method_decorator(ratelimit(key="ip", rate="4/s" if not settings.DEBUG else "44/s"))
|
|
def seo_meta(self, request: Request, *args, **kwargs) -> Response:
|
|
p = self.get_object()
|
|
images = list(p.images.all()[:6])
|
|
rating = {"value": p.rating, "count": p.feedbacks_count}
|
|
title = f"{p.name} | {settings.PROJECT_NAME}"
|
|
description = p.seo_description
|
|
canonical = (
|
|
f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/product/{p.slug}"
|
|
)
|
|
og = {
|
|
"title": title,
|
|
"description": description,
|
|
"type": "product",
|
|
"url": canonical,
|
|
"image": request.build_absolute_uri(images[0].image.url) if images else "",
|
|
}
|
|
tw = {"card": "summary_large_image", "title": title, "description": description}
|
|
|
|
crumbs = [("Home", f"https://{settings.BASE_DOMAIN}/")]
|
|
if p.category:
|
|
for c in p.category.get_ancestors(include_self=True):
|
|
crumbs.append(
|
|
(
|
|
c.name,
|
|
f"https://{settings.BASE_DOMAIN}/{settings.LANGUAGE_CODE}/catalog/{c.slug}",
|
|
)
|
|
)
|
|
crumbs.append((p.name, canonical))
|
|
|
|
json_ld = [org_schema(), website_schema()]
|
|
if crumbs:
|
|
json_ld.append(breadcrumb_schema(crumbs))
|
|
json_ld.append(product_schema(p, images, rating=rating))
|
|
|
|
payload = {
|
|
"title": title,
|
|
"description": description,
|
|
"canonical": canonical,
|
|
"robots": "index,follow",
|
|
"hreflang": request.LANGUAGE_CODE,
|
|
"open_graph": og,
|
|
"twitter": tw,
|
|
"json_ld": json_ld,
|
|
}
|
|
return Response(SeoSnapshotSerializer(payload).data)
|
|
|
|
@method_decorator(ratelimit(key="ip", rate="4/s" if not settings.DEBUG else "44/s"))
|
|
def retrieve(self, request, *args, **kwargs):
|
|
return super().retrieve(request, *args, **kwargs)
|
|
|
|
|
|
@extend_schema_view(**VENDOR_SCHEMA)
|
|
class VendorViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Represents a viewset for managing Vendor objects. "
|
|
"This viewset allows fetching, filtering, and serializing Vendor data. "
|
|
"It defines the queryset, filter configurations, and serializer classes "
|
|
"used to handle different actions. The purpose of this class is to "
|
|
"provide streamlined access to Vendor-related resources through the "
|
|
"Django REST framework."
|
|
)
|
|
|
|
queryset = Vendor.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["name", "markup_percent", "is_active"]
|
|
serializer_class = VendorDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": VendorSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**FEEDBACK_SCHEMA)
|
|
class FeedbackViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Representation of a view set handling Feedback objects. "
|
|
"This class manages operations related to Feedback objects, including listing, "
|
|
"filtering, and retrieving details. The purpose of this view set is to provide "
|
|
"different serializers for different actions and implement permission-based "
|
|
"handling of accessible Feedback objects. It extends the base `SchonViewSet` "
|
|
"and makes use of Django's filtering system for querying data."
|
|
)
|
|
|
|
queryset = Feedback.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = FeedbackFilter
|
|
serializer_class = FeedbackDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": FeedbackSimpleSerializer,
|
|
}
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
if self.request.user.has_perm("core.view_feedback"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
return qs.filter(is_active=True)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
@extend_schema_view(**ORDER_SCHEMA)
|
|
class OrderViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"ViewSet for managing orders and related operations. "
|
|
"This class provides functionality to retrieve, modify, and manage order objects. "
|
|
"It includes various endpoints for handling order operations such as adding or "
|
|
"removing products, performing purchases for registered as well as unregistered "
|
|
"users, and retrieving the current authenticated user's pending orders. "
|
|
"The ViewSet uses multiple serializers based on the specific action being "
|
|
"performed and enforces permissions accordingly while interacting with order data."
|
|
)
|
|
|
|
lookup_field = "uuid"
|
|
lookup_url_kwarg = "lookup_value"
|
|
queryset = Order.objects.prefetch_related("order_products").all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = OrderFilter
|
|
serializer_class = OrderDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": OrderSimpleSerializer,
|
|
"buy": OrderDetailSerializer,
|
|
"add_order_product": AddOrderProductSerializer,
|
|
"remove_order_product": RemoveOrderProductSerializer,
|
|
}
|
|
additional = {"retrieve": "ALLOW"}
|
|
|
|
def get_serializer_class(self):
|
|
return self.action_serializer_classes.get(
|
|
self.action, super().get_serializer_class()
|
|
)
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
user = self.request.user
|
|
|
|
if not user.is_authenticated:
|
|
return qs.filter(user__isnull=True)
|
|
|
|
if user.has_perm("core.view_order"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
|
|
return qs.filter(user=user)
|
|
|
|
def get_object(self):
|
|
lookup_val = self.kwargs[self.lookup_url_kwarg]
|
|
qs = self.get_queryset()
|
|
|
|
try:
|
|
uuid.UUID(lookup_val)
|
|
uuid_q = Q(uuid=lookup_val)
|
|
except ValueError:
|
|
uuid_q = Q()
|
|
|
|
obj = get_object_or_404(qs, uuid_q | Q(human_readable_id=lookup_val))
|
|
self.check_object_permissions(self.request, obj)
|
|
return obj
|
|
|
|
@action(detail=False, methods=("GET",), url_path="current")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def current(self, request: Request, *args, **kwargs) -> Response:
|
|
if not request.user.is_authenticated:
|
|
raise PermissionDenied(permission_denied_message)
|
|
try:
|
|
order = Order.objects.get(user=request.user, status="PENDING")
|
|
except Order.DoesNotExist:
|
|
order = Order.objects.create(user=request.user)
|
|
return Response(
|
|
status=status.HTTP_200_OK,
|
|
data=OrderDetailSerializer(order).data,
|
|
)
|
|
|
|
@action(detail=True, methods=("POST",), url_path="buy")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def buy(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = BuyOrderSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
order = self.get_object()
|
|
if not request.user or request.user.is_anonymous:
|
|
return Response(
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
try:
|
|
instance = order.buy(
|
|
force_balance=serializer.validated_data.get("force_balance"),
|
|
force_payment=serializer.validated_data.get("force_payment"),
|
|
promocode_uuid=serializer.validated_data.get("promocode_uuid"),
|
|
shipping_address=serializer.validated_data.get("shipping_address_uuid"),
|
|
billing_address=serializer.validated_data.get("billing_address_uuid"),
|
|
chosen_products=serializer.validated_data.get("chosen_products"),
|
|
)
|
|
match str(type(instance)):
|
|
case "<class 'engine.payments.models.Transaction'>":
|
|
return Response(
|
|
status=status.HTTP_202_ACCEPTED,
|
|
data=TransactionProcessSerializer(instance).data,
|
|
)
|
|
case "<class 'engine.core.models.Order'>":
|
|
return Response(
|
|
status=status.HTTP_200_OK,
|
|
data=OrderDetailSerializer(instance).data,
|
|
)
|
|
case _:
|
|
raise TypeError(
|
|
_(
|
|
f"wrong type came from order.buy() method: {type(instance)!s}"
|
|
)
|
|
)
|
|
except Order.DoesNotExist:
|
|
name = "Order"
|
|
return Response(
|
|
status=status.HTTP_404_NOT_FOUND,
|
|
data={"detail": _(f"{name} does not exist: {uuid}")},
|
|
)
|
|
except Exception as e:
|
|
return Response(status=status.HTTP_400_BAD_REQUEST, data={"detail": str(e)})
|
|
|
|
@action(detail=False, methods=("POST",), url_path="buy_unregistered")
|
|
@method_decorator(
|
|
ratelimit(key="ip", rate="10/h" if not settings.DEBUG else "888/h")
|
|
)
|
|
def buy_unregistered(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = BuyUnregisteredOrderSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
order = Order.objects.create(status="MOMENTAL")
|
|
products = [p["product_uuid"] for p in serializer.validated_data["products"]]
|
|
try:
|
|
transaction = order.buy_without_registration(
|
|
products=products,
|
|
promocode_uuid=serializer.validated_data.get("promocode_uuid"),
|
|
customer_name=serializer.validated_data.get("customer_name"),
|
|
customer_email=serializer.validated_data.get("customer_email"),
|
|
customer_phone_number=serializer.validated_data.get(
|
|
"customer_phone_number"
|
|
),
|
|
billing_customer_address=serializer.validated_data.get(
|
|
"billing_customer_address_uuid"
|
|
),
|
|
shipping_customer_address=serializer.validated_data.get(
|
|
"shipping_customer_address_uuid"
|
|
),
|
|
payment_method=serializer.validated_data.get("payment_method"),
|
|
)
|
|
return Response(
|
|
status=status.HTTP_201_CREATED,
|
|
data=TransactionProcessSerializer(transaction).data,
|
|
)
|
|
except Exception as e:
|
|
return Response(status=status.HTTP_400_BAD_REQUEST, data={"detail": str(e)})
|
|
|
|
@action(detail=True, methods=("POST",), url_path="add_order_product")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def add_order_product(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = AddOrderProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
order = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.add_orderproduct") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == order.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
order = order.add_product(
|
|
product_uuid=serializer.validated_data.get("product_uuid"),
|
|
attributes=format_attributes(
|
|
serializer.validated_data.get("attributes")
|
|
),
|
|
)
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data
|
|
)
|
|
except Order.DoesNotExist as dne:
|
|
return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": str(dne)})
|
|
except ValueError as ve:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST, data={"detail": str(ve)}
|
|
)
|
|
|
|
@action(detail=True, methods=("POST",), url_path="remove_order_product")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def remove_order_product(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = RemoveOrderProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
order = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.delete_orderproduct") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == order.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
order = order.remove_product(
|
|
product_uuid=serializer.validated_data.get("product_uuid"),
|
|
attributes=format_attributes(
|
|
serializer.validated_data.get("attributes")
|
|
),
|
|
)
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data
|
|
)
|
|
except Order.DoesNotExist as dne:
|
|
return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": str(dne)})
|
|
except ValueError as ve:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST, data={"detail": str(ve)}
|
|
)
|
|
|
|
@action(detail=True, methods=("POST",), url_path="bulk_add_order_products")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def bulk_add_order_products(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = BulkAddOrderProductsSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
lookup_val = kwargs.get(self.lookup_field)
|
|
try:
|
|
order = Order.objects.get(uuid=str(lookup_val))
|
|
if not (
|
|
request.user.has_perm("core.add_orderproduct") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == order.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
order = order.bulk_add_products(
|
|
products=serializer.validated_data.get("products"),
|
|
)
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data
|
|
)
|
|
except Order.DoesNotExist as dne:
|
|
return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": str(dne)})
|
|
except ValueError as ve:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST, data={"detail": str(ve)}
|
|
)
|
|
|
|
@action(detail=True, methods=("POST",), url_path="bulk_remove_order_products")
|
|
@method_decorator(ratelimit(key="ip", rate="1/s" if not settings.DEBUG else "44/s"))
|
|
def bulk_remove_order_products(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = BulkRemoveOrderProductsSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
order = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.delete_orderproduct") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == order.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
order = order.bulk_remove_products(
|
|
products=serializer.validated_data.get("products"),
|
|
)
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=OrderDetailSerializer(order).data
|
|
)
|
|
except Order.DoesNotExist as dne:
|
|
return Response(status=status.HTTP_404_NOT_FOUND, data={"detail": str(dne)})
|
|
except ValueError as ve:
|
|
return Response(
|
|
status=status.HTTP_400_BAD_REQUEST, data={"detail": str(ve)}
|
|
)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
@extend_schema_view(**ORDER_PRODUCT_SCHEMA)
|
|
class OrderProductViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Provides a viewset for managing OrderProduct entities. "
|
|
"This viewset enables CRUD operations and custom actions specific to the "
|
|
"OrderProduct model. It includes filtering, permission checks, and "
|
|
"serializer switching based on the requested action. Additionally, it "
|
|
"provides a detailed action for handling feedback on OrderProduct instances"
|
|
)
|
|
|
|
queryset = OrderProduct.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["order", "product", "status", "is_active"]
|
|
serializer_class = AttributeGroupDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": OrderProductSimpleSerializer,
|
|
"do_feedback": DoFeedbackSerializer,
|
|
}
|
|
additional = {"do_feedback": "ALLOW"}
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
user = self.request.user
|
|
|
|
if user.has_perm("core.view_orderproduct"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
|
|
return qs.filter(user=user)
|
|
|
|
@action(detail=True, methods=("POST",), url_path="do_feedback")
|
|
def do_feedback(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = self.get_serializer(request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
order_product = OrderProduct.objects.get(uuid=str(kwargs.get("pk")))
|
|
if not order_product.order:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
if not (
|
|
request.user.has_perm("core.change_orderproduct") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == order_product.order.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
feedback = order_product.do_feedback(
|
|
rating=serializer.validated_data.get("rating"),
|
|
comment=serializer.validated_data.get("comment"),
|
|
action=serializer.validated_data.get("action"),
|
|
)
|
|
match serializer.validated_data.get("action"):
|
|
case "add":
|
|
return Response(
|
|
data=FeedbackDetailSerializer(feedback).data,
|
|
status=status.HTTP_201_CREATED,
|
|
)
|
|
case "remove":
|
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
|
case _:
|
|
return Response(status=status.HTTP_400_BAD_REQUEST)
|
|
except OrderProduct.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
|
|
@extend_schema_view(**PRODUCT_IMAGE_SCHEMA)
|
|
class ProductImageViewSet(SchonViewSet):
|
|
__doc__ = _("Manages operations related to Product images in the application. ")
|
|
|
|
queryset = ProductImage.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["product", "priority", "is_active"]
|
|
serializer_class = ProductImageDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": ProductImageSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**PROMOCODE_SCHEMA)
|
|
class PromoCodeViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Manages the retrieval and handling of PromoCode instances through various API actions."
|
|
)
|
|
|
|
queryset = PromoCode.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = [
|
|
"code",
|
|
"discount_amount",
|
|
"discount_percent",
|
|
"start_time",
|
|
"end_time",
|
|
"used_on",
|
|
"is_active",
|
|
]
|
|
serializer_class = PromoCodeDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": PromoCodeSimpleSerializer,
|
|
}
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
user = self.request.user
|
|
|
|
if user.has_perm("core.view_promocode"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
|
|
return qs.filter(user=user)
|
|
|
|
|
|
@extend_schema_view(**PROMOTION_SCHEMA)
|
|
class PromotionViewSet(SchonViewSet):
|
|
__doc__ = _("Represents a view set for managing promotions. ")
|
|
|
|
queryset = Promotion.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["name", "discount_percent", "is_active"]
|
|
serializer_class = PromotionDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": PromotionSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**STOCK_SCHEMA)
|
|
class StockViewSet(SchonViewSet):
|
|
__doc__ = _("Handles operations related to Stock data in the system.")
|
|
|
|
queryset = Stock.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["vendor", "product", "sku", "is_active"]
|
|
serializer_class = StockDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": StockSimpleSerializer,
|
|
}
|
|
|
|
|
|
@extend_schema_view(**WISHLIST_SCHEMA)
|
|
class WishlistViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"ViewSet for managing Wishlist operations. "
|
|
"The WishlistViewSet provides endpoints for interacting with a user's wish list, "
|
|
"allowing for the retrieval, modification, and customization of products within "
|
|
"the wish list. This ViewSet facilitates functionality such as adding, removing, "
|
|
"and bulk actions for wishlist products. Permission checks are integrated to "
|
|
"ensure that users can only manage their own wishlists unless explicit permissions "
|
|
"are granted."
|
|
)
|
|
|
|
queryset = Wishlist.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["user", "is_active"]
|
|
serializer_class = WishlistDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": WishlistSimpleSerializer,
|
|
}
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
user = self.request.user
|
|
|
|
if user.has_perm("core.view_wishlist"): # ty:ignore[possibly-missing-attribute]
|
|
return qs
|
|
|
|
return qs.filter(user=user)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=False, methods=("GET",), url_path="current")
|
|
def current(self, request: Request, *args, **kwargs) -> Response:
|
|
if not request.user.is_authenticated:
|
|
raise PermissionDenied(permission_denied_message)
|
|
wishlist = Wishlist.objects.get(user=request.user)
|
|
if not request.user == wishlist.user:
|
|
raise PermissionDenied(permission_denied_message)
|
|
return Response(
|
|
status=status.HTTP_200_OK,
|
|
data=WishlistDetailSerializer(wishlist).data,
|
|
)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=True, methods=("POST",), url_path="add_wishlist_product")
|
|
def add_wishlist_product(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = AddWishlistProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
wishlist = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.change_wishlist") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == wishlist.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
wishlist = wishlist.add_product(
|
|
product_uuid=serializer.validated_data.get("product_uuid"),
|
|
)
|
|
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=WishlistDetailSerializer(wishlist).data
|
|
)
|
|
except Wishlist.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=True, methods=("POST",), url_path="remove_wishlist_product")
|
|
def remove_wishlist_product(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = RemoveWishlistProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
wishlist = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.change_wishlist") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == wishlist.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
wishlist = wishlist.remove_product(
|
|
product_uuid=serializer.validated_data.get("product_uuid"),
|
|
)
|
|
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=WishlistDetailSerializer(wishlist).data
|
|
)
|
|
except Wishlist.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=True, methods=("POST",), url_path="bulk_add_wishlist_product")
|
|
def bulk_add_wishlist_products(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = BulkAddWishlistProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
wishlist = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.change_wishlist") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == wishlist.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
wishlist = wishlist.bulk_add_products(
|
|
product_uuids=serializer.validated_data.get("product_uuids"),
|
|
)
|
|
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=WishlistDetailSerializer(wishlist).data
|
|
)
|
|
except Order.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=True, methods=("POST",), url_path="bulk_remove_wishlist_product")
|
|
def bulk_remove_wishlist_products(
|
|
self, request: Request, *args, **kwargs
|
|
) -> Response:
|
|
serializer = BulkRemoveWishlistProductSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
try:
|
|
wishlist = self.get_object()
|
|
if not (
|
|
request.user.has_perm("core.change_wishlist") # ty:ignore[possibly-missing-attribute]
|
|
or request.user == wishlist.user
|
|
):
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
wishlist = wishlist.bulk_remove_products(
|
|
product_uuids=serializer.validated_data.get("product_uuids"),
|
|
)
|
|
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=WishlistDetailSerializer(wishlist).data
|
|
)
|
|
except Order.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
|
|
@extend_schema_view(**ADDRESS_SCHEMA)
|
|
class AddressViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"This class provides viewset functionality for managing `Address` objects. "
|
|
"The AddressViewSet class enables CRUD operations, filtering, and custom actions "
|
|
"related to address entities. It includes specialized behaviors for different HTTP "
|
|
"methods, serializer overrides, and permission handling based on the request context."
|
|
)
|
|
|
|
pagination_class = None
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_class = AddressFilter
|
|
queryset = Address.objects.all()
|
|
serializer_class = AddressSerializer
|
|
additional = {"create": "ALLOW", "retrieve": "ALLOW"}
|
|
|
|
def get_serializer_class(self):
|
|
if self.action == "create":
|
|
return AddressCreateSerializer
|
|
if self.action == "autocomplete":
|
|
return AddressAutocompleteInputSerializer
|
|
return AddressSerializer
|
|
|
|
def get_queryset(self):
|
|
if self.request.user.has_perm("core.view_address"): # ty:ignore[possibly-missing-attribute]
|
|
return super().get_queryset()
|
|
|
|
if self.request.user.is_authenticated:
|
|
return super().get_queryset().filter(user=self.request.user)
|
|
|
|
return Address.objects.none()
|
|
|
|
def retrieve(self, request: Request, *args, **kwargs) -> Response:
|
|
try:
|
|
address = Address.objects.get(uuid=str(kwargs.get("pk")))
|
|
return Response(
|
|
status=status.HTTP_200_OK, data=self.get_serializer(address).data
|
|
)
|
|
except Address.DoesNotExist:
|
|
return Response(status=status.HTTP_404_NOT_FOUND)
|
|
|
|
def create(self, request: Request, *args, **kwargs) -> Response:
|
|
create_serializer = AddressCreateSerializer(
|
|
data=request.data, context={"request": request}
|
|
)
|
|
create_serializer.is_valid(raise_exception=True)
|
|
|
|
address_obj = create_serializer.create(create_serializer.validated_data)
|
|
|
|
output_serializer = AddressSerializer(address_obj, context={"request": request})
|
|
|
|
return Response(
|
|
status=status.HTTP_201_CREATED,
|
|
data=output_serializer.data,
|
|
)
|
|
|
|
# noinspection PyUnusedLocal
|
|
@action(detail=False, methods=("GET",), url_path="autocomplete")
|
|
def autocomplete(self, request: Request, *args, **kwargs) -> Response:
|
|
serializer = AddressAutocompleteInputSerializer(data=request.query_params)
|
|
serializer.is_valid(raise_exception=True)
|
|
|
|
q = serializer.validated_data["q"]
|
|
limit = serializer.validated_data["limit"]
|
|
|
|
try:
|
|
suggestions = fetch_address_suggestions(query=q, limit=limit)
|
|
suggestion_serializer = AddressSuggestionSerializer(suggestions, many=True)
|
|
return Response(
|
|
suggestion_serializer.data,
|
|
status=status.HTTP_200_OK,
|
|
)
|
|
except Exception as e:
|
|
return Response(
|
|
{"detail": _(f"Geocoding error: {e}")},
|
|
status=status.HTTP_502_BAD_GATEWAY,
|
|
)
|
|
|
|
|
|
@extend_schema_view(**PRODUCT_TAG_SCHEMA)
|
|
class ProductTagViewSet(SchonViewSet):
|
|
__doc__ = _(
|
|
"Handles operations related to Product Tags within the application. "
|
|
"This class provides functionality for retrieving, filtering, and serializing "
|
|
"Product Tag objects. It supports flexible filtering on specific attributes "
|
|
"using the specified filter backend and dynamically uses different serializers "
|
|
"based on the action being performed."
|
|
)
|
|
|
|
queryset = ProductTag.objects.all()
|
|
filter_backends = [DjangoFilterBackend]
|
|
filterset_fields = ["tag_name", "is_active"]
|
|
serializer_class = ProductTagDetailSerializer
|
|
action_serializer_classes = {
|
|
"list": ProductTagSimpleSerializer,
|
|
}
|