Refactor mention decorator from gatekeeper to enrichment pattern

Author: joshuadavidthomas

src/django_github_app/mentions.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import re
+from dataclasses import dataclass
 from enum import Enum
 from typing import NamedTuple
 
 from gidgethub import sansio
 
+from .permissions import Permission
+
 
 class EventAction(NamedTuple):
     event: str
@@ -43,6 +46,13 @@ def all_events(cls) -> list[EventAction]:
         )
 
 
+@dataclass
+class MentionContext:
+    commands: list[str]
+    user_permission: Permission | None
+    scope: MentionScope | None
+
+
 class MentionMatch(NamedTuple):
     mention: str
     command: str | None
@@ -53,7 +63,9 @@ class MentionMatch(NamedTuple):
 QUOTE_PATTERN = re.compile(r"^\s*>.*$", re.MULTILINE)
 
 
-def parse_mentions(text: str, username: str) -> list[MentionMatch]:
+def parse_mentions(event: sansio.Event, username: str) -> list[MentionMatch]:
+    text = event.data.get("comment", {}).get("body", "")
+
     if not text:
         return []
 
@@ -77,11 +89,15 @@ def parse_mentions(text: str, username: str) -> list[MentionMatch]:
     return mentions
 
 
+def get_commands(event: sansio.Event, username: str) -> list[str]:
+    mentions = parse_mentions(event, username)
+    return [m.command for m in mentions if m.command]
+
+
 def check_event_for_mention(
     event: sansio.Event, command: str | None, username: str
 ) -> bool:
-    comment = event.data.get("comment", {}).get("body", "")
-    mentions = parse_mentions(comment, username)
+    mentions = parse_mentions(event, username)
 
     if not mentions:
         return False
@@ -92,21 +108,15 @@ def check_event_for_mention(
     return any(mention.command == command.lower() for mention in mentions)
 
 
-def check_event_scope(event: sansio.Event, scope: MentionScope | None) -> bool:
-    if scope is None:
-        return True
-
-    # For issue_comment events, we need to distinguish between issues and PRs
+def get_event_scope(event: sansio.Event) -> MentionScope | None:
     if event.event == "issue_comment":
         issue = event.data.get("issue", {})
         is_pull_request = "pull_request" in issue and issue["pull_request"] is not None
+        return MentionScope.PR if is_pull_request else MentionScope.ISSUE
 
-        # If scope is ISSUE, we only want actual issues (not PRs)
-        if scope == MentionScope.ISSUE:
-            return not is_pull_request
-        # If scope is PR, we only want pull requests
-        elif scope == MentionScope.PR:
-            return is_pull_request
+    for scope in MentionScope:
+        scope_events = scope.get_events()
+        if any(event_action.event == event.event for event_action in scope_events):
+            return scope
 
-    scope_events = scope.get_events()
-    return any(event_action.event == event.event for event_action in scope_events)
+    return None

src/django_github_app/permissions.py

@@ -133,81 +133,47 @@ def from_event(cls, event: sansio.Event) -> EventInfo:
 
 class PermissionCheck(NamedTuple):
     has_permission: bool
-    error_message: str | None
 
 
-PERMISSION_CHECK_ERROR_MESSAGE = """
-❌ **Permission Denied**
+async def aget_user_permission_from_event(
+    event: sansio.Event, gh: AsyncGitHubAPI
+) -> Permission | None:
+    comment_author, owner, repo = EventInfo.from_event(event)
 
-@{comment_author}, you need at least **{required_permission}** permission to use this command.
+    if not (comment_author and owner and repo):
+        return None
 
-Your current permission level: **{user_permission}**
-"""
+    return await aget_user_permission(gh, owner, repo, comment_author)
 
 
 async def acheck_mention_permission(
     event: sansio.Event, gh: AsyncGitHubAPI, required_permission: Permission
 ) -> PermissionCheck:
-    comment_author, owner, repo = EventInfo.from_event(event)
-
-    if not (comment_author and owner and repo):
-        return PermissionCheck(has_permission=False, error_message=None)
-
-    user_permission = await aget_user_permission(gh, owner, repo, comment_author)
+    user_permission = await aget_user_permission_from_event(event, gh)
 
-    if user_permission >= required_permission:
-        return PermissionCheck(has_permission=True, error_message=None)
+    if user_permission is None:
+        return PermissionCheck(has_permission=False)
 
-    return PermissionCheck(
-        has_permission=False,
-        error_message=PERMISSION_CHECK_ERROR_MESSAGE.format(
-            comment_author=comment_author,
-            required_permission=required_permission.name.lower(),
-            user_permission=user_permission.name.lower(),
-        ),
-    )
+    return PermissionCheck(has_permission=user_permission >= required_permission)
 
 
-def check_mention_permission(
-    event: sansio.Event, gh: SyncGitHubAPI, required_permission: Permission
-) -> PermissionCheck:
+def get_user_permission_from_event(
+    event: sansio.Event, gh: SyncGitHubAPI
+) -> Permission | None:
     comment_author, owner, repo = EventInfo.from_event(event)
 
     if not (comment_author and owner and repo):
-        return PermissionCheck(has_permission=False, error_message=None)
-
-    user_permission = get_user_permission(gh, owner, repo, comment_author)
-
-    if user_permission >= required_permission:
-        return PermissionCheck(has_permission=True, error_message=None)
-
-    return PermissionCheck(
-        has_permission=False,
-        error_message=PERMISSION_CHECK_ERROR_MESSAGE.format(
-            comment_author=comment_author,
-            required_permission=required_permission.name.lower(),
-            user_permission=user_permission.name.lower(),
-        ),
-    )
+        return None
 
+    return get_user_permission(gh, owner, repo, comment_author)
 
-def get_comment_post_url(event: sansio.Event) -> str | None:
-    if event.data.get("action") != "created":
-        return None
 
-    _, owner, repo = EventInfo.from_event(event)
+def check_mention_permission(
+    event: sansio.Event, gh: SyncGitHubAPI, required_permission: Permission
+) -> PermissionCheck:
+    user_permission = get_user_permission_from_event(event, gh)
 
-    if not (owner and repo):
-        return None
+    if user_permission is None:
+        return PermissionCheck(has_permission=False)
 
-    if "issue" in event.data:
-        issue_number = event.data["issue"]["number"]
-        return f"/repos/{owner}/{repo}/issues/{issue_number}/comments"
-    elif "pull_request" in event.data:
-        pr_number = event.data["pull_request"]["number"]
-        return f"/repos/{owner}/{repo}/issues/{pr_number}/comments"
-    elif "commit_sha" in event.data:
-        commit_sha = event.data["commit_sha"]
-        return f"/repos/{owner}/{repo}/commits/{commit_sha}/comments"
-
-    return None
+    return PermissionCheck(has_permission=user_permission >= required_permission)

src/django_github_app/routing.py

@@ -16,13 +16,13 @@
 from ._typing import override
 from .github import AsyncGitHubAPI
 from .github import SyncGitHubAPI
+from .mentions import MentionContext
 from .mentions import MentionScope
 from .mentions import check_event_for_mention
-from .mentions import check_event_scope
-from .permissions import Permission
-from .permissions import acheck_mention_permission
-from .permissions import check_mention_permission
-from .permissions import get_comment_post_url
+from .mentions import get_commands
+from .mentions import get_event_scope
+from .permissions import aget_user_permission_from_event
+from .permissions import get_user_permission_from_event
 
 AsyncCallback = Callable[..., Awaitable[None]]
 SyncCallback = Callable[..., None]
@@ -90,26 +90,15 @@ async def async_wrapper(
                 if not check_event_for_mention(event, command, username):
                     return
 
-                if not check_event_scope(event, scope):
+                event_scope = get_event_scope(event)
+                if scope is not None and event_scope != scope:
                     return
 
-                # Check permissions if required
-                if permission is not None:
-                    required_perm = Permission.from_string(permission)
-                    permission_check = await acheck_mention_permission(
-                        event, gh, required_perm
-                    )
-
-                    if not permission_check.has_permission:
-                        # Post error comment if we have an error message
-                        if permission_check.error_message:
-                            comment_url = get_comment_post_url(event)
-                            if comment_url:
-                                await gh.post(
-                                    comment_url,
-                                    data={"body": permission_check.error_message},
-                                )
-                        return
+                kwargs["mention"] = MentionContext(
+                    commands=get_commands(event, username),
+                    user_permission=await aget_user_permission_from_event(event, gh),
+                    scope=event_scope,
+                )
 
                 await func(event, gh, *args, **kwargs)  # type: ignore[func-returns-value]
 
@@ -123,26 +112,15 @@ def sync_wrapper(
                 if not check_event_for_mention(event, command, username):
                     return
 
-                if not check_event_scope(event, scope):
+                event_scope = get_event_scope(event)
+                if scope is not None and event_scope != scope:
                     return
 
-                # Check permissions if required
-                if permission is not None:
-                    required_perm = Permission.from_string(permission)
-                    permission_check = check_mention_permission(
-                        event, gh, required_perm
-                    )
-
-                    if not permission_check.has_permission:
-                        # Post error comment if we have an error message
-                        if permission_check.error_message:
-                            comment_url = get_comment_post_url(event)
-                            if comment_url:
-                                gh.post(  # type: ignore[unused-coroutine]
-                                    comment_url,
-                                    data={"body": permission_check.error_message},
-                                )
-                        return
+                kwargs["mention"] = MentionContext(
+                    commands=get_commands(event, username),
+                    user_permission=get_user_permission_from_event(event, gh),
+                    scope=event_scope,
+                )
 
                 func(event, gh, *args, **kwargs)