Use plugin_util for workflow completion hook discovery

Replace manual hook registration with automatic discovery using
Galaxy's plugin_config.plugins_dict() framework. Hooks are now
discovered via their plugin_type class attribute.
This commit is contained in:
mvdbeek
2026-01-21 14:44:55 +01:00
parent 4f424a4e2a
commit 73fb2d80ba
4 changed files with 19 additions and 13 deletions
@@ -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():
+3 -3
View File
@@ -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):
"""
@@ -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:
"""
@@ -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:
"""