Webhook Routing

from typing import Any, Awaitable, Callable, Dict, List, FrozenSet

AsyncCallback = Callable[..., Awaitable[None]]

class Router:
    """Route webhook events to registered functions."""
    
    def __init__(self, *other_routers: "Router") -> None:
        """
        Instantiate a new router, optionally inheriting from other routers.
        
        Parameters:
        - *other_routers: Existing routers to inherit routes from
        """
    
    def add(self, func: AsyncCallback, event_type: str, **data_detail: Any) -> None:
        """
        Add a new route for an event type and optional data filtering.
        
        Parameters:
        - func: Async callback function to handle the event
        - event_type: GitHub event type (e.g., "push", "pull_request")
        - **data_detail: Optional data-based filtering (max one level deep)
        
        Raises:
        - TypeError: If more than one data detail level is specified
        """
    
    def register(
        self, 
        event_type: str, 
        **data_detail: Any
    ) -> Callable[[AsyncCallback], AsyncCallback]:
        """
        Decorator to register a function for an event type.
        
        Parameters:
        - event_type: GitHub event type to register for
        - **data_detail: Optional data-based filtering
        
        Returns:
        - Decorator function that registers and returns the callback
        """
    
    def fetch(self, event: sansio.Event) -> FrozenSet[AsyncCallback]:
        """
        Return a set of functions registered for the given event.
        
        Parameters:
        - event: GitHub webhook event to match against
        
        Returns:
        - Frozen set of callback functions that match the event
        """
    
    async def dispatch(self, event: sansio.Event, *args: Any, **kwargs: Any) -> None:
        """
        Dispatch an event to all registered functions.
        
        Parameters:
        - event: GitHub webhook event
        - *args: Additional arguments to pass to callbacks
        - **kwargs: Additional keyword arguments to pass to callbacks
        """

import asyncio
from gidgethub.routing import Router
from gidgethub.sansio import Event

router = Router()

# Register for all push events
@router.register("push")
async def handle_push(event):
    print(f"Push to {event.data['repository']['name']}")
    print(f"Commits: {len(event.data['commits'])}")

# Register for pull request events
@router.register("pull_request")
async def handle_pull_request(event):
    action = event.data['action']
    pr_number = event.data['number']
    print(f"Pull request #{pr_number} was {action}")

# Manual registration (equivalent to decorator)
async def handle_issues(event):
    print(f"Issue event: {event.data['action']}")

router.add(handle_issues, "issues")

# Dispatch events
async def process_webhook(event_data):
    event = Event(event_data, event=event_data['event_type'], 
                  delivery_id="12345")
    await router.dispatch(event)

router = Router()

# Handle only pull request "opened" actions
@router.register("pull_request", action="opened")
async def handle_pr_opened(event):
    pr = event.data['pull_request']
    print(f"New PR: {pr['title']}")
    print(f"Author: {pr['user']['login']}")

# Handle only pull request "closed" actions
@router.register("pull_request", action="closed")
async def handle_pr_closed(event):
    pr = event.data['pull_request']
    if pr['merged']:
        print(f"PR merged: {pr['title']}")
    else:
        print(f"PR closed without merge: {pr['title']}")

# Handle push events to main branch only
@router.register("push", ref="refs/heads/main")
async def handle_main_push(event):
    repo = event.data['repository']['name']
    commits = len(event.data['commits'])
    print(f"Push to main branch of {repo}: {commits} commits")

# Handle issue events with specific labels
@router.register("issues", action="labeled")
async def handle_issue_labeled(event):
    issue = event.data['issue']
    label = event.data['label']['name']
    print(f"Issue #{issue['number']} labeled with '{label}'")

# Create specialized routers
pr_router = Router()
issue_router = Router()

@pr_router.register("pull_request", action="opened")
async def handle_new_pr(event):
    print("New PR opened")

@issue_router.register("issues", action="opened")
async def handle_new_issue(event):
    print("New issue opened")

# Combine routers
main_router = Router(pr_router, issue_router)

# Add additional routes to combined router
@main_router.register("push")
async def handle_push(event):
    print("Push event")

# The main router now has all routes from all routers

import asyncio
from gidgethub.routing import Router
from gidgethub.aiohttp import GitHubAPI
import aiohttp

router = Router()

@router.register("pull_request", action="opened")
async def auto_review_pr(event, gh: GitHubAPI):
    """Automatically request review for new PRs."""
    pr = event.data['pull_request']
    repo_name = event.data['repository']['full_name']
    pr_number = pr['number']
    
    # Request review from team
    await gh.post(
        f"/repos/{repo_name}/pulls/{pr_number}/requested_reviewers",
        data={"team_reviewers": ["core-team"]}
    )
    
    # Add labels based on file changes
    files = await gh.getiter(f"/repos/{repo_name}/pulls/{pr_number}/files")
    labels = []
    
    async for file in files:
        if file['filename'].endswith('.py'):
            labels.append("python")
        elif file['filename'].endswith(('.js', '.ts')):
            labels.append("javascript")
    
    if labels:
        await gh.post(
            f"/repos/{repo_name}/issues/{pr_number}/labels",
            data={"labels": labels}
        )

@router.register("issues", action="opened")
async def triage_issue(event, gh: GitHubAPI):
    """Auto-triage new issues."""
    issue = event.data['issue']
    repo_name = event.data['repository']['full_name']
    issue_number = issue['number']
    
    # Add triage label
    await gh.post(
        f"/repos/{repo_name}/issues/{issue_number}/labels",
        data={"labels": ["needs-triage"]}
    )
    
    # Assign to triage team if it's a bug report
    if "bug" in issue['title'].lower():
        await gh.post(
            f"/repos/{repo_name}/issues/{issue_number}/assignees",
            data={"assignees": ["triage-bot"]}
        )

# Process webhook with GitHub API client
async def handle_webhook(webhook_data, oauth_token):
    async with aiohttp.ClientSession() as session:
        gh = GitHubAPI(session, "webhook-bot/1.0", oauth_token=oauth_token)
        
        event = Event(
            webhook_data,
            event=webhook_data['event_type'],
            delivery_id=webhook_data['delivery_id']
        )
        
        # Dispatch with GitHub API client
        await router.dispatch(event, gh)

import asyncio
from gidgethub.routing import Router
from gidgethub.sansio import Event

async def test_routing():
    router = Router()
    
    # Track which callbacks were called
    called_callbacks = []
    
    @router.register("push")
    async def handle_push(event):
        called_callbacks.append("push")
    
    @router.register("pull_request", action="opened")
    async def handle_pr_opened(event):
        called_callbacks.append("pr_opened")
    
    @router.register("pull_request", action="closed")
    async def handle_pr_closed(event):
        called_callbacks.append("pr_closed")
    
    # Test push event
    push_event = Event(
        {"repository": {"name": "test-repo"}},
        event="push",
        delivery_id="1"
    )
    await router.dispatch(push_event)
    assert "push" in called_callbacks
    
    # Test PR opened event
    pr_event = Event(
        {"action": "opened", "number": 1},
        event="pull_request", 
        delivery_id="2"
    )
    await router.dispatch(pr_event)
    assert "pr_opened" in called_callbacks
    assert "pr_closed" not in called_callbacks
    
    # Test event filtering
    callbacks = router.fetch(pr_event)
    assert len(callbacks) == 1  # Only pr_opened should match

asyncio.run(test_routing())

# Flask example
from flask import Flask, request
import gidgethub.sansio
from gidgethub.routing import Router

app = Flask(__name__)
router = Router()

@router.register("push")
async def handle_push(event):
    print(f"Push to {event.data['repository']['name']}")

@app.route('/webhook', methods=['POST'])
def webhook():
    # Validate and parse webhook
    event = gidgethub.sansio.Event.from_http(
        dict(request.headers),
        request.data,
        secret=app.config['WEBHOOK_SECRET']
    )
    
    # Dispatch in background (Flask doesn't support async views directly)
    import threading
    def dispatch_async():
        asyncio.run(router.dispatch(event))
    
    threading.Thread(target=dispatch_async).start()
    return '', 200

# FastAPI example (native async support)
from fastapi import FastAPI, Request
import gidgethub.sansio

app = FastAPI()
router = Router()

@app.post("/webhook")
async def webhook(request: Request):
    body = await request.body()
    headers = dict(request.headers)
    
    event = gidgethub.sansio.Event.from_http(
        headers, body, secret="webhook_secret"
    )
    
    await router.dispatch(event)
    return {"status": "ok"}

from typing import Any, Awaitable, Callable, Dict, List, FrozenSet
from gidgethub.sansio import Event

# Callback function type for webhook handlers
AsyncCallback = Callable[..., Awaitable[None]]

# Router internal types
_ShallowRoutes = Dict[str, List[AsyncCallback]]
_DeepRoutes = Dict[str, Dict[str, Dict[Any, List[AsyncCallback]]]]