INNER CODE UNIT · Python
deliver_notifications
skyportal/skyportal · services/assistant/app.py:542
def deliver_notifications(session, author_id, notify, context_type, context_id, answer):
"""Notify the recipients a task run named, with its answer.
Groups expand to their members; a named user must share a group with the
author, so a run cannot notify people outside the author's groups.
"""
if not isinstance(notify, dict):
return
wanted = {int(u) for u in (notify.get("users") or [])}
for gid in notify.get("groups") or []:
wanted.update(
session.scalars(
sa.select(GroupUser.user_id).where(GroupUser.group_id == int(gid))
).all()
)
wanted.discard(author_id)
if not wanted:
return