import logging import uuid from typing import Type from uuid import UUID from constance import config from django.conf import settings from django.db.models import Prefetch, Q, OuterRef, Exists 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 djangorestframework_camel_case.render import CamelCaseJSONRenderer 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.renderers import MultiPartRenderer 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 rest_framework_xml.renderers import XMLRenderer from rest_framework_yaml.renderers import YAMLRenderer from core.docs.drf.viewsets import ( ADDRESS_SCHEMA, ATTRIBUTE_GROUP_SCHEMA, ATTRIBUTE_SCHEMA, ATTRIBUTE_VALUE_SCHEMA, CATEGORY_SCHEMA, FEEDBACK_SCHEMA, ORDER_PRODUCT_SCHEMA, ORDER_SCHEMA, PRODUCT_SCHEMA, WISHLIST_SCHEMA, ) from core.filters import AddressFilter, BrandFilter, CategoryFilter, FeedbackFilter, OrderFilter, ProductFilter from core.models import ( Address, Attribute, AttributeGroup, AttributeValue, Brand, Category, Feedback, Order, OrderProduct, Product, ProductImage, ProductTag, PromoCode, Promotion, Stock, Vendor, Wishlist, ) from core.permissions import EvibesPermission from 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 core.serializers.seo import SeoSnapshotSerializer from core.utils import format_attributes from core.utils.messages import permission_denied_message from core.utils.nominatim import fetch_address_suggestions from core.utils.seo_builders import ( brand_schema, breadcrumb_schema, category_schema, item_list_schema, org_schema, product_schema, website_schema, ) from payments.serializers import TransactionProcessSerializer logger = logging.getLogger("django") class EvibesViewSet(ModelViewSet): __doc__ = _( # type: ignore "Defines a viewset for managing Evibes-related operations. " "The EvibesViewSet class inherits from ModelViewSet and provides functionality " "for handling actions and operations on Evibes 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 = [EvibesPermission] renderer_classes = [CamelCaseJSONRenderer, MultiPartRenderer, XMLRenderer, YAMLRenderer] def get_serializer_class(self) -> Type[Serializer]: # noinspection PyTypeChecker return self.action_serializer_classes.get(self.action, super().get_serializer_class()) # type: ignore [arg-type] @extend_schema_view(**ATTRIBUTE_GROUP_SCHEMA) class AttributeGroupViewSet(EvibesViewSet): __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(EvibesViewSet): __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(EvibesViewSet): __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(EvibesViewSet): __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", "attributes", "tags") filter_backends = [DjangoFilterBackend] filterset_class = CategoryFilter serializer_class = CategoryDetailSerializer action_serializer_classes = { "list": CategorySimpleSerializer, } lookup_field = "lookup_value" 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"): return qs return qs.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: category = self.get_object() title = f"{category.name} | {config.PROJECT_NAME}" description = (category.description or "")[:180] canonical = f"https://{config.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://{config.BASE_DOMAIN}/")] if category.get_ancestors().exists(): for c in category.get_ancestors(): crumbs.append((c.name, f"https://{config.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://{config.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) class BrandViewSet(EvibesViewSet): __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 = "lookup_value" 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"): 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} | {config.PROJECT_NAME}" description = (brand.description or "")[:180] canonical = f"https://{config.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://{config.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(EvibesViewSet): __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 `EvibesViewSet` 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"): 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") 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"): 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, ], ) 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} | {config.PROJECT_NAME}" description = (p.description or "")[:180] canonical = f"https://{config.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://{config.BASE_DOMAIN}/")] if p.category: for c in p.category.get_ancestors(include_self=True): crumbs.append((c.name, f"https://{config.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) class VendorViewSet(EvibesViewSet): __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(EvibesViewSet): __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 `EvibesViewSet` " "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"): return qs return qs.filter(is_active=True) # noinspection PyUnusedLocal @extend_schema_view(**ORDER_SCHEMA) class OrderViewSet(EvibesViewSet): __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 = "lookup_value" 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"): return qs return qs.filter(user=user) def get_object(self): lookup_val = self.kwargs[self.lookup_field] 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") 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") 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 "": return Response(status=status.HTTP_202_ACCEPTED, data=TransactionProcessSerializer(instance).data) case "": 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") 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") 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") 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") 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") 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") 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") 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") 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(EvibesViewSet): __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"): 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") 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) class ProductImageViewSet(EvibesViewSet): __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, } class PromoCodeViewSet(EvibesViewSet): __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"): return qs return qs.filter(user=user) class PromotionViewSet(EvibesViewSet): __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, } class StockViewSet(EvibesViewSet): __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(EvibesViewSet): __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"): 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") 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") 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") 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") 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(EvibesViewSet): __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"): 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, ) class ProductTagViewSet(EvibesViewSet): __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, }