import logging from typing import Any import requests from django.core.cache import cache from django.core.exceptions import BadRequest, PermissionDenied from django.http import Http404 from django.utils.translation import gettext_lazy as _ from graphene import UUID, Boolean, Field, Int, List, String from graphene.types.generic import GenericScalar from graphene_django.utils import camelize from core.elasticsearch import process_query from core.graphene import BaseMutation from core.graphene.object_types import ( AddressType, BulkProductInput, OrderType, ProductType, SearchResultsType, WishlistType, FeedbackType, ) from core.models import Address, Category, Order, Product, Wishlist, OrderProduct from core.utils import format_attributes, is_url_safe from core.utils.caching import web_cache from core.utils.emailing import contact_us_email from core.utils.messages import permission_denied_message from core.utils.nominatim import fetch_address_suggestions from payments.graphene.object_types import TransactionType logger = logging.getLogger("django") # noinspection PyUnusedLocal class CacheOperator(BaseMutation): class Meta: description = _("cache I/O") class Arguments: key = String(required=True, description=_("key to look for in or set into the cache")) data = GenericScalar(required=False, description=_("data to store in cache")) timeout = Int( required=False, description=_("timeout in seconds to set the data for into the cache"), ) data = GenericScalar(description=_("cached data")) @staticmethod def mutate(parent, info, key, data=None, timeout=None) -> dict[Any, Any]: # type: ignore [override] return camelize(web_cache(info.context, key, data, timeout)) # noinspection PyUnusedLocal class RequestCursedURL(BaseMutation): class Meta: description = _("request a CORSed URL") class Arguments: url = String(required=True) data = GenericScalar(description=_("camelized JSON data from the requested URL")) @staticmethod def mutate(parent, info, url) -> dict[str, Any]: # type: ignore [override] if not is_url_safe(url): raise BadRequest(_("only URLs starting with http(s):// are allowed")) try: data = cache.get(url, None) if not data: response = requests.get(url, headers={"content-type": "application/json"}) response.raise_for_status() data = camelize(response.json()) cache.set(url, data, 86400) return {"data": data} except Exception as e: return {"data": {"error": str(e)}} # noinspection PyUnusedLocal,PyTypeChecker class AddOrderProduct(BaseMutation): class Meta: description = _("add a product to the order") class Arguments: product_uuid = UUID(required=True) order_uuid = UUID(required=True) attributes = String(required=False) order = Field(OrderType) @staticmethod def mutate(parent, info, product_uuid, order_uuid, attributes=None): # type: ignore [override] user = info.context.user try: order = Order.objects.get(uuid=order_uuid) if not (user.has_perm("core.add_orderproduct") or user == order.user): raise PermissionDenied(permission_denied_message) order = order.add_product(product_uuid=product_uuid, attributes=format_attributes(attributes)) return AddOrderProduct(order=order) except Order.DoesNotExist as dne: raise Http404(_(f"order {order_uuid} not found")) from dne # noinspection PyUnusedLocal class RemoveOrderProduct(BaseMutation): class Meta: description = _("remove a product from the order") class Arguments: product_uuid = UUID(required=True) order_uuid = UUID(required=True) attributes = String(required=False) order = Field(OrderType) @staticmethod def mutate(parent, info, product_uuid, order_uuid, attributes=None) -> AddOrderProduct | None: # type: ignore [override] user = info.context.user try: order = Order.objects.get(uuid=order_uuid) if not (user.has_perm("core.change_orderproduct") or user == order.user): raise PermissionDenied(permission_denied_message) order = order.remove_product(product_uuid=product_uuid, attributes=format_attributes(attributes)) return AddOrderProduct(order=order) except Order.DoesNotExist as dne: raise Http404(_(f"order {order_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class RemoveAllOrderProducts(BaseMutation): class Meta: description = _("remove all products from the order") class Arguments: order_uuid = UUID(required=True) order = Field(OrderType) @staticmethod def mutate(parent, info, order_uuid): # type: ignore [override] user = info.context.user order = Order.objects.get(uuid=order_uuid) if not (user.has_perm("core.delete_orderproduct") or user == order.user): raise PermissionDenied(permission_denied_message) order = order.remove_all_products() return RemoveAllOrderProducts(order=order) # noinspection PyUnusedLocal,PyTypeChecker class RemoveOrderProductsOfAKind(BaseMutation): class Meta: description = _("remove a product from the order") class Arguments: product_uuid = UUID(required=True) order_uuid = UUID(required=True) order = Field(OrderType) @staticmethod def mutate(parent, info, product_uuid, order_uuid): # type: ignore [override] user = info.context.user order = Order.objects.get(uuid=order_uuid) if not (user.has_perm("core.delete_orderproduct") or user == order.user): raise PermissionDenied(permission_denied_message) order = order.remove_products_of_a_kind(product_uuid=product_uuid) return RemoveOrderProductsOfAKind(order=order) # noinspection PyUnusedLocal,PyTypeChecker class BuyOrder(BaseMutation): class Meta: description = _("buy an order") class Arguments: order_uuid = String(required=False) order_hr_id = String(required=False) force_balance = Boolean(required=False) force_payment = Boolean(required=False) promocode_uuid = String(required=False) shipping_address = String(required=False) billing_address = String(required=False) chosen_products = List(BulkProductInput, required=False) order = Field(OrderType, required=False) transaction = Field(TransactionType, required=False) @staticmethod def mutate( parent, info, order_uuid=None, order_hr_id=None, force_balance=False, force_payment=False, promocode_uuid=None, shipping_address=None, billing_address=None, chosen_products=None, ): # type: ignore [override] if not any([order_uuid, order_hr_id]) or all([order_uuid, order_hr_id]): raise BadRequest(_("please provide either order_uuid or order_hr_id - mutually exclusive")) user = info.context.user try: order = None if order_uuid: order = Order.objects.get(user=user, uuid=order_uuid) elif order_hr_id: order = Order.objects.get(user=user, human_readable_id=order_hr_id) instance = order.buy( force_balance=force_balance, force_payment=force_payment, promocode_uuid=promocode_uuid, shipping_address=shipping_address, billing_address=billing_address, chosen_products=chosen_products, ) match str(type(instance)): case "": return BuyOrder(transaction=instance) case "": return BuyOrder(order=instance) case _: raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}")) except Order.DoesNotExist as dne: raise Http404(_(f"order {order_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class BulkOrderAction(BaseMutation): class Meta: description = _("perform an action on a list of products in the order") class Arguments: order_uuid = UUID(required=False) order_hr_id = String(required=False) action = String(required=True, description=_("remove/add")) products = List(BulkProductInput, required=True) order = Field(OrderType, required=False) @staticmethod def mutate( parent, info, action, products, order_uuid=None, order_hr_id=None, ): # type: ignore [override] if not any([order_uuid, order_hr_id]) or all([order_uuid, order_hr_id]): raise BadRequest(_("please provide either order_uuid or order_hr_id - mutually exclusive")) user = info.context.user try: order = None if order_uuid: order = Order.objects.get(user=user, uuid=order_uuid) elif order_hr_id: order = Order.objects.get(user=user, human_readable_id=order_hr_id) # noinspection PyUnreachableCode match action: case "add": order = order.bulk_add_products(products) case "remove": order = order.bulk_remove_products(products) case _: raise BadRequest(_("action must be either add or remove")) return BulkOrderAction(order=order) except Order.DoesNotExist as dne: raise Http404(_(f"order {order_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class BulkWishlistAction(BaseMutation): class Meta: description = _("perform an action on a list of products in the wishlist") class Arguments: wishlist_uuid = UUID(required=False) action = String(required=True, description="remove/add") products = List(BulkProductInput, required=True) wishlist = Field(WishlistType, required=False) @staticmethod def mutate( parent, info, action, products, wishlist_uuid=None, ): # type: ignore [override] if not wishlist_uuid: raise BadRequest(_("please provide wishlist_uuid value")) user = info.context.user try: wishlist = Wishlist.objects.get(user=user, uuid=wishlist_uuid) # noinspection PyUnreachableCode match action: case "add": wishlist = wishlist.bulk_add_products(products) case "remove": wishlist = wishlist.bulk_remove_products(products) case _: raise BadRequest(_("action must be either add or remove")) return BulkWishlistAction(wishlist=wishlist) except Wishlist.DoesNotExist as dne: raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne # noinspection PyUnusedLocal class BuyUnregisteredOrder(BaseMutation): class Meta: description = _("purchase an order without account creation") class Arguments: products = List(UUID, required=True) promocode_uuid = UUID(required=False) customer_name = String(required=True) customer_email = String(required=True) customer_phone = String(required=True) customer_billing_address = String(required=False) customer_shipping_address = String(required=False) payment_method = String(required=True) is_business = Boolean(required=False) transaction = Field(TransactionType, required=False) @staticmethod def mutate( parent, info, products, customer_name, customer_email, customer_phone, customer_billing_address, payment_method, customer_shipping_address=None, promocode_uuid=None, is_business=False, ): # type: ignore [override] order = Order.objects.create(status="MOMENTAL") transaction = order.buy_without_registration( products=products, promocode_uuid=promocode_uuid, customer_name=customer_name, customer_email=customer_email, customer_phone=customer_phone, billing_customer_address=customer_billing_address, shipping_customer_address=customer_shipping_address, payment_method=payment_method, is_business=is_business, ) # noinspection PyTypeChecker return BuyUnregisteredOrder(transaction=transaction) # noinspection PyUnusedLocal,PyTypeChecker class AddWishlistProduct(BaseMutation): class Meta: description = _("add a product to the wishlist") class Arguments: product_uuid = UUID(required=True) wishlist_uuid = UUID(required=True) wishlist = Field(WishlistType) @staticmethod def mutate(parent, info, product_uuid, wishlist_uuid): # type: ignore [override] user = info.context.user try: wishlist = Wishlist.objects.get(uuid=wishlist_uuid) if not (user.has_perm("core.change_wishlist") or user == wishlist.user): raise PermissionDenied(permission_denied_message) wishlist.add_product(product_uuid=product_uuid) return AddWishlistProduct(wishlist=wishlist) except Wishlist.DoesNotExist as dne: raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class RemoveWishlistProduct(BaseMutation): class Meta: description = _("remove a product from the wishlist") class Arguments: product_uuid = UUID(required=True) wishlist_uuid = UUID(required=True) wishlist = Field(WishlistType) @staticmethod def mutate(parent, info, product_uuid, wishlist_uuid): # type: ignore [override] user = info.context.user try: wishlist = Wishlist.objects.get(uuid=wishlist_uuid) if not (user.has_perm("core.change_wishlist") or user == wishlist.user): raise PermissionDenied(permission_denied_message) wishlist.remove_product(product_uuid=product_uuid) return RemoveWishlistProduct(wishlist=wishlist) except Wishlist.DoesNotExist as dne: raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class RemoveAllWishlistProducts(BaseMutation): class Meta: description = _("remove all products from the wishlist") class Arguments: wishlist_uuid = UUID(required=True) wishlist = Field(WishlistType) @staticmethod def mutate(parent, info, wishlist_uuid): # type: ignore [override] user = info.context.user try: wishlist = Wishlist.objects.get(uuid=wishlist_uuid) if not (user.has_perm("core.change_wishlist") or user == wishlist.user): raise PermissionDenied(permission_denied_message) for product in wishlist.products.all(): wishlist.remove_product(product_uuid=product.pk) return RemoveAllWishlistProducts(wishlist=wishlist) except Wishlist.DoesNotExist as dne: raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class BuyWishlist(BaseMutation): class Meta: description = _("buy all products from the wishlist") class Arguments: wishlist_uuid = UUID(required=True) force_balance = Boolean(required=False) force_payment = Boolean(required=False) order = Field(OrderType, required=False) transaction = Field(TransactionType, required=False) @staticmethod def mutate(parent, info, wishlist_uuid, force_balance=False, force_payment=False): # type: ignore [override] user = info.context.user try: wishlist = Wishlist.objects.get(uuid=wishlist_uuid) if not (user.has_perm("core.change_wishlist") or user == wishlist.user): raise PermissionDenied(permission_denied_message) order = Order.objects.create(user=user, status="MOMENTAL") for product in ( wishlist.products.all() if user.has_perm("core.change_wishlist") else wishlist.products.filter(is_active=True) ): order.add_product(product_uuid=product.pk) instance = order.buy(force_balance=force_balance, force_payment=force_payment) match str(type(instance)): case "": return BuyWishlist(transaction=instance) case "": return BuyWishlist(order=instance) case _: raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}")) except Wishlist.DoesNotExist as dne: raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class BuyProduct(BaseMutation): class Meta: description = _("buy a product") class Arguments: product_uuid = UUID(required=True) attributes = String( required=False, description=_("please send the attributes as the string formatted like attr1=value1,attr2=value2"), ) force_balance = Boolean(required=False) force_payment = Boolean(required=False) order = Field(OrderType, required=False) transaction = Field(TransactionType, required=False) @staticmethod def mutate( parent, info, product_uuid, attributes=None, force_balance=False, force_payment=False, ): # type: ignore [override] user = info.context.user order = Order.objects.create(user=user, status="MOMENTAL") order.add_product(product_uuid=product_uuid, attributes=format_attributes(attributes)) instance = order.buy(force_balance=force_balance, force_payment=force_payment) match str(type(instance)): case "": return BuyProduct(transaction=instance) case "": return BuyProduct(order=instance) case _: raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}")) # noinspection PyUnusedLocal,PyTypeChecker class FeedbackProductAction(BaseMutation): class Meta: description = _("add or delete a feedback for orderproduct") class Arguments: order_product_uuid = UUID(required=True) action = String(required=True, description="add/remove") comment = String(required=False) rating = Int(required=False) feedback = Field(FeedbackType, required=False) @staticmethod def mutate(parent, info, order_product_uuid, action, comment=None, rating=None): # type: ignore [override] user = info.context.user try: order_product = OrderProduct.objects.get(uuid=order_product_uuid) if user != order_product.order.user: raise PermissionDenied(permission_denied_message) feedback = None match action: case "add": feedback = order_product.do_feedback(comment=comment, rating=rating, action="add") case "remove": feedback = order_product.do_feedback(action="remove") case _: raise BadRequest(_("action must be either `add` or `remove`")) return FeedbackProductAction(feedback=feedback) except OrderProduct.DoesNotExist as dne: raise Http404(_(f"order product {order_product_uuid} not found")) from dne # noinspection PyUnusedLocal,PyTypeChecker class CreateProduct(BaseMutation): class Arguments: name = String(required=True) description = String() category_uuid = UUID(required=True) product = Field(ProductType) @staticmethod def mutate(parent, info, name, category_uuid, description=None): # type: ignore [override] if not info.context.user.has_perm("core.add_product"): raise PermissionDenied(permission_denied_message) category = Category.objects.get(uuid=category_uuid) product = Product.objects.create(name=name, description=description, category=category) return CreateProduct(product=product) # noinspection PyUnusedLocal,PyTypeChecker class UpdateProduct(BaseMutation): class Arguments: uuid = UUID(required=True) name = String() description = String() category_uuid = UUID() product = Field(ProductType) @staticmethod def mutate(parent, info, uuid, name=None, description=None, category_uuid=None): # type: ignore [override] user = info.context.user if not user.has_perm("core.change_product"): raise PermissionDenied(permission_denied_message) product = Product.objects.get(uuid=uuid) if name: product.name = name if description: product.description = description if category_uuid: product.category = Category.objects.get(uuid=category_uuid) product.save() return UpdateProduct(product=product) # noinspection PyUnusedLocal,PyTypeChecker class DeleteProduct(BaseMutation): class Arguments: uuid = UUID(required=True) ok = Boolean() @staticmethod def mutate(parent, info, uuid): # type: ignore [override] user = info.context.user if not user.has_perm("core.delete_product"): raise PermissionDenied(permission_denied_message) product = Product.objects.get(uuid=uuid) product.delete() return DeleteProduct(ok=True) # noinspection PyUnusedLocal,PyTypeChecker class CreateAddress(BaseMutation): class Arguments: raw_data = String(required=True, description=_("original address string provided by the user")) address = Field(AddressType) @staticmethod def mutate(parent, info, raw_data): # type: ignore [override] user = info.context.user if info.context.user.is_authenticated else None address = Address.objects.create(raw_data=raw_data, user=user) return CreateAddress(address=address) # noinspection PyUnusedLocal class DeleteAddress(BaseMutation): class Arguments: uuid = UUID(required=True) success = Boolean() @staticmethod def mutate(parent, info, uuid): # type: ignore [override] try: address = Address.objects.get(uuid=uuid) if ( info.context.user.is_superuser or info.context.user.has_perm("core.delete_address") or info.context.user == address.user ): address.delete() # noinspection PyTypeChecker return DeleteAddress(success=True) raise PermissionDenied(permission_denied_message) except Address.DoesNotExist as dne: name = "Address" raise Http404(_(f"{name} does not exist: {uuid}")) from dne # noinspection PyUnusedLocal class AutocompleteAddress(BaseMutation): class Arguments: q = String() limit = Int() suggestions = GenericScalar() @staticmethod def mutate(parent, info, q, limit): # type: ignore [override] if 1 > limit > 10: raise BadRequest(_("limit must be between 1 and 10")) try: suggestions = fetch_address_suggestions(query=q, limit=limit) except Exception as e: raise BadRequest(f"geocoding error: {e!s}") from e # noinspection PyTypeChecker return AutocompleteAddress(suggestions=suggestions) # noinspection PyUnusedLocal class ContactUs(BaseMutation): class Arguments: email = String(required=True) name = String(required=True) subject = String(required=True) phone_number = String(required=False) message = String(required=True) received = Boolean(required=True) error = String() @staticmethod def mutate(parent, info, email, name, subject, message, phone_number=None): # type: ignore [override] try: contact_us_email.delay( { "email": email, "name": name, "subject": subject, "phone_number": phone_number, "message": message, } ) # noinspection PyTypeChecker return ContactUs(received=True) except Exception as e: # noinspection PyTypeChecker return ContactUs(received=False, error=str(e)) # noinspection PyArgumentList PyUnusedLocal class Search(BaseMutation): class Arguments: query = String(required=True) results = Field(SearchResultsType) class Meta: description = _("elasticsearch - works like a charm") @staticmethod def mutate(parent, info, query): # type: ignore [override] data = process_query(query=query, request=info.context) # noinspection PyTypeChecker return Search( results=SearchResultsType( products=data["products"], categories=data["categories"], brands=data["brands"], posts=data["posts"], ) )