Egor fureunoir Gorbunov
ad320235d6
- Refactored monetary fields across models to use `DecimalField` for improved precision. - Implemented two-factor authentication (2FA) for admin logins with OTP codes. - Added ability to generate admin OTP via management commands. - Updated Docker Compose override for dev-specific port bindings. - Included template for 2FA OTP verification to enhance security. Additional changes: - Upgraded and downgraded various dependencies (e.g., django-celery-beat and yarl). - Replaced float-based calculations with decimal for consistent rounding behavior. - Improved admin user management commands for activation and OTP generation.
738 lines
25 KiB
Python
738 lines
25 KiB
Python
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, Mutation, String
|
|
from graphene.types.generic import GenericScalar
|
|
|
|
from engine.core.elasticsearch import process_query
|
|
from engine.core.graphene.object_types import (
|
|
AddressType,
|
|
BulkProductInput,
|
|
FeedbackType,
|
|
OrderType,
|
|
SearchResultsType,
|
|
WishlistType,
|
|
)
|
|
from engine.core.models import Address, Order, OrderProduct, Wishlist
|
|
from engine.core.utils import format_attributes, is_url_safe
|
|
from engine.core.utils.caching import web_cache
|
|
from engine.core.utils.emailing import contact_us_email
|
|
from engine.core.utils.messages import permission_denied_message
|
|
from engine.core.utils.nominatim import fetch_address_suggestions
|
|
from engine.payments.graphene.object_types import TransactionType
|
|
from schon.utils.ratelimit import graphql_ratelimit
|
|
from schon.utils.renderers import camelize
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class CacheOperator(Mutation):
|
|
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]:
|
|
return camelize(web_cache(info.context, key, data, timeout))
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class RequestCursedURL(Mutation):
|
|
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]:
|
|
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(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
except Order.DoesNotExist as dne:
|
|
raise Http404(_(f"order {order_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class RemoveOrderProduct(Mutation):
|
|
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):
|
|
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 RemoveOrderProduct(order=order) # ty: ignore[unknown-argument]
|
|
except Order.DoesNotExist as dne:
|
|
raise Http404(_(f"order {order_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class RemoveAllOrderProducts(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class RemoveOrderProductsOfAKind(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class BuyOrder(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="10/h")
|
|
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,
|
|
):
|
|
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( # ty: ignore[possibly-missing-attribute]
|
|
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 "<class 'engine.payments.models.Transaction'>":
|
|
return BuyOrder(transaction=instance) # ty: ignore[unknown-argument]
|
|
case "<class 'engine.core.models.Order'>":
|
|
return BuyOrder(order=instance) # ty: ignore[unknown-argument]
|
|
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(Mutation):
|
|
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,
|
|
):
|
|
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) # ty: ignore[possibly-missing-attribute]
|
|
case "remove":
|
|
order = order.bulk_remove_products(products) # ty: ignore[possibly-missing-attribute]
|
|
case _:
|
|
raise BadRequest(_("action must be either add or remove"))
|
|
|
|
return BulkOrderAction(order=order) # ty: ignore[unknown-argument]
|
|
|
|
except Order.DoesNotExist as dne:
|
|
raise Http404(_(f"order {order_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class BulkWishlistAction(Mutation):
|
|
class Meta:
|
|
description = _("perform an action on a list of products in the wishlist")
|
|
|
|
class Arguments:
|
|
wishlist_uuid = UUID(required=True)
|
|
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,
|
|
):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
except Wishlist.DoesNotExist as dne:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class BuyUnregisteredOrder(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="10/h")
|
|
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,
|
|
):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class AddWishlistProduct(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
except Wishlist.DoesNotExist as dne:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class RemoveWishlistProduct(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
except Wishlist.DoesNotExist as dne:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class RemoveAllWishlistProducts(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
except Wishlist.DoesNotExist as dne:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class BuyWishlist(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="10/h")
|
|
def mutate(parent, info, wishlist_uuid, force_balance=False, force_payment=False):
|
|
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 "<class 'engine.payments.models.Transaction'>":
|
|
return BuyWishlist(transaction=instance) # ty: ignore[unknown-argument]
|
|
case "<class 'engine.core.models.Order'>":
|
|
return BuyWishlist(order=instance) # ty: ignore[unknown-argument]
|
|
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(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="10/h")
|
|
def mutate(
|
|
parent,
|
|
info,
|
|
product_uuid,
|
|
attributes=None,
|
|
force_balance=False,
|
|
force_payment=False,
|
|
):
|
|
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 "<class 'engine.payments.models.Transaction'>":
|
|
return BuyProduct(transaction=instance) # ty: ignore[unknown-argument]
|
|
case "<class 'engine.core.models.Order'>":
|
|
return BuyProduct(order=instance) # ty: ignore[unknown-argument]
|
|
case _:
|
|
raise TypeError(
|
|
_(f"wrong type came from order.buy() method: {type(instance)!s}")
|
|
)
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class FeedbackProductAction(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="10/h")
|
|
def mutate(parent, info, order_product_uuid, action, comment=None, rating=None):
|
|
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) # ty: ignore[unknown-argument]
|
|
except OrderProduct.DoesNotExist as dne:
|
|
raise Http404(_(f"order product {order_product_uuid} not found")) from dne
|
|
|
|
|
|
# noinspection PyUnusedLocal,PyTypeChecker
|
|
class CreateAddress(Mutation):
|
|
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):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class DeleteAddress(Mutation):
|
|
class Arguments:
|
|
uuid = UUID(required=True)
|
|
|
|
success = Boolean()
|
|
|
|
@staticmethod
|
|
def mutate(parent, info, uuid):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
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(Mutation):
|
|
class Arguments:
|
|
q = String()
|
|
limit = Int()
|
|
|
|
suggestions = GenericScalar()
|
|
|
|
@staticmethod
|
|
def mutate(parent, info, q, limit):
|
|
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) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class ContactUs(Mutation):
|
|
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
|
|
@graphql_ratelimit(rate="2/h")
|
|
def mutate(parent, info, email, name, subject, message, phone_number=None):
|
|
try:
|
|
contact_us_email.delay(
|
|
{
|
|
"email": email,
|
|
"name": name,
|
|
"subject": subject,
|
|
"phone_number": phone_number,
|
|
"message": message,
|
|
}
|
|
)
|
|
# noinspection PyTypeChecker
|
|
return ContactUs(received=True) # ty: ignore[unknown-argument]
|
|
except Exception as e:
|
|
# noinspection PyTypeChecker
|
|
return ContactUs(received=False, error=str(e)) # ty: ignore[unknown-argument]
|
|
|
|
|
|
# noinspection PyArgumentList PyUnusedLocal
|
|
class Search(Mutation):
|
|
class Arguments:
|
|
query = String(required=True)
|
|
|
|
results = Field(SearchResultsType)
|
|
|
|
class Meta:
|
|
description = _("elasticsearch - works like a charm")
|
|
|
|
@staticmethod
|
|
def mutate(parent, info, query):
|
|
data = process_query(query=query, request=info.context)
|
|
|
|
if not data:
|
|
return Search(results=None) # ty: ignore[unknown-argument]
|
|
|
|
# noinspection PyTypeChecker
|
|
return Search(
|
|
results=SearchResultsType( # ty: ignore[unknown-argument]
|
|
products=data["products"], # ty: ignore[unknown-argument]
|
|
categories=data["categories"], # ty: ignore[unknown-argument]
|
|
brands=data["brands"], # ty: ignore[unknown-argument]
|
|
posts=data["posts"], # ty: ignore[unknown-argument]
|
|
)
|
|
)
|