589 lines
19 KiB
Python
589 lines
19 KiB
Python
import logging
|
|
|
|
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, OrderType, ProductType, SearchResultsType, WishlistType
|
|
from core.models import Address, Category, Order, Product, Wishlist
|
|
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(__name__)
|
|
|
|
|
|
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):
|
|
return camelize(web_cache(info.context, key, data, timeout))
|
|
|
|
|
|
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):
|
|
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)}}
|
|
|
|
|
|
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):
|
|
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:
|
|
raise Http404(_(f"order {order_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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:
|
|
raise Http404(_(f"order {order_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
class BuyOrder(BaseMutation):
|
|
class Meta:
|
|
description = _("buy an order")
|
|
|
|
class Arguments:
|
|
order_uuid = UUID(required=False)
|
|
order_hr_id = String(required=False)
|
|
force_balance = Boolean(required=False)
|
|
force_payment = Boolean(required=False)
|
|
promocode_uuid = UUID(required=False)
|
|
shipping_address = UUID(required=False)
|
|
billing_address = UUID(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,
|
|
):
|
|
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,
|
|
)
|
|
|
|
match str(type(instance)):
|
|
case "<class 'payments.models.Transaction'>":
|
|
return BuyOrder(transaction=instance)
|
|
case "<class 'core.models.Order'>":
|
|
return BuyOrder(order=instance)
|
|
case _:
|
|
raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}"))
|
|
|
|
except Order.DoesNotExist:
|
|
raise Http404(_(f"order {order_uuid} not found"))
|
|
|
|
|
|
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,
|
|
):
|
|
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,
|
|
)
|
|
return BuyUnregisteredOrder(transaction=transaction)
|
|
|
|
|
|
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):
|
|
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:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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 'payments.models.Transaction'>":
|
|
return BuyWishlist(transaction=instance)
|
|
case "<class 'core.models.Order'>":
|
|
return BuyWishlist(order=instance)
|
|
case _:
|
|
raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}"))
|
|
|
|
except Wishlist.DoesNotExist:
|
|
raise Http404(_(f"wishlist {wishlist_uuid} not found"))
|
|
|
|
|
|
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):
|
|
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 'payments.models.Transaction'>":
|
|
return BuyProduct(transaction=instance)
|
|
case "<class 'core.models.Order'>":
|
|
return BuyProduct(order=instance)
|
|
case _:
|
|
raise TypeError(_(f"wrong type came from order.buy() method: {type(instance)!s}"))
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
class DeleteProduct(BaseMutation):
|
|
class Arguments:
|
|
uuid = UUID(required=True)
|
|
|
|
ok = Boolean()
|
|
|
|
@staticmethod
|
|
def mutate(_parent, info, uuid):
|
|
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)
|
|
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
class DeleteAddress(BaseMutation):
|
|
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()
|
|
return DeleteAddress(success=True)
|
|
|
|
raise PermissionDenied(permission_denied_message)
|
|
|
|
except Address.DoesNotExist:
|
|
name = "Address"
|
|
raise Http404(_(f"{name} does not exist: {uuid}"))
|
|
|
|
|
|
class AutocompleteAddress(BaseMutation):
|
|
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
|
|
|
|
return AutocompleteAddress(suggestions=suggestions)
|
|
|
|
|
|
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):
|
|
try:
|
|
contact_us_email.delay(
|
|
{
|
|
"email": email,
|
|
"name": name,
|
|
"subject": subject,
|
|
"phone_number": phone_number,
|
|
"message": message,
|
|
}
|
|
)
|
|
return ContactUs(received=True)
|
|
except Exception as e:
|
|
return ContactUs(received=False, error=str(e))
|
|
|
|
|
|
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):
|
|
data = process_query(query)
|
|
|
|
return Search(
|
|
results=SearchResultsType(
|
|
products=data["products"],
|
|
categories=data["categories"],
|
|
brands=data["brands"],
|
|
posts=data["posts"],
|
|
)
|
|
)
|