Email notification

Reimplementation of the core CKAN email notifications.

This module reimplements the core CKAN email notifications in order to:

• Be able to provide more visibility onto what is going on. This includes returning the number of emails being sent.

• Modify the default implementation in order to not require an active request.

_notifications_for_activities

_notifications_for_activities(activities, user_dict)

Return one or more email notifications covering the given activities.

This function handles grouping multiple activities into a single digest email.

Parameters:

Name	Type	Description	Default
activities	list of activity dicts like those returned by ckan.logic.action.get.dashboard_activity_list()	the activities to consider	required

Returns:

Type	Description
list of dicts each with keys 'subject' and 'body'	a list of email notifications

Source code in ckanext/saeoss/email_notifications.py

def _notifications_for_activities(activities, user_dict):
    """Return one or more email notifications covering the given activities.

    This function handles grouping multiple activities into a single digest
    email.

    :param activities: the activities to consider
    :type activities: list of activity dicts like those returned by
        ckan.logic.action.get.dashboard_activity_list()

    :returns: a list of email notifications
    :rtype: list of dicts each with keys 'subject' and 'body'

    """
    if not activities:
        return []

    if not user_dict.get("activity_streams_email_notifications"):
        return []

    # We just group all activities into a single "new activity" email that
    # doesn't say anything about _what_ new activities they are.
    # TODO: Here we could generate some smarter content for the emails e.g.
    # say something about the contents of the activities, or single out
    # certain types of activity to be sent in their own individual emails,
    # etc.
    subject = toolkit.ungettext(
        "{n} new activity from {site_title}",
        "{n} new activities from {site_title}",
        len(activities),
    ).format(site_title=toolkit.config.get("ckan.site_title"), n=len(activities))
    jinja_env = get_jinja_env()
    body_template = jinja_env.get_template("email_notifications/email_body.txt")
    rendered_body = body_template.render(
        activities=activities,
        site_url=toolkit.config.get("ckan.site_url"),
        site_title=toolkit.config.get("ckan.site_title"),
    )
    notifications = [{"subject": subject, "body": rendered_body}]

    return notifications

_notifications_from_dashboard_activity_list

_notifications_from_dashboard_activity_list(user_dict, since)

Return any email notifications from the given user's dashboard activity list since since.

Source code in ckanext/saeoss/email_notifications.py

def _notifications_from_dashboard_activity_list(user_dict, since):
    """Return any email notifications from the given user's dashboard activity
    list since `since`.

    """
    # Get the user's dashboard activity stream.
    context = {"model": model, "session": model.Session, "user": user_dict["id"]}
    activities = logic.get_action("dashboard_activity_list")(context, {})

    # Filter out the user's own activities., so they don't get an email every
    # time they themselves do something (we are not Trac).
    activities = [act for act in activities if act["user_id"] != user_dict["id"]]

    # Filter out the old activities.
    strptime = dt.datetime.strptime
    fmt = "%Y-%m-%dT%H:%M:%S.%f"
    activities = [act for act in activities if strptime(act["timestamp"], fmt) > since]
    return _notifications_for_activities(activities, user_dict)

get_notifications

get_notifications(user_dict, since)

Return any email notifications for the given user since since.

For example email notifications about activity streams will be returned for any activities the occurred since since.

Parameters:

Name	Type	Description	Default
user_dict	dictionary	a dictionary representing the user, should contain 'id' and 'name'	required
since		datetime after which to return notifications from	required

Returns:

Type	Description
list of dicts with keys 'subject' and 'body'	a list of email notifications

Source code in ckanext/saeoss/email_notifications.py

def get_notifications(user_dict, since):
    """Return any email notifications for the given user since `since`.

    For example email notifications about activity streams will be returned for
    any activities the occurred since `since`.

    :param user_dict: a dictionary representing the user, should contain 'id'
        and 'name'
    :type user_dict: dictionary

    :param since: datetime after which to return notifications from
    :rtype since: datetime.datetime

    :returns: a list of email notifications
    :rtype: list of dicts with keys 'subject' and 'body'

    """
    notifications = []
    for function in _notifications_functions:
        notifications.extend(function(user_dict, since))
    return notifications

send_notification

send_notification(user, email_dict)

Email email_dict to user.

Source code in ckanext/saeoss/email_notifications.py

def send_notification(user, email_dict):
    """Email `email_dict` to `user`."""
    import ckan.lib.mailer

    if not user.get("email"):
        # FIXME: Raise an exception.
        return

    try:
        ckan.lib.mailer.mail_recipient(
            user["display_name"],
            user["email"],
            email_dict["subject"],
            email_dict["body"],
        )
    except ckan.lib.mailer.MailerException:
        logger.error(ckan.lib.mailer.MailerException)
        raise
    else:
        logger.debug(f"Email sent!")

string_to_timedelta

string_to_timedelta(s)

Parse a string s and return a standard datetime.timedelta object.

Handles days, hours, minutes, seconds, and microseconds.

Accepts strings in these formats:

2 days 14 days 4:35:00 (hours, minutes and seconds) 4:35:12.087465 (hours, minutes, seconds and microseconds) 7 days, 3:23:34 7 days, 3:23:34.087465 .087465 (microseconds only)

Raises:

Type	Description
ckan.logic.ValidationError	if the given string does not match any of the recognised formats

Source code in ckanext/saeoss/email_notifications.py

def string_to_timedelta(s):
    """Parse a string s and return a standard datetime.timedelta object.

    Handles days, hours, minutes, seconds, and microseconds.

    Accepts strings in these formats:

    2 days
    14 days
    4:35:00 (hours, minutes and seconds)
    4:35:12.087465 (hours, minutes, seconds and microseconds)
    7 days, 3:23:34
    7 days, 3:23:34.087465
    .087465 (microseconds only)

    :raises ckan.logic.ValidationError: if the given string does not match any
        of the recognised formats

    """
    patterns = []
    days_only_pattern = "(?P<days>\d+)\s+day(s)?"
    patterns.append(days_only_pattern)
    hms_only_pattern = "(?P<hours>\d?\d):(?P<minutes>\d\d):(?P<seconds>\d\d)"
    patterns.append(hms_only_pattern)
    ms_only_pattern = ".(?P<milliseconds>\d\d\d)(?P<microseconds>\d\d\d)"
    patterns.append(ms_only_pattern)
    hms_and_ms_pattern = hms_only_pattern + ms_only_pattern
    patterns.append(hms_and_ms_pattern)
    days_and_hms_pattern = "{0},\s+{1}".format(days_only_pattern, hms_only_pattern)
    patterns.append(days_and_hms_pattern)
    days_and_hms_and_ms_pattern = days_and_hms_pattern + ms_only_pattern
    patterns.append(days_and_hms_and_ms_pattern)

    for pattern in patterns:
        match = re.match("^{0}$".format(pattern), s)
        if match:
            break

    if not match:
        raise logic.ValidationError("Not a valid time: {0}".format(s))

    gd = match.groupdict()
    days = int(gd.get("days", "0"))
    hours = int(gd.get("hours", "0"))
    minutes = int(gd.get("minutes", "0"))
    seconds = int(gd.get("seconds", "0"))
    milliseconds = int(gd.get("milliseconds", "0"))
    microseconds = int(gd.get("microseconds", "0"))
    delta = dt.timedelta(
        days=days,
        hours=hours,
        minutes=minutes,
        seconds=seconds,
        milliseconds=milliseconds,
        microseconds=microseconds,
    )
    return delta