mirror of
https://github.com/zulip/zulip.git
synced 2025-10-23 16:14:02 +00:00
7051d3416b
Previously, it seemed possible for the scheduled messages API to try to send infinite copies of a message if we had the very poor luck of a persistent failure happening after a message was sent. The failure_message field supports being able to display what happened in the scheduled messages modal, though that's not exposed to the API yet.
37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
from typing import List, Union
|
|
|
|
from django.utils.translation import gettext as _
|
|
|
|
from zerver.lib.exceptions import ResourceNotFoundError
|
|
from zerver.models import (
|
|
APIScheduledDirectMessageDict,
|
|
APIScheduledStreamMessageDict,
|
|
ScheduledMessage,
|
|
UserProfile,
|
|
)
|
|
|
|
|
|
def access_scheduled_message(
|
|
user_profile: UserProfile, scheduled_message_id: int
|
|
) -> ScheduledMessage:
|
|
try:
|
|
return ScheduledMessage.objects.get(id=scheduled_message_id, sender=user_profile)
|
|
except ScheduledMessage.DoesNotExist:
|
|
raise ResourceNotFoundError(_("Scheduled message does not exist"))
|
|
|
|
|
|
def get_undelivered_scheduled_messages(
|
|
user_profile: UserProfile,
|
|
) -> List[Union[APIScheduledDirectMessageDict, APIScheduledStreamMessageDict]]:
|
|
scheduled_messages = ScheduledMessage.objects.filter(
|
|
sender=user_profile,
|
|
# Notably, we don't require failed=False, since we will want
|
|
# to display those to users.
|
|
delivered=False,
|
|
delivery_type=ScheduledMessage.SEND_LATER,
|
|
).order_by("scheduled_timestamp")
|
|
scheduled_message_dicts: List[
|
|
Union[APIScheduledDirectMessageDict, APIScheduledStreamMessageDict]
|
|
] = [scheduled_message.to_dict() for scheduled_message in scheduled_messages]
|
|
return scheduled_message_dicts
|