mattermost/python-sdk/build/lib/mattermost_plugin/hooks.py

# Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
# See LICENSE.txt for license information.

"""
Hook registration system for Mattermost Python plugins.

This module provides the `@hook` decorator and `HookName` enum for registering
plugin methods as hook handlers.

Usage:
    from mattermost_plugin import Plugin, hook, HookName

    class MyPlugin(Plugin):
        @hook(HookName.OnActivate)
        def on_activate(self) -> None:
            self.logger.info("Plugin activated!")

        @hook(HookName.MessageWillBePosted)
        def filter_message(self, context, post):
            if "spam" in post.message.lower():
                return None, "Message rejected: spam detected"
            return post, ""

Alternative usage with string names:
    @hook("OnActivate")
    def on_activate(self) -> None:
        pass
"""

from __future__ import annotations

import asyncio
import inspect
from enum import Enum
from functools import wraps
from typing import (
    Any,
    Callable,
    Optional,
    TypeVar,
    Union,
    overload,
)

# Python 3.9 compatibility: ParamSpec was added in 3.10
try:
    from typing import ParamSpec
except ImportError:
    from typing_extensions import ParamSpec

P = ParamSpec("P")
R = TypeVar("R")
F = TypeVar("F", bound=Callable[..., Any])

# Attribute names used to mark hook metadata on decorated functions
HOOK_MARKER_ATTR = "_mattermost_hook_name"


class HookName(str, Enum):
    """
    Canonical hook names matching server/public/plugin/hooks.go.

    These names are used in the `Implemented()` RPC to tell the server
    which hooks this plugin handles.

    Usage:
        @hook(HookName.OnActivate)
        def on_activate(self) -> None:
            pass
    """

    # Lifecycle hooks
    OnActivate = "OnActivate"
    OnDeactivate = "OnDeactivate"
    OnConfigurationChange = "OnConfigurationChange"
    OnInstall = "OnInstall"
    OnSendDailyTelemetry = "OnSendDailyTelemetry"
    RunDataRetention = "RunDataRetention"
    OnCloudLimitsUpdated = "OnCloudLimitsUpdated"
    ConfigurationWillBeSaved = "ConfigurationWillBeSaved"

    # Message hooks
    MessageWillBePosted = "MessageWillBePosted"
    MessageWillBeUpdated = "MessageWillBeUpdated"
    MessageHasBeenPosted = "MessageHasBeenPosted"
    MessageHasBeenUpdated = "MessageHasBeenUpdated"
    MessagesWillBeConsumed = "MessagesWillBeConsumed"
    MessageHasBeenDeleted = "MessageHasBeenDeleted"
    FileWillBeUploaded = "FileWillBeUploaded"
    ReactionHasBeenAdded = "ReactionHasBeenAdded"
    ReactionHasBeenRemoved = "ReactionHasBeenRemoved"
    NotificationWillBePushed = "NotificationWillBePushed"
    EmailNotificationWillBeSent = "EmailNotificationWillBeSent"
    PreferencesHaveChanged = "PreferencesHaveChanged"

    # User hooks
    UserHasBeenCreated = "UserHasBeenCreated"
    UserWillLogIn = "UserWillLogIn"
    UserHasLoggedIn = "UserHasLoggedIn"
    UserHasBeenDeactivated = "UserHasBeenDeactivated"
    OnSAMLLogin = "OnSAMLLogin"

    # Channel/Team hooks
    ChannelHasBeenCreated = "ChannelHasBeenCreated"
    UserHasJoinedChannel = "UserHasJoinedChannel"
    UserHasLeftChannel = "UserHasLeftChannel"
    UserHasJoinedTeam = "UserHasJoinedTeam"
    UserHasLeftTeam = "UserHasLeftTeam"

    # Command hooks
    ExecuteCommand = "ExecuteCommand"

    # WebSocket hooks
    OnWebSocketConnect = "OnWebSocketConnect"
    OnWebSocketDisconnect = "OnWebSocketDisconnect"
    WebSocketMessageHasBeenPosted = "WebSocketMessageHasBeenPosted"

    # Cluster hooks
    OnPluginClusterEvent = "OnPluginClusterEvent"

    # Shared channels hooks
    OnSharedChannelsSyncMsg = "OnSharedChannelsSyncMsg"
    OnSharedChannelsPing = "OnSharedChannelsPing"
    OnSharedChannelsAttachmentSyncMsg = "OnSharedChannelsAttachmentSyncMsg"
    OnSharedChannelsProfileImageSyncMsg = "OnSharedChannelsProfileImageSyncMsg"

    # Support hooks
    GenerateSupportData = "GenerateSupportData"

    # HTTP hooks (Phase 8 - streaming)
    ServeHTTP = "ServeHTTP"
    # ServeMetrics = "ServeMetrics"  # Deferred to 08-02


# Set of all valid canonical hook names for validation
VALID_HOOK_NAMES: frozenset[str] = frozenset(h.value for h in HookName)


class HookRegistrationError(Exception):
    """Raised when hook registration fails."""

    pass


def get_hook_name(func: Callable[..., Any]) -> Optional[str]:
    """
    Get the canonical hook name from a decorated function.

    Args:
        func: A function that may have been decorated with @hook.

    Returns:
        The canonical hook name if the function is a hook handler, None otherwise.
    """
    return getattr(func, HOOK_MARKER_ATTR, None)


def is_hook_handler(func: Callable[..., Any]) -> bool:
    """
    Check if a function has been decorated as a hook handler.

    Args:
        func: A function to check.

    Returns:
        True if the function is decorated with @hook, False otherwise.
    """
    return hasattr(func, HOOK_MARKER_ATTR)


# Overloads for type checking support
@overload
def hook(name_or_func: F) -> F:
    """Decorator form: @hook (infers name from method)."""
    ...


@overload
def hook(name_or_func: Union[str, HookName]) -> Callable[[F], F]:
    """Decorator form: @hook("OnActivate") or @hook(HookName.OnActivate)."""
    ...


def hook(
    name_or_func: Union[str, HookName, F, None] = None
) -> Union[F, Callable[[F], F]]:
    """
    Decorator to register a method as a plugin hook handler.

    This decorator marks methods on Plugin subclasses as hook handlers.
    The Plugin base class uses `__init_subclass__` to discover these
    methods and build a hook registry.

    Args:
        name_or_func: Either:
            - A HookName enum value (preferred): @hook(HookName.OnActivate)
            - A string hook name: @hook("OnActivate")
            - None to infer from method name: @hook or @hook()

    Returns:
        The decorated function with hook metadata attached.

    Raises:
        HookRegistrationError: If the hook name is invalid.

    Examples:
        # Preferred: explicit hook name with enum
        @hook(HookName.OnActivate)
        def on_activate(self) -> None:
            pass

        # Also allowed: string hook name
        @hook("OnActivate")
        def handle_activate(self) -> None:
            pass

        # Infer from method name (must match PascalCase convention)
        @hook
        def OnActivate(self) -> None:
            pass
    """

    def decorator(func: F) -> F:
        # Determine the canonical hook name
        if isinstance(name_or_func, HookName):
            canonical_name = name_or_func.value
        elif isinstance(name_or_func, str):
            canonical_name = name_or_func
        elif name_or_func is None or callable(name_or_func):
            # Infer from function name - must match exactly
            canonical_name = func.__name__
        else:
            raise HookRegistrationError(
                f"Invalid hook name type: {type(name_or_func).__name__}"
            )

        # Validate the hook name
        if canonical_name not in VALID_HOOK_NAMES:
            raise HookRegistrationError(
                f"Unknown hook name: '{canonical_name}'. "
                f"Valid hooks: {sorted(VALID_HOOK_NAMES)}"
            )

        # Mark the function with hook metadata
        # Preserve async nature of the function by using appropriate wrapper
        if inspect.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                return await func(*args, **kwargs)

            # Attach hook metadata
            setattr(async_wrapper, HOOK_MARKER_ATTR, canonical_name)
            return async_wrapper  # type: ignore[return-value]
        else:
            @wraps(func)
            def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
                return func(*args, **kwargs)

            # Attach hook metadata
            setattr(wrapper, HOOK_MARKER_ATTR, canonical_name)
            return wrapper  # type: ignore[return-value]

    # Handle both @hook and @hook() and @hook(HookName.X)
    if callable(name_or_func):
        # Called as @hook without parentheses
        return decorator(name_or_func)
    else:
        # Called as @hook(...) with arguments
        return decorator