1275 lines
47 KiB
Python
1275 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]: # ty: ignore[invalid-return-type]
|
|
# 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.description or "")[:180]
|
|
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):
|
|
queryset = Brand.objects.all()
|
|
|
|
if self.request.user.has_perm("view_category"): # ty:ignore[possibly-missing-attribute]
|
|
queryset = queryset.prefetch_related("categories")
|
|
else:
|
|
queryset = queryset.prefetch_related(
|
|
Prefetch("categories", queryset=Category.objects.filter(is_active=True))
|
|
)
|
|
|
|
return queryset
|
|
|
|
# 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.description or "")[:180]
|
|
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.description or "")[:180]
|
|
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,
|
|
}
|