fureunoir/schon
fureunoir/schon
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Features: 1) Add support for custom slugify_function across models; 2) Introduce logging for debug output in get_slug_fields; 3) Enhance help text for key fields to improve clarity;

Browse source 
Fixes: 1) Correct type hints and comments for better accuracy; 2) Address formatting inconsistencies in exception messages and queries;

Extra: Refactor long lines for improved readability and maintainability across functions and methods;
This commit is contained in:
Egor Pavlovich Gorbunov 2025-06-21 22:23:54 +03:00
parent 8ae4c35028
commit 9eb4ee72df
2 changed files with 113 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

143
core/models.py
View file

@ -35,6 +35,7 @@ from django.http import Http404
from django.utils import timezone
from django.utils.encoding import force_bytes
from django.utils.http import urlsafe_base64_encode
from django.utils.text import slugify
from django.utils.translation import gettext_lazy as _
from django_extensions.db.fields import AutoSlugField
from django_prometheus.models import ExportModelOperationsMixin
@ -92,7 +93,9 @@ class Vendor(ExportModelOperationsMixin("vendor"), NiceModel):
authentication: dict = JSONField( # type: ignore
blank=True,
null=True,
help_text=_("stores credentials and endpoints required for vendor communication"),
help_text=_(
"stores credentials and endpoints required for vendor communication"
),
verbose_name=_("authentication info"),
)
markup_percent: int = IntegerField( # type: ignore
@ -217,6 +220,7 @@ class Category(ExportModelOperationsMixin("category"), NiceModel, MPTTModel):
"parent__name",
"name",
),
slugify_function=slugify,
allow_unicode=True,
unique=True,
editable=False,
@ -237,7 +241,10 @@ class Category(ExportModelOperationsMixin("category"), NiceModel, MPTTModel):
def get_tree_depth(self):
if self.is_leaf_node():
return 0
return self.get_descendants().aggregate(max_depth=Max("level"))["max_depth"] - self.get_level()
return (
self.get_descendants().aggregate(max_depth=Max("level"))["max_depth"]
- self.get_level()
)
class Meta:
verbose_name = _("category")
@ -290,9 +297,10 @@ class Brand(ExportModelOperationsMixin("brand"), NiceModel):
max_length=88,
overwrite=True,
null=True,
slugify_function=slugify,
verbose_name=_("brand slug"),
)
priority: int = PositiveIntegerField(
priority: int = PositiveIntegerField( # type: ignore
default=0,
null=False,
blank=False,
@ -368,6 +376,7 @@ class Product(ExportModelOperationsMixin("product"), NiceModel):
overwrite=True,
allow_unicode=True,
unique=True,
slugify_function=slugify,
editable=False,
null=True,
)
@ -398,7 +407,9 @@ class Product(ExportModelOperationsMixin("product"), NiceModel):
cache_key = f"product_feedbacks_count_{self.pk}"
feedbacks_count = cache.get(cache_key)
if feedbacks_count is None:
feedbacks_count = Feedback.objects.filter(order_product__product_id=self.pk).count()
feedbacks_count = Feedback.objects.filter(
order_product__product_id=self.pk
).count()
cache.set(cache_key, feedbacks_count, 604800)
return feedbacks_count
@ -697,7 +708,9 @@ class Wishlist(ExportModelOperationsMixin("wishlist"), NiceModel):
class Documentary(ExportModelOperationsMixin("attribute_group"), NiceModel):
is_publicly_visible = True
product: ForeignKey = ForeignKey(to=Product, on_delete=CASCADE, related_name="documentaries")
product: ForeignKey = ForeignKey(
to=Product, on_delete=CASCADE, related_name="documentaries"
)
document = FileField(upload_to=get_product_uuid_as_path)
class Meta:
@ -825,7 +838,9 @@ class PromoCode(ExportModelOperationsMixin("promocode"), NiceModel):
self.discount_amount is None and self.discount_percent is None
):
raise ValidationError(
_("only one type of discount should be defined (amount or percent), but not both or neither.")
_(
"only one type of discount should be defined (amount or percent), but not both or neither."
)
)
super().save(**kwargs)
@ -846,11 +861,15 @@ class PromoCode(ExportModelOperationsMixin("promocode"), NiceModel):
if self.discount_type == "percent":
amount -= round(amount * (self.discount_percent / 100), 2)
order.attributes.update({"promocode": str(self.uuid), "final_price": amount})
order.attributes.update(
{"promocode": str(self.uuid), "final_price": amount}
)
order.save()
elif self.discount_type == "amount":
amount -= round(float(self.discount_amount), 2)
order.attributes.update({"promocode": str(self.uuid), "final_price": amount})
order.attributes.update(
{"promocode": str(self.uuid), "final_price": amount}
)
order.save()
else:
raise ValueError(_(f"invalid discount type for promocode {self.uuid}"))
@ -958,7 +977,8 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
sum(
(
order_product.buy_price * order_product.quantity
if order_product.status not in FAILED_STATUSES and order_product.buy_price is not None
if order_product.status not in FAILED_STATUSES
and order_product.buy_price is not None
else 0.0
)
for order_product in self.order_products.all()
@ -982,7 +1002,9 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
attributes = []
if self.status not in ["PENDING", "MOMENTAL"]:
raise ValueError(_("you cannot add products to an order that is not a pending one"))
raise ValueError(
_("you cannot add products to an order that is not a pending one")
)
try:
product = Product.objects.get(uuid=product_uuid)
@ -991,7 +1013,9 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
buy_price = product.price
promotions = Promotion.objects.filter(is_active=True, products__in=[product]).order_by("discount_percent")
promotions = Promotion.objects.filter(
is_active=True, products__in=[product]
).order_by("discount_percent")
if promotions.exists():
buy_price -= round(product.price * (promotions.first().discount_percent / 100), 2) # type: ignore
@ -1004,7 +1028,9 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
)
if not is_created and update_quantity:
if product.quantity < order_product.quantity + 1:
raise BadRequest(_("you cannot add more products than available in stock"))
raise BadRequest(
_("you cannot add more products than available in stock")
)
order_product.quantity += 1
order_product.buy_price = product.price
order_product.save()
@ -1025,7 +1051,9 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
attributes = {}
if self.status not in ["PENDING", "MOMENTAL"]:
raise ValueError(_("you cannot remove products from an order that is not a pending one"))
raise ValueError(
_("you cannot remove products from an order that is not a pending one")
)
try:
product = Product.objects.get(uuid=product_uuid)
order_product = self.order_products.get(product=product, order=self)
@ -1044,12 +1072,16 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
raise Http404(_(f"{name} does not exist: {product_uuid}"))
except OrderProduct.DoesNotExist:
name = "OrderProduct"
query = f"product: {product_uuid}, order: {self.uuid}, attributes: {attributes}"
query = (
f"product: {product_uuid}, order: {self.uuid}, attributes: {attributes}"
)
raise Http404(_(f"{name} does not exist with query <{query}>"))
def remove_all_products(self):
if self.status not in ["PENDING", "MOMENTAL"]:
raise ValueError(_("you cannot remove products from an order that is not a pending one"))
raise ValueError(
_("you cannot remove products from an order that is not a pending one")
)
for order_product in self.order_products.all():
self.order_products.remove(order_product)
order_product.delete()
@ -1057,7 +1089,9 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
def remove_products_of_a_kind(self, product_uuid: str):
if self.status not in ["PENDING", "MOMENTAL"]:
raise ValueError(_("you cannot remove products from an order that is not a pending one"))
raise ValueError(
_("you cannot remove products from an order that is not a pending one")
)
try:
product = Product.objects.get(uuid=product_uuid)
order_product = self.order_products.get(product=product, order=self)
@ -1070,7 +1104,10 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
@property
def is_whole_digital(self):
return self.order_products.count() == self.order_products.filter(product__is_digital=True).count()
return (
self.order_products.count()
== self.order_products.filter(product__is_digital=True).count()
)
def apply_promocode(self, promocode_uuid: str):
try:
@ -1085,7 +1122,11 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
if self.is_whole_digital:
return
else:
raise ValueError(_("you can only buy physical products with shipping address specified"))
raise ValueError(
_(
"you can only buy physical products with shipping address specified"
)
)
if billing_address_uuid and not shipping_address_uuid:
shipping_address = Address.objects.get(uuid=billing_address_uuid)
@ -1115,9 +1156,13 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
shipping_address: str | None = None,
) -> Self | Transaction | None:
if config.DISABLED_COMMERCE:
raise DisabledCommerceError(_("you can not buy at this moment, please try again in a few minutes"))
raise DisabledCommerceError(
_("you can not buy at this moment, please try again in a few minutes")
)
if (not force_balance and not force_payment) or (force_balance and force_payment):
if (not force_balance and not force_payment) or (
force_balance and force_payment
):
raise ValueError(_("invalid force value"))
self.apply_addresses(billing_address, shipping_address)
@ -1133,12 +1178,16 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
if force_payment:
force = "payment"
amount = self.apply_promocode(promocode_uuid) if promocode_uuid else self.total_price
amount = (
self.apply_promocode(promocode_uuid) if promocode_uuid else self.total_price
)
match force:
case "balance":
if self.user.payments_balance.amount < amount: # type: ignore
raise NotEnoughMoneyError(_("insufficient funds to complete the order"))
raise NotEnoughMoneyError(
_("insufficient funds to complete the order")
)
self.status = "CREATED"
self.buy_time = timezone.now()
self.order_products.all().update(status="DELIVERING")
@ -1156,9 +1205,13 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
return self
def buy_without_registration(self, products: list, promocode_uuid: str, **kwargs) -> Transaction | None:
def buy_without_registration(
self, products: list, promocode_uuid: str, **kwargs
) -> Transaction | None:
if config.DISABLED_COMMERCE:
raise DisabledCommerceError(_("you can not buy at this moment, please try again in a few minutes"))
raise DisabledCommerceError(
_("you can not buy at this moment, please try again in a few minutes")
)
if len(products) < 1:
raise ValueError(_("you cannot purchase an empty order!"))
@ -1179,17 +1232,25 @@ class Order(ExportModelOperationsMixin("order"), NiceModel):
available_payment_methods = cache.get("payment_methods").get("payment_methods")
if payment_method not in available_payment_methods:
raise ValueError(_(f"invalid payment method: {payment_method} from {available_payment_methods}"))
raise ValueError(
_(
f"invalid payment method: {payment_method} from {available_payment_methods}"
)
)
billing_customer_address_uuid = kwargs.get("billing_customer_address")
shipping_customer_address_uuid = kwargs.get("shipping_customer_address")
self.apply_addresses(billing_customer_address_uuid, shipping_customer_address_uuid)
self.apply_addresses(
billing_customer_address_uuid, shipping_customer_address_uuid
)
for product_uuid in products:
self.add_product(product_uuid)
amount = self.apply_promocode(promocode_uuid) if promocode_uuid else self.total_price
amount = (
self.apply_promocode(promocode_uuid) if promocode_uuid else self.total_price
)
self.status = "CREATED"
@ -1334,7 +1395,9 @@ class OrderProduct(ExportModelOperationsMixin("order_product"), NiceModel):
"errors": [
{
"detail": (
error if error else f"Something went wrong with {self.uuid} for some reason..."
error
if error
else f"Something went wrong with {self.uuid} for some reason..."
)
},
]
@ -1359,7 +1422,9 @@ class OrderProduct(ExportModelOperationsMixin("order_product"), NiceModel):
return self.download.url
return ""
def do_feedback(self, rating: int = 10, comment: str = "", action: str = "add") -> Optional["Feedback"]:
def do_feedback(
self, rating: int = 10, comment: str = "", action: str = "add"
) -> Optional["Feedback"]:
if action not in ["add", "remove"]:
raise ValueError(_(f"wrong action specified for feedback: {action}"))
if action == "remove" and self.feedback:
@ -1367,9 +1432,13 @@ class OrderProduct(ExportModelOperationsMixin("order_product"), NiceModel):
return None
if action == "add" and not self.feedback:
if self.order.status not in ["MOMENTAL", "PENDING"]: # type: ignore
return Feedback.objects.create(rating=rating, comment=comment, order_product=self)
return Feedback.objects.create(
rating=rating, comment=comment, order_product=self
)
else:
raise ValueError(_("you cannot feedback an order which is not received"))
raise ValueError(
_("you cannot feedback an order which is not received")
)
return None
@ -1389,11 +1458,11 @@ class DigitalAssetDownload(ExportModelOperationsMixin("attribute_group"), NiceMo
@property
def url(self):
if self.order_product.status != "FINISHED":
raise ValueError(_("you can not download a digital asset for a non-finished order"))
raise ValueError(
_("you can not download a digital asset for a non-finished order")
)
return (
f"https://api.{config.BASE_DOMAIN}/download/{urlsafe_base64_encode(force_bytes(self.order_product.uuid))}"
)
return f"https://api.{config.BASE_DOMAIN}/download/{urlsafe_base64_encode(force_bytes(self.order_product.uuid))}"
class Feedback(ExportModelOperationsMixin("feedback"), NiceModel):
@ -1410,7 +1479,9 @@ class Feedback(ExportModelOperationsMixin("feedback"), NiceModel):
on_delete=CASCADE,
blank=False,
null=False,
help_text=_("references the specific product in an order that this feedback is about"),
help_text=_(
"references the specific product in an order that this feedback is about"
),
verbose_name=_("related order product"),
)
rating: float = FloatField( # type: ignore

6
core/utils/db.py
View file

@ -1,8 +1,11 @@
import logging
from django.db.models import Model
from django.db.models.constants import LOOKUP_SEP
from django.utils.translation import gettext_lazy as _
from django_extensions.db.fields import AutoSlugField
logger = logging.getLogger(__name__)
def list_to_queryset(model: Model, data: list):
if not isinstance(model, Model):
@ -21,6 +24,9 @@ class TweakedAutoSlugField(AutoSlugField):
lookup_value_path = lookup_value.split(LOOKUP_SEP)
attr = model_instance
logger.debug(f"get_slug_fields: lookup_value_path is {lookup_value_path} and attr is {attr}")
for elem in lookup_value_path:
try:
attr = getattr(attr, elem)