diff --git a/lib/galaxy/workflow/completion_hooks/__init__.py b/lib/galaxy/workflow/completion_hooks/__init__.py index dab2e73b078..fd17ff7db29 100644 --- a/lib/galaxy/workflow/completion_hooks/__init__.py +++ b/lib/galaxy/workflow/completion_hooks/__init__.py @@ -8,7 +8,10 @@ a workflow invocation completes. To add a new hook: 1. Create a new module in this package 2. Define a class that extends WorkflowCompletionHook -3. Register it in WorkflowCompletionHookRegistry._load_hooks() +3. Set a unique `plugin_type` class attribute +4. Export the class via `__all__` + +Hooks are automatically discovered via the plugin_type attribute. """ import logging @@ -17,10 +20,10 @@ from typing import ( TYPE_CHECKING, ) +import galaxy.workflow.completion_hooks from galaxy.structured_app import MinimalManagerApp +from galaxy.util import plugin_config from galaxy.workflow.completion_hooks.base import WorkflowCompletionHook -from galaxy.workflow.completion_hooks.export import ExportToFileSourceHook -from galaxy.workflow.completion_hooks.notification import SendNotificationHook if TYPE_CHECKING: from galaxy.model import WorkflowInvocationCompletion @@ -55,11 +58,8 @@ class WorkflowCompletionHookRegistry: def _load_hooks(self) -> None: """Load and instantiate all available hooks.""" - # Register all available hooks - self.hooks = { - "export_to_file_source": ExportToFileSourceHook, - "send_notification": SendNotificationHook, - } + # Auto-discover hooks via plugin_type attribute + self.hooks = plugin_config.plugins_dict(galaxy.workflow.completion_hooks, "plugin_type") # Instantiate all hooks for hook_name, hook_class in self.hooks.items(): diff --git a/lib/galaxy/workflow/completion_hooks/base.py b/lib/galaxy/workflow/completion_hooks/base.py index c0ae9a6c3cc..c5edd02ae5b 100644 --- a/lib/galaxy/workflow/completion_hooks/base.py +++ b/lib/galaxy/workflow/completion_hooks/base.py @@ -26,7 +26,7 @@ class WorkflowCompletionHook(ABC): To implement a hook: 1. Extend this class - 2. Set a unique `name` attribute + 2. Set a unique `plugin_type` attribute 3. Implement the `execute()` method 4. Optionally override `is_applicable()` to add conditions @@ -34,8 +34,8 @@ class WorkflowCompletionHook(ABC): for the same completion (e.g., after a Celery task retry). """ - # Unique identifier for this hook - must be set by subclasses - name: str = "base" + # Plugin type identifier for automatic discovery - must be set by subclasses + plugin_type: str def __init__(self, app: MinimalManagerApp): """ diff --git a/lib/galaxy/workflow/completion_hooks/export.py b/lib/galaxy/workflow/completion_hooks/export.py index 25615e0b578..c6bd47b7562 100644 --- a/lib/galaxy/workflow/completion_hooks/export.py +++ b/lib/galaxy/workflow/completion_hooks/export.py @@ -6,6 +6,9 @@ file source (e.g., remote storage). """ import logging + +__all__ = ("ExportToFileSourceHook",) + from typing import ( Any, Optional, @@ -52,7 +55,7 @@ class ExportToFileSourceHook(WorkflowCompletionHook): } """ - name = "export_to_file_source" + plugin_type = "export_to_file_source" def execute(self, completion: "WorkflowInvocationCompletion") -> None: """ diff --git a/lib/galaxy/workflow/completion_hooks/notification.py b/lib/galaxy/workflow/completion_hooks/notification.py index cb2d4cf97ff..4ab247a477d 100644 --- a/lib/galaxy/workflow/completion_hooks/notification.py +++ b/lib/galaxy/workflow/completion_hooks/notification.py @@ -5,6 +5,9 @@ This hook sends a notification to the user when their workflow completes. """ import logging + +__all__ = ("SendNotificationHook",) + from typing import TYPE_CHECKING from galaxy.schema.notifications import ( @@ -32,7 +35,7 @@ class SendNotificationHook(WorkflowCompletionHook): and a summary of job states. """ - name = "send_notification" + plugin_type = "send_notification" def execute(self, completion: "WorkflowInvocationCompletion") -> None: """