feat: асинхронная обработка уведомлений для requests и documents
Обновлены сигналы для полностью асинхронной обработки: - requests_app: новые заявки, комментарии, изменения статуса - documents: новые документы для всех сотрудников Добавлена задача process_document_notifications_task Все циклы for recipient in recipients теперь выполняются в фоне: ✅ Сигналы используют transaction.on_commit + .delay() ✅ Signal handlers сразу возвращают управление ✅ API не блокируется при массовых рассылках Fixes: Блокировка транзакций при создании заявок/документов
This commit is contained in:
parent
914d90736b
commit
6c99e4bf8c
@ -23,13 +23,16 @@ Employee = get_user_model()
|
||||
@receiver(post_save, sender=Document)
|
||||
def create_document_notification(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Создает уведомления при создании нового документа.
|
||||
Создает уведомления при создании нового документа асинхронно.
|
||||
|
||||
Уведомления отправляются:
|
||||
- Всем активным сотрудникам (если sent_to_all=True)
|
||||
- Сотрудникам выбранных отделов (через m2m_changed signal)
|
||||
- Выбранным получателям (через m2m_changed signal)
|
||||
"""
|
||||
from django.db import transaction
|
||||
from notifications.tasks import process_document_notifications_task
|
||||
|
||||
logger.info(
|
||||
f"[notification_signals] post_save Document id={instance.pk} "
|
||||
f"created={created} sent_to_all={instance.sent_to_all}"
|
||||
@ -40,12 +43,25 @@ def create_document_notification(sender, instance, created, **kwargs):
|
||||
|
||||
document = instance
|
||||
|
||||
# Если документ отправляется всем - создаем уведомления сразу
|
||||
# Если документ отправляется всем - создаем уведомления сразу асинхронно
|
||||
if document.sent_to_all:
|
||||
logger.info(
|
||||
f"[notification_signals] Calling notify_all_employees for doc={document.pk}"
|
||||
f"[notification_signals] Queuing task for doc={document.pk} (sent_to_all=True)"
|
||||
)
|
||||
notify_all_employees(document)
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_document_notifications_task.delay(document_id=document.pk)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to queue document notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_all_employees(document)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
else:
|
||||
logger.info(
|
||||
f"[notification_signals] Skipping notifications (sent_to_all=False), "
|
||||
|
||||
@ -732,3 +732,87 @@ def process_request_notifications_task(
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process request notifications: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
|
||||
@shared_task(
|
||||
name="documents.process_document_notifications",
|
||||
bind=True,
|
||||
max_retries=3,
|
||||
default_retry_delay=30,
|
||||
)
|
||||
def process_document_notifications_task(self, document_id: int):
|
||||
"""
|
||||
Обрабатывает уведомления о новом документе асинхронно.
|
||||
|
||||
Args:
|
||||
document_id: ID документа
|
||||
"""
|
||||
try:
|
||||
from documents.models import Document
|
||||
from notifications.services import NotificationService
|
||||
|
||||
document = Document.objects.select_related(
|
||||
'uploaded_by'
|
||||
).prefetch_related('recipients', 'departments').get(id=document_id)
|
||||
|
||||
recipients = set()
|
||||
|
||||
# Определяем получателей
|
||||
if document.sent_to_all:
|
||||
# Все активные сотрудники
|
||||
recipients = set(User.objects.filter(is_active=True))
|
||||
else:
|
||||
# Конкретные получатели
|
||||
recipients.update(document.recipients.filter(is_active=True))
|
||||
|
||||
# Сотрудники отделов
|
||||
for dept in document.departments.all():
|
||||
dept_employees = dept.employees.filter(
|
||||
departments_links__is_active=True,
|
||||
is_active=True
|
||||
)
|
||||
recipients.update(dept_employees)
|
||||
|
||||
# Исключаем загрузившего
|
||||
if document.uploaded_by:
|
||||
recipients.discard(document.uploaded_by)
|
||||
|
||||
if not recipients:
|
||||
return {"status": "success", "notifications_sent": 0}
|
||||
|
||||
notifications_count = 0
|
||||
|
||||
# Отправляем уведомления
|
||||
for recipient in recipients:
|
||||
NotificationService.create_notification(
|
||||
recipient=recipient,
|
||||
notification_type_code='document_new',
|
||||
title='Новый документ для ознакомления',
|
||||
message=f'Документ "{document.title}" требует ознакомления',
|
||||
content_object=document,
|
||||
action_url=f'/documents/{document.id}/',
|
||||
metadata={
|
||||
'document_id': document.id,
|
||||
'document_title': document.title,
|
||||
'uploaded_by_id': document.uploaded_by.id if document.uploaded_by else None,
|
||||
}
|
||||
)
|
||||
notifications_count += 1
|
||||
|
||||
logger.info(
|
||||
f"Processed document {document_id}: sent {notifications_count} notifications"
|
||||
)
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"document_id": document_id,
|
||||
"notifications_sent": notifications_count
|
||||
}
|
||||
|
||||
except Document.DoesNotExist:
|
||||
logger.error(f"Document {document_id} not found")
|
||||
return {"status": "error", "reason": "document_not_found"}
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception(f"Failed to process document notifications: {exc}")
|
||||
raise self.retry(exc=exc)
|
||||
|
||||
@ -63,8 +63,28 @@ def create_request_notifications(sender, instance, created, **kwargs):
|
||||
new_status = request_obj.status
|
||||
|
||||
if old_status != new_status:
|
||||
# Передаем текущий request_obj с актуальным approver_id
|
||||
notify_status_change(request_obj, old_status, new_status)
|
||||
from notifications.tasks import process_request_notifications_task
|
||||
|
||||
# Определяем action по статусу
|
||||
action = 'approved' if new_status == 'approved' else \
|
||||
'rejected' if new_status == 'rejected' else 'updated'
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_request_notifications_task.delay(
|
||||
request_id=request_obj.id,
|
||||
action=action
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to queue status change notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_status_change(request_obj, old_status, new_status)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
except Exception as e:
|
||||
logger.exception(f"[SIGNAL ERROR] create_request_notifications: {e}")
|
||||
print(f"❌ [SIGNAL ERROR] create_request_notifications: {e}")
|
||||
@ -91,13 +111,31 @@ def notify_on_recipients_changed(sender, instance, action, **kwargs):
|
||||
Срабатывает когда recipients устанавливаются через .set(), .add() и т.д.
|
||||
"""
|
||||
try:
|
||||
from notifications.tasks import process_request_notifications_task
|
||||
|
||||
# Проверяем что это завершение операции установки recipients
|
||||
if action == "post_add" and instance.id in _pending_new_requests:
|
||||
print(
|
||||
f"🎯 [M2M_SIGNAL] Recipients установлены для заявления #{instance.id}, отправляем уведомления"
|
||||
)
|
||||
_pending_new_requests.discard(instance.id)
|
||||
notify_new_request(instance)
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_request_notifications_task.delay(
|
||||
request_id=instance.id,
|
||||
action='created'
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to queue new request notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_new_request(instance)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
except Exception as e:
|
||||
logger.exception(f"[SIGNAL ERROR] notify_on_recipients_changed: {e}")
|
||||
print(f"❌ [SIGNAL ERROR] notify_on_recipients_changed: {e}")
|
||||
@ -112,6 +150,8 @@ def notify_on_cc_users_changed(sender, instance, action, **kwargs):
|
||||
также отправляем уведомления.
|
||||
"""
|
||||
try:
|
||||
from notifications.tasks import process_request_notifications_task
|
||||
|
||||
# Если это завершение добавления cc_users И заявление все еще ожидает уведомлений
|
||||
if action == "post_add" and instance.id in _pending_new_requests:
|
||||
# Проверяем что recipients пусты (значит уведомления еще не отправлены)
|
||||
@ -120,7 +160,23 @@ def notify_on_cc_users_changed(sender, instance, action, **kwargs):
|
||||
f"🎯 [M2M_SIGNAL] CC users установлены для заявления #{instance.id} (без recipients), отправляем уведомления"
|
||||
)
|
||||
_pending_new_requests.discard(instance.id)
|
||||
notify_new_request(instance)
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_request_notifications_task.delay(
|
||||
request_id=instance.id,
|
||||
action='created'
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to queue new request notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_new_request(instance)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
except Exception as e:
|
||||
logger.exception(f"[SIGNAL ERROR] notify_on_cc_users_changed: {e}")
|
||||
print(f"❌ [SIGNAL ERROR] notify_on_cc_users_changed: {e}")
|
||||
@ -129,7 +185,7 @@ def notify_on_cc_users_changed(sender, instance, action, **kwargs):
|
||||
@receiver(post_save, sender=RequestComment)
|
||||
def create_comment_notification(sender, instance, created, **kwargs):
|
||||
"""
|
||||
Создает уведомление при добавлении комментария к заявлению.
|
||||
Создает уведомления при добавлении комментария к заявлению асинхронно.
|
||||
Уведомляет:
|
||||
- Автора заявления
|
||||
- Всех получателей
|
||||
@ -139,9 +195,26 @@ def create_comment_notification(sender, instance, created, **kwargs):
|
||||
"""
|
||||
if not created:
|
||||
return
|
||||
|
||||
comment = instance
|
||||
request_obj = comment.request
|
||||
|
||||
from notifications.tasks import process_request_notifications_task
|
||||
|
||||
def send_task():
|
||||
"""Отложенная отправка задачи после commit транзакции"""
|
||||
try:
|
||||
process_request_notifications_task.delay(
|
||||
request_id=instance.request.id,
|
||||
action='comment',
|
||||
comment_id=instance.id
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to queue comment notifications task: {e}, "
|
||||
f"falling back to sync"
|
||||
)
|
||||
# Fallback на старую логику
|
||||
notify_request_comment(instance)
|
||||
|
||||
transaction.on_commit(send_task)
|
||||
author = comment.author
|
||||
recipients_set = set()
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user