opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #17914 from davelopez/add_email_notifications_channel

Browse Source 
Add `email` notifications channel
This commit is contained in:
John Davis
2024-04-16 14:12:05 -04:00
committed by GitHub
parent 0a68b5df25 23164deced
commit b60bbbd24a
34 changed files with 1031 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/api/notifications.preferences.ts
+11 -4
View File
@@ -1,11 +1,18 @@
import { type components, fetcher } from "@/api/schema";
export type UserNotificationPreferences = components["schemas"]["UserNotificationPreferences"];
type UserNotificationPreferences = components["schemas"]["UserNotificationPreferences"];
export interface UserNotificationPreferencesExtended extends UserNotificationPreferences {
supportedChannels: string[];
}
const getNotificationsPreferences = fetcher.path("/api/notifications/preferences").method("get").create();
export async function getNotificationsPreferencesFromServer() {
const { data } = await getNotificationsPreferences({});
return data;
export async function getNotificationsPreferencesFromServer(): Promise<UserNotificationPreferencesExtended> {
const { data, headers } = await getNotificationsPreferences({});
return {
...data,
supportedChannels: headers.get("supported-channels")?.split(",") ?? [],
};
}
type UpdateUserNotificationPreferencesRequest = components["schemas"]["UpdateUserNotificationPreferencesRequest"];
client/src/api/notifications.ts
+1 -1
View File
@@ -25,7 +25,7 @@ export type NewSharedItemNotificationContentItemType =
type UserNotificationUpdateRequest = components["schemas"]["UserNotificationUpdateRequest"];
type NotificationCreateRequest = components["schemas"]["NotificationCreateRequest"];
export type NotificationCreateRequest = components["schemas"]["NotificationCreateRequest"];
type NotificationResponse = components["schemas"]["NotificationResponse"];
client/src/api/schema/schema.ts
+20 -11
View File
@@ -1255,6 +1255,9 @@ export interface paths {
/**
* Returns the current user's preferences for notifications.
* @description Anonymous users cannot have notification preferences. They will receive only broadcasted notifications.
*
* - The settings will contain all possible channels, but the client should only show the ones that are really supported by the server.
* The supported channels are returned in the `supported-channels` header.
*/
get: operations["get_notification_preferences_api_notifications_preferences_get"];
/**
@@ -9564,6 +9567,7 @@ export interface components {
* Channels
* @description The channels that the user wants to receive notifications from for this category.
* @default {
* "email": true,
* "push": true
* }
*/
@@ -9580,6 +9584,12 @@ export interface components {
* @description The settings for each channel of a notification category.
*/
NotificationChannelSettings: {
/**
* Email
* @description Whether the user wants to receive email notifications for this category. This setting will be ignored unless the server supports asynchronous tasks.
* @default true
*/
email?: boolean;
/**
* Push
* @description Whether the user wants to receive push notifications in the browser for this category.
@@ -9628,10 +9638,7 @@ export interface components {
*/
variant: components["schemas"]["NotificationVariant"];
};
/**
* NotificationCreateRequest
* @description Contains the recipients and the notification to create.
*/
/** NotificationCreateRequest */
NotificationCreateRequest: {
/**
* Notification
@@ -9642,7 +9649,7 @@ export interface components {
* Recipients
* @description The recipients of the notification. Can be a combination of users, groups and roles.
*/
recipients: components["schemas"]["NotificationRecipients"];
recipients: components["schemas"]["NotificationRecipientsRequest"];
};
/** NotificationCreatedResponse */
NotificationCreatedResponse: {
@@ -9657,11 +9664,8 @@ export interface components {
*/
total_notifications_sent: number;
};
/**
* NotificationRecipients
* @description The recipients of a notification. Can be a combination of users, groups and roles.
*/
NotificationRecipients: {
/** NotificationRecipientsRequest */
NotificationRecipientsRequest: {
/**
* Group IDs
* @description The list of encoded group IDs of the groups that should receive the notification.
@@ -19983,7 +19987,9 @@ export interface operations {
/** @description Successful Response */
200: {
content: {
"application/json": components["schemas"]["NotificationCreatedResponse"];
"application/json":
| components["schemas"]["NotificationCreatedResponse"]
| components["schemas"]["AsyncTaskResultSummary"];
};
};
/** @description Validation Error */
@@ -20155,6 +20161,9 @@ export interface operations {
/**
* Returns the current user's preferences for notifications.
* @description Anonymous users cannot have notification preferences. They will receive only broadcasted notifications.
*
* - The settings will contain all possible channels, but the client should only show the ones that are really supported by the server.
* The supported channels are returned in the `supported-channels` header.
*/
parameters?: {
/** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
client/src/components/Notifications/NotificationsList.vue
+9 -1
View File
@@ -19,9 +19,17 @@ library.add(faCog, faHourglassHalf, faRetweet);
const notificationsStore = useNotificationsStore();
const { notifications, loadingNotifications } = storeToRefs(notificationsStore);
interface Props {
shouldOpenPreferences?: boolean;
}
const props = withDefaults(defineProps<Props>(), {
shouldOpenPreferences: false,
});
const showUnread = ref(false);
const showShared = ref(false);
const preferencesOpen = ref(false);
const preferencesOpen = ref(props.shouldOpenPreferences);
const selectedNotificationIds = ref<string[]>([]);
const haveSelected = computed(() => selectedNotificationIds.value.length > 0);
client/src/components/User/Notifications/NotificationsPreferences.vue
+6 -7
View File
@@ -8,7 +8,7 @@ import { computed, ref, watch } from "vue";
import {
getNotificationsPreferencesFromServer,
updateNotificationsPreferencesOnServer,
UserNotificationPreferences,
UserNotificationPreferencesExtended,
} from "@/api/notifications.preferences";
import { useConfig } from "@/composables/config";
import { Toast } from "@/composables/toast";
@@ -39,14 +39,15 @@ const { config } = useConfig(true);
const loading = ref(false);
const errorMessage = ref<string | null>(null);
const pushNotificationsGranted = ref(pushNotificationsEnabled());
const notificationsPreferences = ref<UserNotificationPreferences["preferences"]>({});
const notificationsPreferences = ref<UserNotificationPreferencesExtended["preferences"]>({});
const supportedChannels = ref<string[]>([]);
const categories = computed(() => Object.keys(notificationsPreferences.value));
const showPreferences = computed(() => {
return !loading.value && config.value.enable_notification_system && notificationsPreferences.value;
});
const categoryDescriptionMap = {
const categoryDescriptionMap: Record<string, string> = {
message: "You will receive notifications when someone sends you a message.",
new_shared_item: "You will receive notifications when someone shares an item with you.",
};
@@ -55,6 +56,7 @@ async function getNotificationsPreferences() {
loading.value = true;
await getNotificationsPreferencesFromServer()
.then((data) => {
supportedChannels.value = data.supportedChannels;
notificationsPreferences.value = data.preferences;
})
.catch((error: any) => {
@@ -148,10 +150,7 @@ watch(
switch />
</div>
<div
v-for="channel in Object.keys(notificationsPreferences[category].channels)"
:key="channel"
class="category-channel">
<div v-for="channel in supportedChannels" :key="channel" class="category-channel">
<BFormCheckbox
v-model="notificationsPreferences[category].channels[channel]"
v-localize
client/src/components/admin/Notifications/NotificationForm.vue
+1 -2
View File
@@ -7,7 +7,7 @@ import { computed, type Ref, ref } from "vue";
import { useRouter } from "vue-router/composables";
import { getAllGroups } from "@/api/groups";
import { sendNotification } from "@/api/notifications";
import { NotificationCreateRequest, sendNotification } from "@/api/notifications";
import { getAllRoles } from "@/api/roles";
import { type components } from "@/api/schema";
import { getAllUsers } from "@/api/users";
@@ -25,7 +25,6 @@ library.add(faInfoCircle);
type SelectOption = [string, string];
type NotificationCreateData = components["schemas"]["NotificationCreateData"];
type NotificationCreateRequest = components["schemas"]["NotificationCreateRequest"];
interface MessageNotificationCreateData extends NotificationCreateData {
category: "message";
client/src/entry/analysis/router.js
+3
View File
@@ -450,6 +450,9 @@ export function getRouter(Galaxy) {
path: "user/notifications",
component: NotificationsList,
redirect: redirectIf(!Galaxy.config.enable_notification_system, "/") || redirectAnon(),
props: (route) => ({
shouldOpenPreferences: Boolean(route.query.preferences),
}),
},
{
path: "user/notifications/preferences",
doc/source/admin/galaxy_options.rst
+11
View File
@@ -5463,6 +5463,17 @@
:Type: int
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
``dispatch_notifications_interval``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
:Description:
The interval in seconds between attempts to dispatch notifications
to users (every 10 minutes by default). Runs in a Celery task.
:Default: ``600``
:Type: int
~~~~~~~~~~~~~~~~~~~~~~
``help_forum_api_url``
~~~~~~~~~~~~~~~~~~~~~~
lib/galaxy/celery/__init__.py
+4 -1
View File
@@ -238,7 +238,10 @@ def setup_periodic_tasks(config, celery_app):
beat_schedule: Dict[str, Dict[str, Any]] = {}
schedule_task("prune_history_audit_table", config.history_audit_table_prune_interval)
schedule_task("cleanup_short_term_storage", config.short_term_storage_cleanup_interval)
schedule_task("cleanup_expired_notifications", config.expired_notifications_cleanup_interval)
if config.enable_notification_system:
schedule_task("cleanup_expired_notifications", config.expired_notifications_cleanup_interval)
schedule_task("dispatch_pending_notifications", config.dispatch_notifications_interval)
if config.object_store_cache_monitor_driver in ["auto", "celery"]:
schedule_task("clean_object_store_caches", config.object_store_cache_monitor_interval)
lib/galaxy/celery/tasks.py
+19
View File
@@ -43,6 +43,7 @@ from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.objectstore import BaseObjectStore
from galaxy.objectstore.caching import check_caches
from galaxy.queue_worker import GalaxyQueueWorker
from galaxy.schema.notifications import NotificationCreateRequest
from galaxy.schema.tasks import (
ComputeDatasetHashTaskRequest,
GenerateHistoryContentDownload,
@@ -483,3 +484,21 @@ def cleanup_expired_notifications(notification_manager: NotificationManager):
@galaxy_task(action="prune object store cache directories")
def clean_object_store_caches(object_store: BaseObjectStore):
check_caches(object_store.cache_targets())
@galaxy_task(action="send notifications to all recipients")
def send_notification_to_recipients_async(
request: NotificationCreateRequest, notification_manager: NotificationManager
):
"""Send a notification to a list of users."""
_, notifications_sent = notification_manager.send_notification_to_recipients(request=request)
log.info(f"Successfully sent {notifications_sent} notifications.")
@galaxy_task(action="dispatch pending notifications")
def dispatch_pending_notifications(notification_manager: NotificationManager):
"""Dispatch pending notifications."""
count = notification_manager.dispatch_pending_notifications_via_channels()
if count:
log.info(f"Successfully dispatched {count} notifications.")
lib/galaxy/config/sample/galaxy.yml.sample
+4
View File
@@ -2911,6 +2911,10 @@ galaxy:
# a Celery task.
#expired_notifications_cleanup_interval: 86400
# The interval in seconds between attempts to dispatch notifications
# to users (every 10 minutes by default). Runs in a Celery task.
#dispatch_notifications_interval: 600
# The URL pointing to the Galaxy Help Forum API base URL. The API must
# be compatible with Discourse API (https://docs.discourse.org/).
#help_forum_api_url: https://help.galaxyproject.org/
lib/galaxy/config/schemas/config_schema.yml
+7
View File
@@ -3990,6 +3990,13 @@ mapping:
desc: |
The interval in seconds between attempts to delete all expired notifications from the database (every 24 hours by default). Runs in a Celery task.
dispatch_notifications_interval:
type: int
required: false
default: 600
desc: |
The interval in seconds between attempts to dispatch notifications to users (every 10 minutes by default). Runs in a Celery task.
help_forum_api_url:
type: str
required: false
lib/galaxy/config/templates/mail/notifications/message-email.html
+84
View File
@@ -0,0 +1,84 @@
Use this template to customize the HTML-formatted email your users will receive
when a new notification of category "message" is sent to them.
Copy the file to {{ templates_dir }}/mail/notifications/message-email.html and modify as required.
If you are adding URLs, remember that only absolute URLs (with
a domain name) make sense in email! They can be served from any stable
location, including your Galaxy server or GitHub.
The following variables are available for inserting into the HTML with Jinja2
syntax, like {{ variable_name }}. They will be rendered into the text before
the email is sent:
- name The user's name
- user_email The user's email
- date Date and time of the notification
- hostname Your galaxy's hostname (i.e. usegalaxy.* or the value in `server_name` from the galaxy config file)
- contact_email Your galaxy's contact email
- notification_settings_url The URL to the user's notification settings to manage their subscriptions
- content The message payload
- subject The message subject
- content The message content in HTML (converted from Markdown)
- galaxy_url The URL to the Galaxy instance (i.e. https://usegalaxy.*)
Template begins here >>>>>>
<!DOCTYPE html>
<html lang="en" dir="ltr">
<head>
<meta charset="utf-8">
<title>[Galaxy] New message received: {{ content['subject'] }}</title>
<link href="https://fonts.googleapis.com/css2?family=Roboto&display=swap" rel="stylesheet">
</head>
<body style="font-family: 'Roboto', sans-serif;">
<p style="font-size: 12pt;">
Hello {{ name }},<br><br>
You have received a new message on <b>{{ date }}</b> from the Galaxy Team at <b>{{ hostname }}</b>, here are the details:
<br><br>
</p>
<p style="font-size: 12pt;">
<strong>Subject:</strong>
<br>
{{ content['subject'] }}
<br><br>
<strong>Message:</strong>
<br>
{{ content['message'] }}
<br><br>
</p>
<p style="font-size: 12pt;">
Thank you for using Galaxy!
</p>
<p style="font-size: 12pt;">
Regards,<br>
Your Galaxy Team at <a href="{{ galaxy_url }}">{{ hostname }}</a>
</p>
<p style="font-size: 10pt;">
You received this email because you are subscribed to receive notifications from the Galaxy Team.
{% if notification_settings_url %}
You can manage your notification settings <a href="{{ notification_settings_url }}">here</a>.
{% endif %}
<br>
{% if contact_email %}
This is an automated email. If you have any questions or concerns, please do not reply to this email, instead, contact us at <a href="mailto:{{ contact_email }}">{{ contact_email }}</a>.
{% endif %}
</p>
<img
style="width: 130px; height: auto; margin: 15px 0;"
src="https://galaxyproject.org/images/galaxy-logos/galaxy_project_logo_square.png"
alt="Galaxy project logo"
>
<br>
</body>
</html>
lib/galaxy/config/templates/mail/notifications/message-email.txt
+49
View File
@@ -0,0 +1,49 @@
Use this template to customize the text email your users will receive
when a new notification of category "message" is sent to them.
Copy the file to {{ templates_dir }}/mail/notifications/message-email.txt and modify as required.
If you are adding URLs, remember that only absolute URLs (with
a domain name) make sense in email! They can be served from any stable
location, including your Galaxy server or GitHub.
The following variables are available for inserting into the text like
{{ variable_name }}. They will be rendered into the text before the email is
sent:
- name The user's name
- user_email The user's email
- date Date and time of the notification
- hostname Your galaxy's hostname (i.e. usegalaxy.* or the value in `server_name` from the galaxy config file)
- contact_email Your galaxy's contact email
- notification_settings_url The URL to the user's notification settings to manage their subscriptions
- content The message payload
- subject The message subject
- content The message content in HTML (converted from Markdown)
- galaxy_url The URL to the Galaxy instance (i.e. https://usegalaxy.*)
Template begins here>>>>>>
Hello {{ name }},
You have received a new message on {{ date }} from the Galaxy Team at {{ hostname }}, here are the details:
Subject:
{{ content['subject'] }}
Message:
{{ content['message'] }}
Thank you for using Galaxy!
Regards,
Your Galaxy Team at {{ hostname }}
---
You received this email because you are subscribed to receive notifications from the Galaxy Team.
{% if notification_settings_url %}
To manage your notification settings, please visit {{ notification_settings_url }}.
{% endif %}
{% if contact_email %}
This is an automated email. If you have any questions or concerns, please do not reply to this email, instead, contact us at {{ contact_email }}.
{% endif %}
lib/galaxy/config/templates/mail/notifications/new_shared_item-email.html
+80
View File
@@ -0,0 +1,80 @@
Use this template to customize the HTML email your users will receive
when a new notification of category "new_shared_item" is sent to them.
Copy the file to {{ templates_dir }}/mail/notifications/new_shared_item-email.html and modify as required.
If you are adding URLs, remember that only absolute URLs (with
a domain name) make sense in email! They can be served from any stable
location, including your Galaxy server or GitHub.
The following variables are available for inserting into the HTML with Jinja2
syntax, like {{ variable_name }}. They will be rendered into the text before
the email is sent:
- name The user's name
- user_email The user's email
- date Date and time of the notification
- hostname Your galaxy's hostname (i.e. usegalaxy.* or the value in `server_name` from the galaxy config file)
- contact_email Your galaxy's contact email
- notification_settings_url The URL to the user's notification settings to manage their subscriptions
- content The new_shared_item payload
- item_type The type of the shared item
- item_name The name of the shared item
- owner_name The name of the owner of the shared item
- slug The slug of the shared item. Used for the link to the item.
- galaxy_url The URL to the Galaxy instance (i.e. https://usegalaxy.*)
Template begins here >>>>>>
<!DOCTYPE html>
<html lang="en" dir="ltr">
<head>
<meta charset="utf-8">
<title>[Galaxy] New {{ content['item_type'] }} shared with you: </title>
<link href="https://fonts.googleapis.com/css2?family=Roboto&display=swap" rel="stylesheet">
</head>
<body style="font-family: 'Roboto', sans-serif;">
<p style="font-size: 12pt;">
Hello {{ name }},<br><br>
A new <b>{{ content['item_type'] }}</b> named <i>{{ content['item_name'] }}</i> has been shared with you on <b>{{ date }}</b> by <b>{{ content['owner_name'] }}</b>.
</p>
<p style="font-size: 12pt;">
To access the shared {{ content['item_type'] }}, please visit the following link:
<br>
<a href="{{galaxy_url}}/{{ content['slug'] }}">{{ content['item_name'] }}</a>
<br><br>
<p style="font-size: 12pt;">
Thank you for using Galaxy!
</p>
<p style="font-size: 12pt;">
Regards,<br>
Your Galaxy Team at <a href="{{ galaxy_url }}">{{ hostname }}</a>
</p>
<p style="font-size: 10pt;">
You received this email because you are subscribed to receive notifications when another user shares an item with you.
{% if notification_settings_url %}
You can manage your notification settings <a href="{{ notification_settings_url }}">here</a>.
{% endif %}
<br>
{% if contact_email %}
This is an automated email. If you have any questions or concerns, please do not reply to this email, instead, contact us at <a href="mailto:{{ contact_email }}">{{ contact_email }}</a>.
{% endif %}
</p>
<img
style="width: 130px; height: auto; margin: 15px 0;"
src="https://galaxyproject.org/images/galaxy-logos/galaxy_project_logo_square.png"
alt="Galaxy project logo"
>
<br>
</body>
</html>
lib/galaxy/config/templates/mail/notifications/new_shared_item-email.txt
+50
View File
@@ -0,0 +1,50 @@
Use this template to customize the text email your users will receive
when a new notification of category "new_shared_item" is sent to them.
Copy the file to {{ templates_dir }}/mail/notifications/new_shared_item-email.txt and modify as required.
If you are adding URLs, remember that only absolute URLs (with
a domain name) make sense in email! They can be served from any stable
location, including your Galaxy server or GitHub.
The following variables are available for inserting into the text like
{{ variable_name }}. They will be rendered into the text before the email is
sent:
- name The user's name
- user_email The user's email
- date Date and time of the notification
- hostname Your galaxy's hostname (i.e. usegalaxy.* or the value in `server_name` from the galaxy config file)
- contact_email Your galaxy's contact email
- notification_settings_url The URL to the user's notification settings to manage their subscriptions
- content The new_shared_item payload
- item_type The type of the shared item
- item_name The name of the shared item
- owner_name The name of the owner of the shared item
- slug The slug of the shared item. Used for the link to the item.
- galaxy_url The URL to the Galaxy instance (i.e. https://usegalaxy.*)
Template begins here>>>>>>
Hello {{ name }},
A new {{ content['item_type'] }} has been shared with you on {{ date }} by {{ content['owner_name'] }}.
To access the shared {{ content['item_type'] }}, please visit the following link:
{{ galaxy_url }}/{{ content['slug'] }}
Thank you for using Galaxy!
Regards,
Your Galaxy Team at {{ hostname }}
---
You received this email because you are subscribed to receive notifications when another user shares an item with you.
{% if notification_settings_url %}
To manage your notification settings, please visit {{ notification_settings_url }}.
{% endif %}
{% if contact_email %}
This is an automated email. If you have any questions or concerns, please do not reply to this email, instead, contact us at {{ contact_email }}.
{% endif %}
lib/galaxy/managers/notification.py
+314 -21
View File
@@ -1,13 +1,22 @@
import logging
from datetime import datetime
from enum import Enum
from typing import (
cast,
Dict,
List,
NamedTuple,
Optional,
Set,
Tuple,
Type,
)
from urllib.parse import urlparse
from pydantic import ValidationError
from pydantic import (
BaseModel,
ValidationError,
)
from sqlalchemy import (
and_,
delete,
@@ -22,11 +31,15 @@ from sqlalchemy.orm import InstrumentedAttribute
from sqlalchemy.sql import Select
from typing_extensions import Protocol
from galaxy.config import GalaxyAppConfiguration
from galaxy.config import (
GalaxyAppConfiguration,
templates,
)
from galaxy.exceptions import (
ConfigDoesNotAllowException,
ObjectNotFound,
)
from galaxy.managers.markdown_util import to_html
from galaxy.model import (
GroupRoleAssociation,
Notification,
@@ -38,9 +51,14 @@ from galaxy.model import (
from galaxy.model.base import transaction
from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.schema.notifications import (
AnyNotificationContent,
BroadcastNotificationCreateRequest,
MandatoryNotificationCategory,
MessageNotificationContent,
NewSharedItemNotificationContent,
NotificationBroadcastUpdateRequest,
NotificationCategorySettings,
NotificationChannelSettings,
NotificationCreateData,
NotificationCreateRequest,
NotificationRecipients,
@@ -49,6 +67,13 @@ from galaxy.schema.notifications import (
UserNotificationPreferences,
UserNotificationUpdateRequest,
)
from galaxy.util import (
send_mail,
unicodify,
)
log = logging.getLogger(__name__)
NOTIFICATION_PREFERENCES_SECTION_NAME = "notifications"
@@ -58,6 +83,21 @@ class CleanupResultSummary(NamedTuple):
deleted_associations_count: int
class NotificationRecipientResolverStrategy(Protocol):
def resolve_users(self, recipients: NotificationRecipients) -> List[User]:
pass
class NotificationChannelPlugin(Protocol):
config: GalaxyAppConfiguration
def __init__(self, config: GalaxyAppConfiguration):
self.config = config
def send(self, notification: Notification, user: User):
raise NotImplementedError
class NotificationManager:
"""Manager class to interact with the database models related with Notifications."""
@@ -89,6 +129,7 @@ class NotificationManager:
Notification.expiration_time,
Notification.content,
]
self.channel_plugins = self._register_supported_channels()
@property
def notifications_enabled(self):
@@ -110,31 +151,118 @@ class NotificationManager:
if not self.notifications_enabled:
raise ConfigDoesNotAllowException("Notifications are disabled in this Galaxy.")
@property
def can_send_notifications_async(self):
return self.config.enable_celery_tasks
def send_notification_to_recipients(self, request: NotificationCreateRequest) -> Tuple[Optional[Notification], int]:
"""
Creates a new notification and associates it with all the recipient users.
It takes into account the user's notification preferences to decide if the notification should be sent to them.
No other notification channel is used here, only the internal database associations are created.
"""
self.ensure_notifications_enabled()
recipient_users = self.recipient_resolver.resolve(request.recipients)
notifications_sent = len(recipient_users)
notification = self._create_notification_model(request.notification)
notification = self._create_notification_model(request.notification, request.galaxy_url)
self.sa_session.add(notification)
self._send_to_users(notification, recipient_users)
with transaction(self.sa_session):
self.sa_session.commit()
notifications_sent = self._create_associations(notification, recipient_users)
with transaction(self.sa_session):
self.sa_session.commit()
return notification, notifications_sent
def _send_to_users(self, notification: Notification, users: List[User]):
# TODO: Move this potentially expensive operation to a task?
def _create_associations(self, notification: Notification, users: List[User]) -> int:
success_count = 0
for user in users:
if self._user_is_subscribed_to_notification(user, notification.category): # type:ignore[arg-type]
user_notification_association = UserNotificationAssociation(user, notification)
self.sa_session.add(user_notification_association)
try:
if self._is_user_subscribed_to_category(user, notification.category): # type:ignore[arg-type]
user_notification_association = UserNotificationAssociation(user, notification)
self.sa_session.add(user_notification_association)
success_count += 1
except Exception as e:
log.error(f"Error sending notification to user {user.id}. Reason: {unicodify(e)}")
continue
return success_count
def _user_is_subscribed_to_notification(self, user: User, category: PersonalNotificationCategory) -> bool:
def dispatch_pending_notifications_via_channels(self) -> int:
"""
Dispatches all pending notifications to the users depending on the configured channels.
This is meant to be called periodically by a background task.
"""
self.ensure_notifications_enabled()
pending_notifications = self.get_pending_notifications()
# Mark all pending notifications as dispatched
for notification in pending_notifications:
notification.dispatched = True
with transaction(self.sa_session):
self.sa_session.commit()
# Do the actual dispatching
for notification in pending_notifications:
self._dispatch_notification_to_users(notification)
return len(pending_notifications)
def get_pending_notifications(self):
"""
Returns all pending notifications that have not been dispatched yet
but are due and ready to be sent to the users.
"""
stmt = select(Notification).where(Notification.dispatched == false(), self._notification_is_active)
return self.sa_session.execute(stmt).scalars().all()
def _dispatch_notification_to_users(self, notification: Notification):
users = self._get_associated_users(notification)
for user in users:
category_settings = self._get_user_category_settings(user, notification.category) # type:ignore[arg-type]
if not self._is_subscribed_to_category(category_settings):
continue
self._send_via_channels(notification, user, category_settings.channels)
def _get_associated_users(self, notification: Notification):
stmt = (
select(User)
.join(
UserNotificationAssociation,
UserNotificationAssociation.user_id == User.id,
)
.where(
UserNotificationAssociation.notification_id == notification.id,
)
)
return self.sa_session.execute(stmt).scalars().all()
def _is_user_subscribed_to_category(self, user: User, category: PersonalNotificationCategory) -> bool:
category_settings = self._get_user_category_settings(user, category)
return self._is_subscribed_to_category(category_settings)
def _send_via_channels(self, notification: Notification, user: User, channel_settings: NotificationChannelSettings):
channels = channel_settings.model_fields_set
for channel in channels:
if channel not in self.channel_plugins:
log.warning(f"Notification channel '{channel}' is not supported.")
continue
if getattr(channel_settings, channel, False) is False:
continue # User opted out of this channel
plugin = self.channel_plugins[channel]
plugin.send(notification, user)
def _is_subscribed_to_category(self, category_settings: NotificationCategorySettings) -> bool:
return category_settings.enabled
def _get_user_category_settings(
self, user: User, category: PersonalNotificationCategory
) -> NotificationCategorySettings:
notification_preferences = self.get_user_notification_preferences(user)
category_settings = notification_preferences.get(category)
return category_settings.enabled
return category_settings
def create_broadcast_notification(self, request: BroadcastNotificationCreateRequest):
"""Creates a broadcasted notification.
@@ -280,7 +408,7 @@ class NotificationManager:
else None
)
try:
return UserNotificationPreferences.parse_raw(current_notification_preferences)
return UserNotificationPreferences.model_validate_json(current_notification_preferences)
except ValidationError:
# Gracefully return default preferences is they don't exist or get corrupted
return UserNotificationPreferences.default()
@@ -289,12 +417,30 @@ class NotificationManager:
self, user: User, request: UpdateUserNotificationPreferencesRequest
) -> UserNotificationPreferences:
"""Updates the user's notification preferences with the requested changes."""
notification_preferences = self.get_user_notification_preferences(user)
notification_preferences.update(request.preferences)
user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME] = notification_preferences.model_dump_json()
preferences = self.get_user_notification_preferences(user)
preferences.update(request.preferences)
user.preferences[NOTIFICATION_PREFERENCES_SECTION_NAME] = preferences.model_dump_json()
with transaction(self.sa_session):
self.sa_session.commit()
return notification_preferences
return preferences
def _register_supported_channels(self) -> Dict[str, NotificationChannelPlugin]:
"""Registers the supported notification channels in this server."""
supported_channels: Dict[str, NotificationChannelPlugin] = {
# Push notifications are handled client-side so no real plugin is needed
"push": NoOpNotificationChannelPlugin(self.config),
}
if self.can_send_notifications_async:
# Most additional channels require asynchronous processing and will be
# handled by Celery tasks. Add their plugins here.
supported_channels["email"] = EmailNotificationChannelPlugin(self.config)
return supported_channels
def get_supported_channels(self) -> Set[str]:
"""Returns the set of supported notification channels in this server."""
return set(self.channel_plugins.keys())
def cleanup_expired_notifications(self) -> CleanupResultSummary:
"""
@@ -318,7 +464,9 @@ class NotificationManager:
return CleanupResultSummary(deleted_notifications_count, deleted_associations_count)
def _create_notification_model(self, payload: NotificationCreateData):
def _create_notification_model(
self, payload: NotificationCreateData, galaxy_url: Optional[str] = None
) -> Notification:
notification = Notification(
payload.source,
payload.category,
@@ -327,6 +475,7 @@ class NotificationManager:
)
notification.publication_time = payload.publication_time
notification.expiration_time = payload.expiration_time
notification.galaxy_url = galaxy_url
return notification
def _user_notifications_query(
@@ -372,9 +521,8 @@ class NotificationManager:
return stmt
class NotificationRecipientResolverStrategy(Protocol):
def resolve_users(self, recipients: NotificationRecipients) -> List[User]:
pass
# --------------------------------------
# Notification Recipients Resolver Implementations
class NotificationRecipientResolver:
@@ -480,3 +628,148 @@ class RecursiveCTEStrategy(NotificationRecipientResolverStrategy):
def resolve_users(self, recipients: NotificationRecipients) -> List[User]:
# TODO Implement resolver using recursive CTEs?
return []
# --------------------------------------
# Notification Channel Plugins Implementations
class NoOpNotificationChannelPlugin(NotificationChannelPlugin):
def send(self, notification: Notification, user: User):
pass
class TemplateFormats(str, Enum):
HTML = "html"
TXT = "txt"
class NotificationContext(BaseModel):
"""Information passed to the email template to render the body."""
name: str
user_email: str
date: str
hostname: str
contact_email: str
notification_settings_url: str
content: AnyNotificationContent
galaxy_url: Optional[str] = None
class EmailNotificationTemplateBuilder(Protocol):
config: GalaxyAppConfiguration
notification: Notification
user: User
def __init__(self, config: GalaxyAppConfiguration, notification: Notification, user: User):
self.config = config
self.notification = notification
self.user = user
def get_content(self, template_format: TemplateFormats) -> AnyNotificationContent:
"""Processes the notification content to be rendered in the email body.
This should be implemented by each concrete template builder.
"""
raise NotImplementedError
def get_subject(self) -> str:
"""Returns the subject of the email to be sent.
This should be implemented by each concrete template builder.
"""
raise NotImplementedError
def build_context(self, template_format: TemplateFormats) -> NotificationContext:
notification = self.notification
user = self.user
notification_date = notification.publication_time if notification.publication_time else notification.create_time
hostname = (
urlparse(self.notification.galaxy_url).hostname if self.notification.galaxy_url else self.config.server_name
)
notification_settings_url = (
f"{self.notification.galaxy_url}/user/notifications?preferences=true"
if self.notification.galaxy_url
else None
)
contact_email = self.config.error_email_to or None
return NotificationContext(
name=user.username,
user_email=user.email,
date=notification_date.strftime("%B %d, %Y"),
hostname=hostname,
contact_email=contact_email,
notification_settings_url=notification_settings_url,
content=self.get_content(template_format),
galaxy_url=self.notification.galaxy_url,
)
def get_body(self, template_format: TemplateFormats) -> str:
template_path = f"mail/notifications/{self.notification.category}-email.{template_format.value}"
context = self.build_context(template_format)
return templates.render(
template_path,
context.model_dump(),
self.config.templates_dir,
)
class MessageEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder):
markdown_to = {
TemplateFormats.HTML: to_html,
TemplateFormats.TXT: lambda x: x, # TODO: strip markdown?
}
def get_content(self, template_format: TemplateFormats) -> AnyNotificationContent:
content = MessageNotificationContent.model_construct(**self.notification.content) # type:ignore[arg-type]
content.message = self.markdown_to[template_format](content.message)
return content
def get_subject(self) -> str:
content = cast(MessageNotificationContent, self.get_content(TemplateFormats.TXT))
return f"[Galaxy] New message: {content.subject}"
class NewSharedItemEmailNotificationTemplateBuilder(EmailNotificationTemplateBuilder):
def get_content(self, template_format: TemplateFormats) -> AnyNotificationContent:
content = NewSharedItemNotificationContent.model_construct(**self.notification.content) # type:ignore[arg-type]
return content
def get_subject(self) -> str:
content = cast(NewSharedItemNotificationContent, self.get_content(TemplateFormats.TXT))
return f"[Galaxy] New {content.item_type} shared with you: {content.item_name}"
class EmailNotificationChannelPlugin(NotificationChannelPlugin):
# Register the supported email templates here
email_templates_by_category: Dict[PersonalNotificationCategory, Type[EmailNotificationTemplateBuilder]] = {
PersonalNotificationCategory.message: MessageEmailNotificationTemplateBuilder,
PersonalNotificationCategory.new_shared_item: NewSharedItemEmailNotificationTemplateBuilder,
}
def send(self, notification: Notification, user: User):
try:
category = cast(PersonalNotificationCategory, notification.category)
email_template_builder = self.email_templates_by_category.get(category)
if email_template_builder is None:
log.warning(f"No email template found for notification category '{notification.category}'.")
return
template_builder = email_template_builder(self.config, notification, user)
subject = template_builder.get_subject()
text_body = template_builder.get_body(TemplateFormats.TXT)
html_body = template_builder.get_body(TemplateFormats.HTML)
send_mail(
frm=self.config.email_from,
to=user.email,
subject=subject,
body=text_body,
config=self.config,
html=html_body,
)
except Exception as e:
log.error(f"Error sending email notification to user {user.id}. Reason: {unicodify(e)}")
pass
lib/galaxy/model/__init__.py
+13 -5
View File
@@ -2937,13 +2937,21 @@ class Notification(Base, Dictifiable, RepresentById):
expiration_time: Mapped[Optional[datetime]] = mapped_column(
default=now() + timedelta(days=30 * 6)
) # The expiration date, expired notifications will be permanently removed from DB regularly
source: Mapped[Optional[str]] = mapped_column(String(32), index=True) # Who (or what) generated the notification
category: Mapped[Optional[str]] = mapped_column(
String(64), index=True
source: Mapped[str] = mapped_column(
String(32), index=True, nullable=True
) # Who (or what) generated the notification
category: Mapped[str] = mapped_column(
String(64), index=True, nullable=True
) # Category of the notification, defines its contents. Used for filtering, un/subscribing, etc
variant: Mapped[Optional[str]] = mapped_column(
String(16), index=True
variant: Mapped[str] = mapped_column(
String(16), index=True, nullable=True
) # Defines the 'importance' of the notification ('info', 'warning', 'urgent', etc.). Used for filtering, highlight rendering, etc
dispatched: Mapped[bool] = mapped_column(
Boolean, index=True, default=False
) # Whether the notification has been dispatched to users via other channels
galaxy_url: Mapped[Optional[str]] = mapped_column(
String(255)
) # The URL to the Galaxy instance, used for generating links in the notification
# A bug in early 23.1 led to values being stored as json string, so we use this special type to process the result value twice.
# content should always be a dict
content: Mapped[Optional[bytes]] = mapped_column(DoubleEncodedJsonType)
lib/galaxy/model/migrations/alembic/versions_gxy/303a5583a030_add_dispatched_column_to_notifications.py
+55
View File
@@ -0,0 +1,55 @@
"""add dispatched column to notifications
Revision ID: 303a5583a030
Revises: 55f02fd8ab6c
Create Date: 2024-04-04 11:45:54.018829
"""
import sqlalchemy as sa
from alembic import op
from sqlalchemy import (
Boolean,
Column,
)
from galaxy.model.database_object_names import build_index_name
from galaxy.model.migrations.util import (
add_column,
drop_column,
drop_index,
transaction,
)
# revision identifiers, used by Alembic.
revision = "303a5583a030"
down_revision = "55f02fd8ab6c"
branch_labels = None
depends_on = None
# database object names used in this revision
table_name = "notification"
column_name = "dispatched"
publication_time_column_name = "publication_time"
index_name = build_index_name(table_name, column_name)
def upgrade():
with transaction():
add_column(
table_name,
Column(column_name, Boolean(), index=True, nullable=False, default=False, server_default=sa.false()),
)
# Set as already dispatched any notifications older than the current time to avoid sending them again
table = sa.sql.table(
table_name,
Column(column_name, Boolean()),
Column(publication_time_column_name, sa.DateTime()),
)
op.execute(table.update().where(table.c.publication_time <= sa.func.now()).values(dispatched=True))
def downgrade():
with transaction():
drop_index(index_name, table_name)
drop_column(table_name, column_name)
lib/galaxy/model/migrations/alembic/versions_gxy/5924fbf10430_add_galaxy_url_column_to_notifications.py
+36
View File
@@ -0,0 +1,36 @@
"""add galaxy_url column to notifications
Revision ID: 5924fbf10430
Revises: 303a5583a030
Create Date: 2024-04-11 09:56:26.200231
"""
from sqlalchemy import (
Column,
String,
)
from galaxy.model.migrations.util import (
add_column,
drop_column,
)
# revision identifiers, used by Alembic.
revision = "5924fbf10430"
down_revision = "303a5583a030"
branch_labels = None
depends_on = None
# database object names used in this revision
table_name = "notification"
column_name = "galaxy_url"
def upgrade():
add_column(table_name, Column(column_name, String(255)))
def downgrade():
drop_column(table_name, column_name)
lib/galaxy/schema/generics.py
+52
View File
@@ -0,0 +1,52 @@
from typing import (
Any,
Tuple,
Type,
TypeVar,
)
from pydantic import BaseModel
from pydantic.json_schema import GenerateJsonSchema
from galaxy.schema.fields import (
DecodedDatabaseIdField,
EncodedDatabaseIdField,
)
DatabaseIdT = TypeVar("DatabaseIdT")
ref_to_name = {}
class GenericModel(BaseModel):
@classmethod
def model_parametrized_name(cls, params: Tuple[Type[Any], ...]) -> str:
suffix = cls.__determine_suffix__(params)
class_name = cls.__name__.split("Generic", 1)[-1]
return f"{class_name}{suffix}"
@classmethod
def __get_pydantic_core_schema__(cls, *args, **kwargs):
result = super().__get_pydantic_core_schema__(*args, **kwargs)
ref_to_name[result["ref"]] = cls.__name__
return result
@classmethod
def __determine_suffix__(cls, params: Tuple[Type[Any], ...]) -> str:
suffix = "Incoming"
if params[0] is EncodedDatabaseIdField:
suffix = "Response"
elif params[0] is DecodedDatabaseIdField:
suffix = "Request"
return suffix
class CustomJsonSchema(GenerateJsonSchema):
def get_defs_ref(self, core_mode_ref):
full_def = super().get_defs_ref(core_mode_ref)
choices = self._prioritized_defsref_choices[full_def]
ref, mode = core_mode_ref
if ref in ref_to_name:
for i, choice in enumerate(choices):
choices[i] = choice.replace(choices[0], ref_to_name[ref]) # type: ignore[call-overload]
return full_def
lib/galaxy/schema/invocation.py
+5 -33
View File
@@ -6,9 +6,6 @@ from typing import (
Generic,
List,
Optional,
Tuple,
Type,
TypeVar,
Union,
)
@@ -20,7 +17,6 @@ from pydantic import (
UUID1,
UUID4,
)
from pydantic.json_schema import GenerateJsonSchema
from typing_extensions import (
Annotated,
Literal,
@@ -33,6 +29,10 @@ from galaxy.schema.fields import (
literal_to_value,
ModelClassField,
)
from galaxy.schema.generics import (
DatabaseIdT,
GenericModel,
)
from galaxy.schema.schema import (
CreateTimeField,
DataItemSourceType,
@@ -86,38 +86,10 @@ class CancelReason(str, Enum):
cancelled_on_review = "cancelled_on_review"
DatabaseIdT = TypeVar("DatabaseIdT")
ref_to_name = {}
class InvocationMessageBase(BaseModel):
class InvocationMessageBase(GenericModel):
reason: Union[CancelReason, FailureReason, WarningReason]
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
@classmethod
def model_parametrized_name(cls, params: Tuple[Type[Any], ...]) -> str:
suffix = "Response" if params[0] is EncodedDatabaseIdField else "Incoming"
class_name = cls.__name__.split("Generic", 1)[-1]
return f"{class_name}{suffix}"
@classmethod
def __get_pydantic_core_schema__(cls, *args, **kwargs):
result = super().__get_pydantic_core_schema__(*args, **kwargs)
ref_to_name[result["ref"]] = cls.__name__
return result
class CustomJsonSchema(GenerateJsonSchema):
def get_defs_ref(self, core_mode_ref):
full_def = super().get_defs_ref(core_mode_ref)
choices = self._prioritized_defsref_choices[full_def]
ref, mode = core_mode_ref
if ref in ref_to_name:
for i, choice in enumerate(choices):
choices[i] = choice.replace(choices[0], ref_to_name[ref]) # type: ignore[call-overload]
return full_def
class GenericInvocationCancellationReviewFailed(InvocationMessageBase, Generic[DatabaseIdT]):
reason: Literal[CancelReason.cancelled_on_review]
lib/galaxy/schema/notifications.py
+34 -8
View File
@@ -3,6 +3,7 @@ from enum import Enum
from typing import (
Any,
Dict,
Generic,
List,
Optional,
Union,
@@ -22,6 +23,10 @@ from galaxy.schema.fields import (
DecodedDatabaseIdField,
EncodedDatabaseIdField,
)
from galaxy.schema.generics import (
DatabaseIdT,
GenericModel,
)
from galaxy.schema.schema import Model
from galaxy.schema.types import (
AbsoluteOrRelativeUrl,
@@ -259,30 +264,30 @@ class NotificationCreateData(Model):
)
class NotificationRecipients(Model):
class GenericNotificationRecipients(GenericModel, Generic[DatabaseIdT]):
"""The recipients of a notification. Can be a combination of users, groups and roles."""
user_ids: List[DecodedDatabaseIdField] = Field(
user_ids: List[DatabaseIdT] = Field(
default=[],
title="User IDs",
description="The list of encoded user IDs of the users that should receive the notification.",
)
group_ids: List[DecodedDatabaseIdField] = Field(
group_ids: List[DatabaseIdT] = Field(
default=[],
title="Group IDs",
description="The list of encoded group IDs of the groups that should receive the notification.",
)
role_ids: List[DecodedDatabaseIdField] = Field(
role_ids: List[DatabaseIdT] = Field(
default=[],
title="Role IDs",
description="The list of encoded role IDs of the roles that should receive the notification.",
)
class NotificationCreateRequest(Model):
class GenericNotificationCreate(GenericModel, Generic[DatabaseIdT]):
"""Contains the recipients and the notification to create."""
recipients: NotificationRecipients = Field(
recipients: GenericNotificationRecipients[DatabaseIdT] = Field(
...,
title="Recipients",
description="The recipients of the notification. Can be a combination of users, groups and roles.",
@@ -294,6 +299,20 @@ class NotificationCreateRequest(Model):
)
class NotificationCreateRequest(GenericNotificationCreate[int]):
galaxy_url: Optional[str] = Field(
None,
title="Galaxy URL",
description="The URL of the Galaxy instance. Used to generate links in the notification content.",
)
NotificationRecipients = GenericNotificationRecipients[int]
NotificationCreateRequestBody = GenericNotificationCreate[DecodedDatabaseIdField]
class BroadcastNotificationCreateRequest(NotificationCreateData):
"""A notification create request specific for broadcasting."""
@@ -405,8 +424,15 @@ class NotificationChannelSettings(Model):
title="Push",
description="Whether the user wants to receive push notifications in the browser for this category.",
)
# TODO: Add more channels
# email: bool # Not supported for now
email: bool = Field(
default=True,
title="Email",
description=(
"Whether the user wants to receive email notifications for this category. "
"This setting will be ignored unless the server supports asynchronous tasks."
),
)
# TODO: Add more channels here and implement the corresponding plugin in lib/galaxy/managers/notification.py
# matrix: bool # Possible future Matrix.org integration?
lib/galaxy/webapps/galaxy/api/notifications.py
+20 -6
View File
@@ -3,7 +3,10 @@ API operations on Notification objects.
"""
import logging
from typing import Optional
from typing import (
Optional,
Union,
)
from fastapi import (
Body,
@@ -19,7 +22,7 @@ from galaxy.schema.notifications import (
BroadcastNotificationResponse,
NotificationBroadcastUpdateRequest,
NotificationCreatedResponse,
NotificationCreateRequest,
NotificationCreateRequestBody,
NotificationsBatchRequest,
NotificationsBatchUpdateResponse,
NotificationStatusSummary,
@@ -30,6 +33,7 @@ from galaxy.schema.notifications import (
UserNotificationsBatchUpdateRequest,
UserNotificationUpdateRequest,
)
from galaxy.schema.schema import AsyncTaskResultSummary
from galaxy.schema.types import OffsetNaiveDatetime
from galaxy.webapps.galaxy.api.common import NotificationIdPathParam
from galaxy.webapps.galaxy.services.notifications import NotificationService
@@ -66,10 +70,20 @@ class FastAPINotifications:
)
def get_notification_preferences(
self,
response: Response,
trans: ProvidesUserContext = DependsOnTrans,
) -> UserNotificationPreferences:
"""Anonymous users cannot have notification preferences. They will receive only broadcasted notifications."""
return self.service.get_user_notification_preferences(trans)
"""Anonymous users cannot have notification preferences. They will receive only broadcasted notifications.
- The settings will contain all possible channels, but the client should only show the ones that are really supported by the server.
The supported channels are returned in the `supported-channels` header.
"""
result = self.service.get_user_notification_preferences(trans)
# Inform the client which channels are really supported by the server since the settings will contain all possible channels.
response.headers["supported-channels"] = str.join(
",", self.service.notification_manager.get_supported_channels()
)
return result
@router.put(
"/api/notifications/preferences",
@@ -219,8 +233,8 @@ class FastAPINotifications:
def send_notification(
self,
trans: ProvidesUserContext = DependsOnTrans,
payload: NotificationCreateRequest = Body(),
) -> NotificationCreatedResponse:
payload: NotificationCreateRequestBody = Body(),
) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]:
"""Sends a notification to a list of recipients (users, groups or roles)."""
return self.service.send_notification(sender_context=trans, payload=payload)
lib/galaxy/webapps/galaxy/fast_app.py
+1 -1
View File
@@ -12,7 +12,7 @@ from fastapi.openapi.constants import REF_TEMPLATE
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import Response
from galaxy.schema.invocation import CustomJsonSchema
from galaxy.schema.generics import CustomJsonSchema
from galaxy.version import VERSION
from galaxy.webapps.base.api import (
add_exception_handler,
lib/galaxy/webapps/galaxy/services/histories.py
+3 -3
View File
@@ -40,7 +40,6 @@ from galaxy.managers.histories import (
HistoryManager,
HistorySerializer,
)
from galaxy.managers.notification import NotificationManager
from galaxy.managers.users import UserManager
from galaxy.model import HistoryDatasetAssociation
from galaxy.model.base import transaction
@@ -89,6 +88,7 @@ from galaxy.webapps.galaxy.services.base import (
ServesExportStores,
ServiceBase,
)
from galaxy.webapps.galaxy.services.notifications import NotificationService
from galaxy.webapps.galaxy.services.sharable import ShareableService
log = logging.getLogger(__name__)
@@ -121,7 +121,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
history_export_manager: HistoryExportManager,
filters: HistoryFilters,
short_term_storage_allocator: ShortTermStorageAllocator,
notification_manager: NotificationManager,
notification_service: NotificationService,
):
super().__init__(security)
self.manager = manager
@@ -131,7 +131,7 @@ class HistoriesService(ServiceBase, ConsumesModelStores, ServesExportStores):
self.citations_manager = citations_manager
self.history_export_manager = history_export_manager
self.filters = filters
self.shareable_service = ShareableHistoryService(self.manager, self.serializer, notification_manager)
self.shareable_service = ShareableHistoryService(self.manager, self.serializer, notification_service)
self.short_term_storage_allocator = short_term_storage_allocator
def index(
lib/galaxy/webapps/galaxy/services/notifications.py
+40 -5
View File
@@ -4,8 +4,10 @@ from typing import (
NoReturn,
Optional,
Set,
Union,
)
from galaxy.celery.tasks import send_notification_to_recipients_async
from galaxy.exceptions import (
AdminRequiredException,
AuthenticationRequired,
@@ -23,6 +25,7 @@ from galaxy.schema.notifications import (
NotificationBroadcastUpdateRequest,
NotificationCreatedResponse,
NotificationCreateRequest,
NotificationCreateRequestBody,
NotificationResponse,
NotificationsBatchUpdateResponse,
NotificationStatusSummary,
@@ -33,7 +36,11 @@ from galaxy.schema.notifications import (
UserNotificationResponse,
UserNotificationUpdateRequest,
)
from galaxy.webapps.galaxy.services.base import ServiceBase
from galaxy.schema.schema import AsyncTaskResultSummary
from galaxy.webapps.galaxy.services.base import (
async_task_summary,
ServiceBase,
)
class NotificationService(ServiceBase):
@@ -41,12 +48,40 @@ class NotificationService(ServiceBase):
self.notification_manager = notification_manager
def send_notification(
self, sender_context: ProvidesUserContext, payload: NotificationCreateRequest
) -> NotificationCreatedResponse:
"""Sends a notification to a list of recipients (users, groups or roles)."""
self, sender_context: ProvidesUserContext, payload: NotificationCreateRequestBody
) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]:
"""Sends a notification to a list of recipients (users, groups or roles).
Before sending the notification, it checks if the requesting user has the necessary permissions to do so.
"""
self.notification_manager.ensure_notifications_enabled()
self._ensure_user_can_send_notifications(sender_context)
notification, recipient_user_count = self.notification_manager.send_notification_to_recipients(payload)
galaxy_url = (
str(sender_context.url_builder("/", qualified=True)).rstrip("/") if sender_context.url_builder else None
)
request = NotificationCreateRequest.model_construct(
notification=payload.notification,
recipients=payload.recipients,
galaxy_url=galaxy_url,
)
return self.send_notification_internal(request)
def send_notification_internal(
self, request: NotificationCreateRequest, force_sync: bool = False
) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]:
"""Sends a notification to a list of recipients (users, groups or roles).
If `force_sync` is set to `True`, the notification recipients will be processed synchronously instead of
in a background task.
Note: This function is meant for internal use from other services that don't need to check sender permissions.
"""
if self.notification_manager.can_send_notifications_async and not force_sync:
result = send_notification_to_recipients_async.delay(request)
summary = async_task_summary(result)
return summary
notification, recipient_user_count = self.notification_manager.send_notification_to_recipients(request)
return NotificationCreatedResponse(
total_notifications_sent=recipient_user_count,
notification=NotificationResponse.model_validate(notification),
lib/galaxy/webapps/galaxy/services/pages.py
+3 -3
View File
@@ -8,7 +8,6 @@ from galaxy.managers.markdown_util import (
internal_galaxy_markdown_to_pdf,
to_basic_markdown,
)
from galaxy.managers.notification import NotificationManager
from galaxy.managers.pages import (
PageManager,
PageSerializer,
@@ -33,6 +32,7 @@ from galaxy.webapps.galaxy.services.base import (
ensure_celery_tasks_enabled,
ServiceBase,
)
from galaxy.webapps.galaxy.services.notifications import NotificationService
from galaxy.webapps.galaxy.services.sharable import ShareableService
log = logging.getLogger(__name__)
@@ -51,12 +51,12 @@ class PagesService(ServiceBase):
manager: PageManager,
serializer: PageSerializer,
short_term_storage_allocator: ShortTermStorageAllocator,
notification_manager: NotificationManager,
notification_service: NotificationService,
):
super().__init__(security)
self.manager = manager
self.serializer = serializer
self.shareable_service = ShareableService(self.manager, self.serializer, notification_manager)
self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
self.short_term_storage_allocator = short_term_storage_allocator
def index(
lib/galaxy/webapps/galaxy/services/sharable.py
+21 -9
View File
@@ -12,7 +12,6 @@ from typing import (
from sqlalchemy import false
from galaxy.managers import base
from galaxy.managers.notification import NotificationManager
from galaxy.managers.sharable import (
SharableModelManager,
SharableModelSerializer,
@@ -41,6 +40,7 @@ from galaxy.schema.schema import (
SharingStatus,
UserIdentifier,
)
from galaxy.webapps.galaxy.services.notifications import NotificationService
log = logging.getLogger(__name__)
@@ -67,11 +67,11 @@ class ShareableService:
self,
manager: SharableModelManager,
serializer: SharableModelSerializer,
notification_manager: NotificationManager,
notification_service: NotificationService,
) -> None:
self.manager = manager
self.serializer = serializer
self.notification_manager = notification_manager
self.notification_service = notification_service
def set_slug(self, trans, id: DecodedDatabaseIdField, payload: SetSlugPayload):
item = self._get_item_by_id(trans, id)
@@ -117,7 +117,8 @@ class ShareableService:
base_status = self._get_sharing_status(trans, item)
status = self.share_with_status_cls.model_construct(**base_status.model_dump(), extra=extra)
status.errors.extend(errors)
self._send_notification_to_users(users_to_notify, item, status)
galaxy_url = str(trans.url_builder("/", qualified=True)).rstrip("/") if trans.url_builder else None
self._send_notification_to_users(users_to_notify, item, status, galaxy_url)
return status
def _share_with_options(
@@ -173,10 +174,20 @@ class ShareableService:
return send_to_users, send_to_err
def _send_notification_to_users(self, users_to_notify: Set[User], item: SharableItem, status: ShareWithStatus):
if self.notification_manager.notifications_enabled and not status.errors and users_to_notify:
request = SharedItemNotificationFactory.build_notification_request(item, users_to_notify, status)
self.notification_manager.send_notification_to_recipients(request)
def _send_notification_to_users(
self, users_to_notify: Set[User], item: SharableItem, status: ShareWithStatus, galaxy_url: Optional[str] = None
):
if (
self.notification_service.notification_manager.notifications_enabled
and not status.errors
and users_to_notify
):
request = SharedItemNotificationFactory.build_notification_request(
item, users_to_notify, status, galaxy_url
)
# We can set force_sync=True here because we already have the set of users to notify
# and there is no need to resolve them asynchronously as no groups or roles are involved.
self.notification_service.send_notification_internal(request, force_sync=True)
class SharedItemNotificationFactory:
@@ -191,7 +202,7 @@ class SharedItemNotificationFactory:
@staticmethod
def build_notification_request(
item: SharableItem, users_to_notify: Set[User], status: ShareWithStatus
item: SharableItem, users_to_notify: Set[User], status: ShareWithStatus, galaxy_url: Optional[str] = None
) -> NotificationCreateRequest:
user_ids = [user.id for user in users_to_notify]
request = NotificationCreateRequest(
@@ -207,5 +218,6 @@ class SharedItemNotificationFactory:
slug=status.username_and_slug,
),
),
galaxy_url=galaxy_url,
)
return request
lib/galaxy/webapps/galaxy/services/visualizations.py
+3 -3
View File
@@ -2,7 +2,6 @@ import logging
from typing import Tuple
from galaxy import exceptions
from galaxy.managers.notification import NotificationManager
from galaxy.managers.visualizations import (
VisualizationManager,
VisualizationSerializer,
@@ -13,6 +12,7 @@ from galaxy.schema.visualization import (
)
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.webapps.galaxy.services.base import ServiceBase
from galaxy.webapps.galaxy.services.notifications import NotificationService
from galaxy.webapps.galaxy.services.sharable import ShareableService
log = logging.getLogger(__name__)
@@ -30,12 +30,12 @@ class VisualizationsService(ServiceBase):
security: IdEncodingHelper,
manager: VisualizationManager,
serializer: VisualizationSerializer,
notification_manager: NotificationManager,
notification_service: NotificationService,
):
super().__init__(security)
self.manager = manager
self.serializer = serializer
self.shareable_service = ShareableService(self.manager, self.serializer, notification_manager)
self.shareable_service = ShareableService(self.manager, self.serializer, notification_service)
# TODO: add the rest of the API actions here and call them directly from the API controller
lib/galaxy/webapps/galaxy/services/workflows.py
+3 -3
View File
@@ -13,7 +13,6 @@ from galaxy import (
web,
)
from galaxy.managers.context import ProvidesUserContext
from galaxy.managers.notification import NotificationManager
from galaxy.managers.workflows import (
RefactorResponse,
WorkflowContentsManager,
@@ -33,6 +32,7 @@ from galaxy.schema.workflows import (
)
from galaxy.util.tool_shed.tool_shed_registry import Registry
from galaxy.webapps.galaxy.services.base import ServiceBase
from galaxy.webapps.galaxy.services.notifications import NotificationService
from galaxy.webapps.galaxy.services.sharable import ShareableService
from galaxy.workflow.run import queue_invoke
from galaxy.workflow.run_request import build_workflow_run_configs
@@ -51,12 +51,12 @@ class WorkflowsService(ServiceBase):
workflow_contents_manager: WorkflowContentsManager,
serializer: WorkflowSerializer,
tool_shed_registry: Registry,
notification_manager: NotificationManager,
notification_service: NotificationService,
):
self._workflows_manager = workflows_manager
self._workflow_contents_manager = workflow_contents_manager
self._serializer = serializer
self.shareable_service = ShareableService(workflows_manager, serializer, notification_manager)
self.shareable_service = ShareableService(workflows_manager, serializer, notification_service)
self._tool_shed_registry = tool_shed_registry
def index(
test/integration/test_notifications.py
+68 -16
View File
@@ -44,7 +44,7 @@ def notification_broadcast_test_data(subject: Optional[str] = None, message: Opt
}
class TestNotificationsIntegration(IntegrationTestCase):
class NotificationsIntegrationBase(IntegrationTestCase):
dataset_populator: DatasetPopulator
task_based = False
framework_tool_and_types = False
@@ -52,6 +52,7 @@ class TestNotificationsIntegration(IntegrationTestCase):
@classmethod
def handle_galaxy_config_kwds(cls, config):
super().handle_galaxy_config_kwds(config)
config["enable_celery_tasks"] = False
config["enable_notification_system"] = True
def setUp(self):
@@ -66,12 +67,18 @@ class TestNotificationsIntegration(IntegrationTestCase):
before_creating_notifications = datetime.utcnow()
# Only user1 will receive this notification
created_response_1 = self._send_test_notification_to([user1["id"]], message="test_notification_status 1")
assert created_response_1["total_notifications_sent"] == 1
subject1 = f"notification_{uuid4()}"
created_response_1 = self._send_test_notification_to(
[user1["id"]], subject=subject1, message="test_notification_status 1"
)
self._assert_notifications_sent(created_response_1, expected_count=1)
# Both user1 and user2 will receive this notification
created_response_2 = self._send_test_notification_to([user1["id"], user2["id"]], "test_notification_status 2")
assert created_response_2["total_notifications_sent"] == 2
subject2 = f"notification_{uuid4()}"
created_response_2 = self._send_test_notification_to(
[user1["id"], user2["id"]], subject=subject2, message="test_notification_status 2"
)
self._assert_notifications_sent(created_response_2, expected_count=2)
# All users will receive this broadcasted notification
created_response_3 = self._send_broadcast_notification("test_notification_status 3")
@@ -113,7 +120,8 @@ class TestNotificationsIntegration(IntegrationTestCase):
assert len(status["broadcasts"]) == 0
# Updating a notification association value should return that notification in the next request
notification_id = created_response_2["notification"]["id"]
notification_id = self._get_notification_id_by_subject(subject2)
assert notification_id is not None
self._update_notification(notification_id, update_state={"seen": True})
status = self._get_notifications_status_since(after_creating_notifications)
assert status["total_unread_count"] == 0
@@ -131,16 +139,21 @@ class TestNotificationsIntegration(IntegrationTestCase):
user1 = self._create_test_user()
user2 = self._create_test_user()
subject = f"notification_{uuid4()}"
created_response = self._send_test_notification_to(
[user1["id"]], message="test_user_cannot_access_other_users_notifications"
[user1["id"]], subject=subject, message="test_user_cannot_access_other_users_notifications"
)
notification_id = created_response["notification"]["id"]
self._assert_notifications_sent(created_response, expected_count=1)
notification_id = None
with self._different_user(user1["email"]):
notification_id = self._get_notification_id_by_subject(subject)
assert notification_id is not None
response = self._get(f"notifications/{notification_id}")
self._assert_status_code_is_ok(response)
with self._different_user(user2["email"]):
assert notification_id is not None
response = self._get(f"notifications/{notification_id}")
self._assert_status_code_is(response, 404)
@@ -150,13 +163,15 @@ class TestNotificationsIntegration(IntegrationTestCase):
before_creating_notifications = datetime.utcnow()
subject = f"notification_{uuid4()}"
created_response = self._send_test_notification_to(
[user1["id"], user2["id"]], message="test_delete_notification_by_user"
[user1["id"], user2["id"]], subject=subject, message="test_delete_notification_by_user"
)
assert created_response["total_notifications_sent"] == 2
notification_id = created_response["notification"]["id"]
self._assert_notifications_sent(created_response, expected_count=2)
with self._different_user(user1["email"]):
notification_id = self._get_notification_id_by_subject(subject)
assert notification_id is not None
response = self._get(f"notifications/{notification_id}")
self._assert_status_code_is_ok(response)
self._delete(f"notifications/{notification_id}")
@@ -169,6 +184,7 @@ class TestNotificationsIntegration(IntegrationTestCase):
assert len(status["broadcasts"]) == 0
with self._different_user(user2["email"]):
assert notification_id is not None
response = self._get(f"notifications/{notification_id}")
self._assert_status_code_is_ok(response)
@@ -195,11 +211,14 @@ class TestNotificationsIntegration(IntegrationTestCase):
def test_update_notifications(self):
recipient_user = self._create_test_user()
created_user_notification_response = self._send_test_notification_to(
[recipient_user["id"]], subject="User Notification to update"
)
assert created_user_notification_response["total_notifications_sent"] == 1
user_notification_id = created_user_notification_response["notification"]["id"]
subject = f"User Notification to update {uuid4()}"
created_user_notification_response = self._send_test_notification_to([recipient_user["id"]], subject=subject)
self._assert_notifications_sent(created_user_notification_response, expected_count=1)
with self._different_user(recipient_user["email"]):
user_notification_id = self._get_notification_id_by_subject(subject)
assert user_notification_id is not None
created_broadcast_notification_response = self._send_broadcast_notification(
subject="Broadcasted Notification to update"
@@ -402,3 +421,36 @@ class TestNotificationsIntegration(IntegrationTestCase):
def _update_notification(self, notification_id: str, update_state: Dict[str, Any]):
update_response = self._put(f"notifications/{notification_id}", data=update_state, json=True)
self._assert_status_code_is(update_response, 204)
def _assert_notifications_sent(self, response, expected_count: int = 0):
if self.task_based:
task_id = response["id"]
assert task_id is not None
self.dataset_populator.wait_on_task_id(task_id)
else:
assert response["total_notifications_sent"] == expected_count
def _get_notification_id_by_subject(self, subject: str) -> Optional[str]:
notifications = self._get("notifications").json()
for notification in notifications:
if notification["content"]["subject"] == subject:
return notification["id"]
return None
class TestNotificationsIntegration(NotificationsIntegrationBase):
task_based = False
@classmethod
def handle_galaxy_config_kwds(cls, config):
super().handle_galaxy_config_kwds(config)
config["enable_celery_tasks"] = False
class TestNotificationsIntegrationTaskBased(NotificationsIntegrationBase):
task_based = True
@classmethod
def handle_galaxy_config_kwds(cls, config):
super().handle_galaxy_config_kwds(config)
config["enable_celery_tasks"] = True
test/unit/app/managers/test_NotificationManager.py
+1
View File
@@ -81,6 +81,7 @@ class NotificationManagerBaseTestCase(NotificationsBaseTestCase):
user_ids=[user.id for user in users],
),
notification=notification_data,
galaxy_url="https://test.galaxy.url",
)
created_notification, notifications_sent = self.notification_manager.send_notification_to_recipients(request)
return created_notification, notifications_sent
test/unit/app/test_celery.py
-4
View File
@@ -26,10 +26,6 @@ def test_default_configuration():
"task": "galaxy.cleanup_short_term_storage",
"schedule": galaxy_conf.short_term_storage_cleanup_interval,
}
assert conf.beat_schedule["cleanup-expired-notifications"] == {
"task": "galaxy.cleanup_expired_notifications",
"schedule": galaxy_conf.expired_notifications_cleanup_interval,
}
def test_galaxycelery_trim_module_name():