Features: 1) Add b2b_auth_token and users fields to Vendor model with associated migration; 2) Introduce unique constraint for b2b_auth_token with migration; 3) Enhance VendorAdmin and Vendor model's save method to manage related users and token fields automatically;

Fixes: 1) Adjust `is_business` property logic for accuracy; 2) Fix import cleanup in serializers and utils files;

Extra: Refactor `core.models`, `core.utils`, and `core.vendors` for improved type annotations and other minor adjustments; Expand mypy exclusions in `pyproject.toml`.
This commit is contained in:
Egor Pavlovich Gorbunov 2025-07-28 11:56:32 +03:00
parent 5ae198911b
commit f52227973b
11 changed files with 146 additions and 62 deletions

View file

@ -312,8 +312,8 @@ class VendorAdmin(FieldsetsMixin, ActivationActionsMixin, ModelAdmin): # type:
readonly_fields = ("uuid", "modified", "created")
form = VendorForm
general_fields = ["is_active", "name", "markup_percent", "authentication"]
relation_fields = []
general_fields = ["is_active", "name", "markup_percent", "authentication", "b2b_auth_token"]
relation_fields = ["users"]
@register(Feedback)

View file

@ -13,6 +13,7 @@ from graphene import (
ObjectType,
String,
relay,
Boolean,
)
from graphene.types.generic import GenericScalar
from graphene_django import DjangoObjectType

View file

@ -0,0 +1,33 @@
# Generated by Django 5.2 on 2025-07-28 08:55
import core.utils
from django.conf import settings
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("core", "0035_alter_brand_slug_alter_category_slug_and_more"),
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.AddField(
model_name="vendor",
name="b2b_auth_token",
field=models.CharField(
blank=True,
default=core.utils.generate_human_readable_token,
max_length=20,
null=True,
),
),
migrations.AddField(
model_name="vendor",
name="users",
field=models.ManyToManyField(
blank=True, related_name="vendors", to=settings.AUTH_USER_MODEL
),
),
]

View file

@ -0,0 +1,25 @@
# Generated by Django 5.2 on 2025-07-28 08:55
import core.utils
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("core", "0036_vendor_b2b_auth_token_vendor_users"),
]
operations = [
migrations.AlterField(
model_name="vendor",
name="b2b_auth_token",
field=models.CharField(
blank=True,
default=core.utils.generate_human_readable_token,
max_length=20,
null=True,
unique=True,
),
),
]

View file

@ -50,6 +50,7 @@ from core.utils import (
generate_human_readable_id,
get_product_uuid_as_path,
get_random_code,
generate_human_readable_token,
)
from core.utils.db import TweakedAutoSlugField, unicode_slugify_function
from core.utils.lists import FAILED_STATUSES
@ -164,10 +165,23 @@ class Vendor(ExportModelOperationsMixin("vendor"), NiceModel): # type: ignore [
null=False,
unique=True,
)
users = ManyToManyField(to="vibes_auth.User", related_name="vendors", blank=True)
b2b_auth_token = CharField(default=generate_human_readable_token, max_length=20, unique=True, null=True, blank=True)
def __str__(self) -> str:
return self.name
def save(self, **kwargs):
users = self.users.filter(is_active=True)
users = users.exclude(attributes__icontains="is_business")
if users.count() > 0:
for user in users:
if not user.attributes:
user.attributes = {}
user.attributes.update({"is_business": True})
user.save()
return super().save(**kwargs)
class Meta:
verbose_name = _("vendor")
verbose_name_plural = _("vendors")
@ -1404,9 +1418,11 @@ class Order(ExportModelOperationsMixin("order"), NiceModel): # type: ignore [mi
@property
def is_business(self) -> bool:
return self.attributes.get("is_business", False) if self.attributes else False
return (self.user.attributes.get("is_business", False) if self.user else False) or (
self.attributes.get("is_business", False) if self.attributes else False
)
def save(self, **kwargs):
def save(self, **kwargs: dict) -> Self:
pending_orders = 0
if self.user:
pending_orders = self.user.orders.filter(status="PENDING").count()
@ -1559,8 +1575,8 @@ class Order(ExportModelOperationsMixin("order"), NiceModel): # type: ignore [mi
shipping_address = billing_address
else:
billing_address = Address.objects.get(uuid=billing_address_uuid) # type: ignore [misc]
shipping_address = Address.objects.get(uuid=shipping_address_uuid) # type: ignore [misc]
billing_address = Address.objects.get(uuid=billing_address_uuid)
shipping_address = Address.objects.get(uuid=shipping_address_uuid)
self.billing_address = billing_address
self.shipping_address = shipping_address

View file

@ -7,7 +7,6 @@ from django.contrib.auth.models import AnonymousUser
from django.core.cache import cache
from django.db.models.functions import Length
from rest_framework.fields import JSONField, SerializerMethodField
from rest_framework.request import Request
from rest_framework.serializers import ModelSerializer
from rest_framework_recursive.fields import RecursiveField
@ -31,7 +30,6 @@ from core.models import (
)
from core.serializers.simple import CategorySimpleSerializer, ProductSimpleSerializer
from core.serializers.utility import AddressSerializer
from vibes_auth.models import User
logger = logging.getLogger("django")

View file

@ -147,3 +147,10 @@ def generate_human_readable_id(length: int = 6) -> str:
chars.insert(pos, "-")
return "".join(chars)
def generate_human_readable_token() -> str:
"""
Generate a human-readable token of 20 characters (from the Crockford set),
"""
return "".join([secrets.choice(CROCKFORD) for _ in range(20)])

View file

@ -1,9 +1,11 @@
import json
from contextlib import suppress
from decimal import Decimal
from math import ceil, log10
from typing import Any
from django.db import IntegrityError
from django.db.models import QuerySet
from core.elasticsearch import process_query
from core.models import (
@ -73,13 +75,13 @@ class AbstractVendor:
instance.
"""
def __init__(self, vendor_name=None, currency="USD"):
def __init__(self, vendor_name: str = None, currency: str = "USD") -> None:
self.vendor_name = vendor_name
self.currency = currency
self.blocked_attributes = []
@staticmethod
def chunk_data(data, num_chunks=20):
def chunk_data(data: dict = list, num_chunks: int = 20) -> list[dict[Any, Any]] | None:
total = len(data)
if total == 0:
return []
@ -87,7 +89,7 @@ class AbstractVendor:
return [data[i : i + chunk_size] for i in range(0, total, chunk_size)]
@staticmethod
def auto_convert_value(value: Any):
def auto_convert_value(value: Any) -> tuple[Any, str]:
"""
Attempts to convert a value to a more specific type.
Handles booleans, numbers, objects (dicts), and arrays (lists),
@ -163,7 +165,7 @@ class AbstractVendor:
queryset.delete()
return chosen
def auto_resolve_category(self, category_name: str):
def auto_resolve_category(self, category_name: str = "") -> Category | None:
if category_name:
try:
search = process_query(category_name)
@ -181,7 +183,7 @@ class AbstractVendor:
return self.auto_resolver_helper(Category, category_name)
def auto_resolve_brand(self, brand_name: str):
def auto_resolve_brand(self, brand_name: str = "") -> Brand | None:
if brand_name:
try:
search = process_query(brand_name)
@ -220,7 +222,7 @@ class AbstractVendor:
return round(price, 2)
def resolve_price_with_currency(self, price, provider, currency=None):
def resolve_price_with_currency(self, price: float | int | Decimal, provider: str, currency: str = ""):
rates = get_rates(provider)
rate = rates.get(currency or self.currency)
@ -263,7 +265,7 @@ class AbstractVendor:
return float(psychological)
def get_vendor_instance(self):
def get_vendor_instance(self) -> Vendor | None:
try:
vendor = Vendor.objects.get(name=self.vendor_name)
if vendor.is_active:
@ -272,16 +274,16 @@ class AbstractVendor:
except Vendor.DoesNotExist as dne:
raise Exception(f"No matching vendor found with name {self.vendor_name!r}...") from dne
def get_products(self):
def get_products(self) -> None:
pass
def get_products_queryset(self):
def get_products_queryset(self) -> QuerySet[Product] | None:
return Product.objects.filter(stocks__vendor=self.get_vendor_instance(), orderproduct__isnull=True)
def get_stocks_queryset(self):
def get_stocks_queryset(self) -> QuerySet[Stock] | None:
return Stock.objects.filter(product__in=self.get_products_queryset(), product__orderproduct__isnull=True)
def get_attribute_values_queryset(self):
def get_attribute_values_queryset(self) -> QuerySet[AttributeValue] | None:
return AttributeValue.objects.filter(
product__in=self.get_products_queryset(), product__orderproduct__isnull=True
)
@ -300,7 +302,7 @@ class AbstractVendor:
case _:
raise ValueError(f"Invalid method {method!r} for products update...")
def delete_inactives(self, inactivation_method: str = "deactivate", size: int = 5000):
def delete_inactives(self, inactivation_method: str = "deactivate", size: int = 5000) -> None:
match inactivation_method:
case "deactivate":
filter_kwargs: dict[str, Any] = {"is_active": False}
@ -318,12 +320,12 @@ class AbstractVendor:
ProductImage.objects.filter(product_id__in=batch_ids).delete()
Product.objects.filter(pk__in=batch_ids).delete()
def delete_belongings(self):
def delete_belongings(self) -> None:
self.get_products_queryset().delete()
self.get_stocks_queryset().delete()
self.get_attribute_values_queryset().delete()
def process_attribute(self, key: str, value, product: Product, attr_group: AttributeGroup):
def process_attribute(self, key: str, value, product: Product, attr_group: AttributeGroup) -> None:
if not value:
return
@ -375,5 +377,5 @@ class AbstractVendor:
pass
def delete_stale():
def delete_stale() -> None:
Product.objects.filter(stocks__isnull=True, orderproduct__isnull=True).delete()

View file

@ -50,7 +50,8 @@ CONSTANCE_CONFIG = OrderedDict(
]
)
CONSTANCE_CONFIG_FIELDSETS = OrderedDict({
CONSTANCE_CONFIG_FIELDSETS = OrderedDict(
{
gettext_noop("General Options"): (
"PROJECT_NAME",
"FRONTEND_DOMAIN",
@ -86,7 +87,8 @@ CONSTANCE_CONFIG_FIELDSETS = OrderedDict({
"ADVERTISEMENT_DATA",
"ANALYTICS_DATA",
),
})
}
)
EXPOSABLE_KEYS = [
"PROJECT_NAME",

View file

@ -105,7 +105,7 @@ linting = ["black", "isort", "flake8", "bandit"]
[tool.mypy]
strict = true
disable_error_code = ["no-redef", "import-untyped"]
exclude = ["*/migrations/*"]
exclude = ["*/migrations/*", "storefront/*"]
plugins = ["mypy_django_plugin.main", "mypy_drf_plugin.main"]
[tool.django-stubs]

View file

@ -54,7 +54,7 @@ class UserAdmin(ActivationActionsMixin, BaseUserAdmin): # type: ignore [misc]
(None, {"fields": ("email", "password")}),
(
_("personal info"),
{"fields": ("first_name", "last_name", "phone_number", "avatar")},
{"fields": ("first_name", "last_name", "phone_number", "avatar", "address_set")},
),
(
_("permissions"),