fix: Обработка race condition при дублировании slug товаров

Реализована трёхуровневая защита от IntegrityError при создании товаров с одинаковым названием:

1. SlugService улучшен:
   - Добавлен метод _get_base_queryset() для работы с мягким удалением
   - Проверка всех объектов включая удалённые (all_objects)
   - Новый метод get_next_available_slug() для retry обработки
   - Максимум 100 попыток поиска уникального slug

2. BaseProductEntity.save() защищена:
   - transaction.atomic() для атомарности операции
   - Retry логика с 5 попытками при IntegrityError
   - При конфликте добавляется суффикс (-1, -2, -3...)
   - Fallback на timestamp если суффиксы исчерпаны

3. Views обрабатывают IntegrityError:
   - ProductCreateView.form_valid() перехватывает ошибку
   - ProductUpdateView.form_valid() перехватывает ошибку
   - Пользователю показывается дружелюбное сообщение об ошибке
   - Нет 500 ошибок - вместо этого form_invalid() с сообщением

Эффект:
- До: User создаёт товар "Роза красная" 2 раза → IntegrityError → 500 ошибка
- После: User создаёт товар "Роза красная" 2 раза → slug автоматически становится "roza-krasnaya-1"

Протестировано на Django shell и синтаксис проверен.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-11-15 11:57:11 +03:00
parent 3426cdd091
commit 0bdb42401a
3 changed files with 176 additions and 35 deletions

View File

@@ -12,6 +12,27 @@ class SlugService:
Используется моделями Product, ProductKit, ProductCategory, ProductTag.
"""
@staticmethod
def _get_base_queryset(model_class):
"""
Получает базовый queryset для модели.
Использует all_objects если модель имеет мягкое удаление, иначе objects.
Это важно для проверки уникальности - нужно проверять ALL объекты,
включая удаленные, иначе могут быть дубли между deleted и active товарами.
Args:
model_class: Класс модели
Returns:
QuerySet для проверки уникальности
"""
# Если модель использует мягкое удаление (все наследники BaseProductEntity),
# используем all_objects для проверки ALL записей
if hasattr(model_class, 'all_objects'):
return model_class.all_objects
return model_class.objects
@staticmethod
def generate_unique_slug(name, model_class, instance_pk=None):
"""
@@ -38,10 +59,11 @@ class SlugService:
# Обеспечиваем уникальность
slug = base_slug
counter = 1
queryset = SlugService._get_base_queryset(model_class)
while True:
# Проверяем существование slug, исключая текущий экземпляр если это обновление
query = model_class.objects.filter(slug=slug)
query = queryset.filter(slug=slug)
if instance_pk:
query = query.exclude(pk=instance_pk)
@@ -54,6 +76,42 @@ class SlugService:
return slug
@staticmethod
def get_next_available_slug(base_slug, model_class, instance_pk=None, start_counter=1, max_attempts=100):
"""
Получает следующий доступный slug с числовым суффиксом.
Используется при retry обработке IntegrityError.
Args:
base_slug (str): Базовый slug (например, "test-product")
model_class: Класс модели для проверки уникальности
instance_pk (int, optional): ID текущего экземпляра
start_counter (int): Начальное значение счетчика (по умолчанию 1)
max_attempts (int): Максимум попыток (по умолчанию 100)
Returns:
str: Доступный slug с суффиксом (например, "test-product-1")
Raises:
ValueError: Если не найден доступный slug после max_attempts попыток
"""
queryset = SlugService._get_base_queryset(model_class)
for counter in range(start_counter, start_counter + max_attempts):
slug = f"{base_slug}-{counter}"
query = queryset.filter(slug=slug)
if instance_pk:
query = query.exclude(pk=instance_pk)
if not query.exists():
return slug
# Если все попытки исчерпаны
raise ValueError(
f"Не удалось сгенерировать уникальный slug для '{base_slug}' "
f"после {max_attempts} попыток"
)
@staticmethod
def transliterate(text):
"""