Django messages

hypha.apply.activity.adapters.django_messages

DjangoMessagesAdapter

Bases: AdapterBase

adapter_type class-attribute instance-attribute

adapter_type = 'Django'

always_send class-attribute instance-attribute

always_send = True

messages class-attribute instance-attribute

messages = {BATCH_REVIEWERS_UPDATED: 'batch_reviewers_updated', BATCH_TRANSITION: 'batch_transition', BATCH_DETERMINATION_OUTCOME: 'batch_determinations', REMOVE_DOCUMENT: gettext('Successfully removed document'), SKIPPED_REPORT: 'handle_skipped_report', REPORT_FREQUENCY_CHANGED: 'handle_report_frequency', DISABLED_REPORTING: gettext('Reporting disabled'), CREATE_REMINDER: gettext('Reminder created'), DELETE_REMINDER: gettext('Reminder deleted')}

message

message(message_type, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def message(self, message_type, **kwargs):
    try:
        message = self.messages[message_type]
    except KeyError:
        # We don't know how to handle that message type
        return

    try:
        # see if its a method on the adapter
        method = getattr(self, message)
    except AttributeError:
        return self.render_message(message, **kwargs)
    else:
        # Delegate all responsibility to the custom method
        return method(**kwargs)

render_message

render_message(message, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def render_message(self, message, **kwargs):
    return message.format(**kwargs)

extra_kwargs

extra_kwargs(message_type, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def extra_kwargs(self, message_type, **kwargs):
    return {}

get_neat_related

get_neat_related(message_type, related)

Source code in hypha/apply/activity/adapters/base.py

def get_neat_related(self, message_type, related):
    # We translate the related kwarg into something we can understand
    try:
        neat_name = neat_related[message_type]
    except KeyError:
        # Message type doesn't expect a related object
        if related:
            raise ValueError(
                f"Unexpected 'related' kwarg provided for {message_type}"
            ) from None
        return {}
    else:
        if not related:
            raise ValueError(f"{message_type} expects a 'related' kwarg")
        return {neat_name: related}

process_batch

process_batch(message_type, events, request, user, sources, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process_batch(
    self, message_type, events, request, user, sources, related=None, **kwargs
):
    events_by_source = {event.source.id: event for event in events}
    for recipient in self.batch_recipients(
        message_type, sources, user=user, **kwargs
    ):
        recipients = recipient["recipients"]
        sources = recipient["sources"]
        events = [events_by_source[source.id] for source in sources]
        self.process_send(
            message_type,
            recipients,
            events,
            request,
            user,
            sources=sources,
            source=None,
            related=related,
            **kwargs,
        )

process

process(message_type, event, request, user, source, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process(
    self, message_type, event, request, user, source, related=None, **kwargs
):
    recipients = self.recipients(
        message_type,
        source=source,
        related=related,
        user=user,
        request=request,
        **kwargs,
    )
    self.process_send(
        message_type,
        recipients,
        [event],
        request,
        user,
        source,
        related=related,
        **kwargs,
    )

process_send

process_send(message_type, recipients, events, request, user, source, sources=None, related=None, **kwargs)

Source code in hypha/apply/activity/adapters/base.py

def process_send(
    self,
    message_type,
    recipients,
    events,
    request,
    user,
    source,
    sources=None,
    related=None,
    **kwargs,
):
    if sources is None:
        sources = []
    try:
        # If this was a batch action we want to pull out the submission
        source = sources[0]
    except IndexError:
        pass

    kwargs = {
        "request": request,
        "user": user,
        "source": source,
        "sources": sources,
        "related": related,
        **kwargs,
    }
    kwargs.update(self.get_neat_related(message_type, related))
    kwargs.update(self.extra_kwargs(message_type, **kwargs))

    for recipient in recipients:
        # Allow for customization of message based on recipient string (will vary based on adapter)
        message_kwargs = {**kwargs, "recipient": recipient}
        message = self.message(message_type, **message_kwargs)
        if not message:
            continue

        message_logs = self.create_logs(message, recipient, *events)

        if settings.SEND_MESSAGES or self.always_send:
            status = self.send_message(
                message, recipient=recipient, logs=message_logs, **kwargs
            )
        else:
            status = "Message not sent as SEND_MESSAGES==FALSE"

        message_logs.update_status(status)

        if not settings.SEND_MESSAGES:
            if recipient:
                debug_message = "{} [to: {}]: {}".format(
                    self.adapter_type, recipient, message
                )
            else:
                debug_message = "{}: {}".format(self.adapter_type, message)
            messages.add_message(request, messages.DEBUG, debug_message)

create_logs

create_logs(message, recipient, *events)

Source code in hypha/apply/activity/adapters/base.py

def create_logs(self, message, recipient, *events):
    from ..models import Message

    messages = Message.objects.bulk_create(
        Message(**self.log_kwargs(message, recipient, event)) for event in events
    )
    return Message.objects.filter(id__in=[message.id for message in messages])

log_kwargs

log_kwargs(message, recipient, event)

Source code in hypha/apply/activity/adapters/base.py

def log_kwargs(self, message, recipient, event):
    return {
        "type": self.adapter_type,
        "content": message,
        "recipient": recipient or "",
        "event": event,
    }

batch_reviewers_updated

batch_reviewers_updated(added, sources, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def batch_reviewers_updated(self, added, sources, **kwargs):
    reviewers_text = " ".join(
        [
            _("{user} as {name},").format(user=str(user), name=role.name)
            for role, user in added
            if user
        ]
    )

    return _("Batch reviewers added: {reviewers_text} to ").format(
        reviewers_text=reviewers_text
    ) + ", ".join(
        ['"{title}"'.format(title=source.title_text_display) for source in sources]
    )

handle_report_frequency

handle_report_frequency(config, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def handle_report_frequency(self, config, **kwargs):
    new_schedule = config.get_frequency_display()
    return _(
        "Successfully updated reporting frequency. They will now report {new_schedule} starting on {schedule_start}"
    ).format(new_schedule=new_schedule, schedule_start=config.schedule_start)

handle_skipped_report

handle_skipped_report(report, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def handle_skipped_report(self, report, **kwargs):
    if report.skipped:
        return _(
            "Successfully skipped a Report for {start_date} to {end_date}"
        ).format(start_date=report.start_date, end_date=report.end_date)
    else:
        return _(
            "Successfully unskipped a Report for {start_date} to {end_date}"
        ).format(start_date=report.start_date, end_date=report.end_date)

batch_transition

batch_transition(sources, transitions, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def batch_transition(self, sources, transitions, **kwargs):
    base_message = "Successfully updated:"
    transition = "{submission} [{old_display} → {new_display}]."
    transition_messages = [
        transition.format(
            submission=submission.title_text_display,
            old_display=transitions[submission.id],
            new_display=submission.phase,
        )
        for submission in sources
    ]
    messages = [base_message, *transition_messages]
    return " ".join(messages)

batch_determinations

batch_determinations(sources, determinations, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def batch_determinations(self, sources, determinations, **kwargs):
    submissions = sources
    outcome = determinations[submissions[0].id].clean_outcome

    base_message = _("Successfully determined as {outcome}: ").format(
        outcome=outcome
    )
    submissions_text = [submission.title_text_display for submission in submissions]
    return base_message + ", ".join(submissions_text)

recipients

recipients(*args, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def recipients(self, *args, **kwargs):
    return [None]

batch_recipients

batch_recipients(message_type, sources, *args, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def batch_recipients(self, message_type, sources, *args, **kwargs):
    return [
        {
            "recipients": [None],
            "sources": sources,
        }
    ]

send_message

send_message(message, request, **kwargs)

Source code in hypha/apply/activity/adapters/django_messages.py

def send_message(self, message, request, **kwargs):
    messages.add_message(request, messages.INFO, message)