Files
octopus/myproject/products/services/slug_service.py
Andrey Smakotin 0bdb42401a 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>
2025-11-15 11:57:11 +03:00

131 lines
5.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Сервис для генерации уникальных slug для моделей.
Централизует логику транслитерации и обеспечения уникальности.
"""
from django.utils.text import slugify
from unidecode import unidecode
class SlugService:
"""
Статический сервис для генерации уникальных slug.
Используется моделями 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):
"""
Генерирует уникальный slug из названия с транслитерацией кириллицы.
Args:
name (str): Исходное название для генерации slug
model_class (Model): Класс модели для проверки уникальности
instance_pk (int, optional): ID текущего экземпляра (для исключения при обновлении)
Returns:
str: Уникальный slug
Example:
>>> SlugService.generate_unique_slug("Роза красная", Product, None)
'roza-krasnaya'
>>> SlugService.generate_unique_slug("Роза красная", Product, None) # если уже существует
'roza-krasnaya-1'
"""
# Транслитерируем кириллицу в латиницу, затем применяем slugify
transliterated_name = unidecode(name)
base_slug = slugify(transliterated_name)
# Обеспечиваем уникальность
slug = base_slug
counter = 1
queryset = SlugService._get_base_queryset(model_class)
while True:
# Проверяем существование slug, исключая текущий экземпляр если это обновление
query = queryset.filter(slug=slug)
if instance_pk:
query = query.exclude(pk=instance_pk)
if not query.exists():
break
# Если slug занят, добавляем счетчик
slug = f"{base_slug}-{counter}"
counter += 1
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):
"""
Транслитерирует текст (кириллицу в латиницу).
Args:
text (str): Текст для транслитерации
Returns:
str: Транслитерированный текст
Example:
>>> SlugService.transliterate("Привет мир")
'Privet mir'
"""
return unidecode(text)