0
# Webhook Routing
1

2
Event routing system for handling GitHub webhook events with pattern matching and automatic dispatch to registered callback functions. The router supports both simple event type matching and data-based filtering for precise event handling.
3

4
## Capabilities
5

6
### Router Class
7

8
Central routing system for webhook event dispatch.
9

10
```python { .api }
11
from typing import Any, Awaitable, Callable, Dict, List, FrozenSet
12

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

15
class Router:
16
    """Route webhook events to registered functions."""
17
    
18
    def __init__(self, *other_routers: "Router") -> None:
19
        """
20
        Instantiate a new router, optionally inheriting from other routers.
21
        
22
        Parameters:
23
        - *other_routers: Existing routers to inherit routes from
24
        """
25
    
26
    def add(self, func: AsyncCallback, event_type: str, **data_detail: Any) -> None:
27
        """
28
        Add a new route for an event type and optional data filtering.
29
        
30
        Parameters:
31
        - func: Async callback function to handle the event
32
        - event_type: GitHub event type (e.g., "push", "pull_request")
33
        - **data_detail: Optional data-based filtering (max one level deep)
34
        
35
        Raises:
36
        - TypeError: If more than one data detail level is specified
37
        """
38
    
39
    def register(
40
        self, 
41
        event_type: str, 
42
        **data_detail: Any
43
    ) -> Callable[[AsyncCallback], AsyncCallback]:
44
        """
45
        Decorator to register a function for an event type.
46
        
47
        Parameters:
48
        - event_type: GitHub event type to register for
49
        - **data_detail: Optional data-based filtering
50
        
51
        Returns:
52
        - Decorator function that registers and returns the callback
53
        """
54
    
55
    def fetch(self, event: sansio.Event) -> FrozenSet[AsyncCallback]:
56
        """
57
        Return a set of functions registered for the given event.
58
        
59
        Parameters:
60
        - event: GitHub webhook event to match against
61
        
62
        Returns:
63
        - Frozen set of callback functions that match the event
64
        """
65
    
66
    async def dispatch(self, event: sansio.Event, *args: Any, **kwargs: Any) -> None:
67
        """
68
        Dispatch an event to all registered functions.
69
        
70
        Parameters:
71
        - event: GitHub webhook event
72
        - *args: Additional arguments to pass to callbacks
73
        - **kwargs: Additional keyword arguments to pass to callbacks
74
        """
75
```
76

77
## Usage Examples
78

79
### Basic Event Routing
80

81
```python
82
import asyncio
83
from gidgethub.routing import Router
84
from gidgethub.sansio import Event
85

86
router = Router()
87

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

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

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

105
router.add(handle_issues, "issues")
106

107
# Dispatch events
108
async def process_webhook(event_data):
109
    event = Event(event_data, event=event_data['event_type'], 
110
                  delivery_id="12345")
111
    await router.dispatch(event)
112
```
113

114
### Data-Based Event Filtering
115

116
```python
117
router = Router()
118

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

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

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

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

150
### Router Composition
151

152
```python
153
# Create specialized routers
154
pr_router = Router()
155
issue_router = Router()
156

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

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

165
# Combine routers
166
main_router = Router(pr_router, issue_router)
167

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

173
# The main router now has all routes from all routers
174
```
175

176
### Advanced Event Processing
177

178
```python
179
import asyncio
180
from gidgethub.routing import Router
181
from gidgethub.aiohttp import GitHubAPI
182
import aiohttp
183

184
router = Router()
185

186
@router.register("pull_request", action="opened")
187
async def auto_review_pr(event, gh: GitHubAPI):
188
    """Automatically request review for new PRs."""
189
    pr = event.data['pull_request']
190
    repo_name = event.data['repository']['full_name']
191
    pr_number = pr['number']
192
    
193
    # Request review from team
194
    await gh.post(
195
        f"/repos/{repo_name}/pulls/{pr_number}/requested_reviewers",
196
        data={"team_reviewers": ["core-team"]}
197
    )
198
    
199
    # Add labels based on file changes
200
    files = await gh.getiter(f"/repos/{repo_name}/pulls/{pr_number}/files")
201
    labels = []
202
    
203
    async for file in files:
204
        if file['filename'].endswith('.py'):
205
            labels.append("python")
206
        elif file['filename'].endswith(('.js', '.ts')):
207
            labels.append("javascript")
208
    
209
    if labels:
210
        await gh.post(
211
            f"/repos/{repo_name}/issues/{pr_number}/labels",
212
            data={"labels": labels}
213
        )
214

215
@router.register("issues", action="opened")
216
async def triage_issue(event, gh: GitHubAPI):
217
    """Auto-triage new issues."""
218
    issue = event.data['issue']
219
    repo_name = event.data['repository']['full_name']
220
    issue_number = issue['number']
221
    
222
    # Add triage label
223
    await gh.post(
224
        f"/repos/{repo_name}/issues/{issue_number}/labels",
225
        data={"labels": ["needs-triage"]}
226
    )
227
    
228
    # Assign to triage team if it's a bug report
229
    if "bug" in issue['title'].lower():
230
        await gh.post(
231
            f"/repos/{repo_name}/issues/{issue_number}/assignees",
232
            data={"assignees": ["triage-bot"]}
233
        )
234

235
# Process webhook with GitHub API client
236
async def handle_webhook(webhook_data, oauth_token):
237
    async with aiohttp.ClientSession() as session:
238
        gh = GitHubAPI(session, "webhook-bot/1.0", oauth_token=oauth_token)
239
        
240
        event = Event(
241
            webhook_data,
242
            event=webhook_data['event_type'],
243
            delivery_id=webhook_data['delivery_id']
244
        )
245
        
246
        # Dispatch with GitHub API client
247
        await router.dispatch(event, gh)
248
```
249

250
### Testing Event Routing
251

252
```python
253
import asyncio
254
from gidgethub.routing import Router
255
from gidgethub.sansio import Event
256

257
async def test_routing():
258
    router = Router()
259
    
260
    # Track which callbacks were called
261
    called_callbacks = []
262
    
263
    @router.register("push")
264
    async def handle_push(event):
265
        called_callbacks.append("push")
266
    
267
    @router.register("pull_request", action="opened")
268
    async def handle_pr_opened(event):
269
        called_callbacks.append("pr_opened")
270
    
271
    @router.register("pull_request", action="closed")
272
    async def handle_pr_closed(event):
273
        called_callbacks.append("pr_closed")
274
    
275
    # Test push event
276
    push_event = Event(
277
        {"repository": {"name": "test-repo"}},
278
        event="push",
279
        delivery_id="1"
280
    )
281
    await router.dispatch(push_event)
282
    assert "push" in called_callbacks
283
    
284
    # Test PR opened event
285
    pr_event = Event(
286
        {"action": "opened", "number": 1},
287
        event="pull_request", 
288
        delivery_id="2"
289
    )
290
    await router.dispatch(pr_event)
291
    assert "pr_opened" in called_callbacks
292
    assert "pr_closed" not in called_callbacks
293
    
294
    # Test event filtering
295
    callbacks = router.fetch(pr_event)
296
    assert len(callbacks) == 1  # Only pr_opened should match
297

298
asyncio.run(test_routing())
299
```
300

301
### Web Framework Integration
302

303
```python
304
# Flask example
305
from flask import Flask, request
306
import gidgethub.sansio
307
from gidgethub.routing import Router
308

309
app = Flask(__name__)
310
router = Router()
311

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

316
@app.route('/webhook', methods=['POST'])
317
def webhook():
318
    # Validate and parse webhook
319
    event = gidgethub.sansio.Event.from_http(
320
        dict(request.headers),
321
        request.data,
322
        secret=app.config['WEBHOOK_SECRET']
323
    )
324
    
325
    # Dispatch in background (Flask doesn't support async views directly)
326
    import threading
327
    def dispatch_async():
328
        asyncio.run(router.dispatch(event))
329
    
330
    threading.Thread(target=dispatch_async).start()
331
    return '', 200
332

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

337
app = FastAPI()
338
router = Router()
339

340
@app.post("/webhook")
341
async def webhook(request: Request):
342
    body = await request.body()
343
    headers = dict(request.headers)
344
    
345
    event = gidgethub.sansio.Event.from_http(
346
        headers, body, secret="webhook_secret"
347
    )
348
    
349
    await router.dispatch(event)
350
    return {"status": "ok"}
351
```
352

353
## Event Types
354

355
Common GitHub webhook event types that can be registered:
356

357
- `"push"` - Repository push events
358
- `"pull_request"` - Pull request events (actions: opened, closed, synchronize, etc.)
359
- `"issues"` - Issue events (actions: opened, closed, labeled, etc.)
360
- `"issue_comment"` - Issue and PR comments
361
- `"pull_request_review"` - PR review events
362
- `"release"` - Release events
363
- `"fork"` - Repository fork events
364
- `"star"` - Repository star events
365
- `"watch"` - Repository watch events
366
- `"deployment"` - Deployment events
367
- `"status"` - Commit status updates
368
- `"check_run"` - Check run events
369
- `"workflow_run"` - GitHub Actions workflow events
370

371
## Types
372

373
```python { .api }
374
from typing import Any, Awaitable, Callable, Dict, List, FrozenSet
375
from gidgethub.sansio import Event
376

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

380
# Router internal types
381
_ShallowRoutes = Dict[str, List[AsyncCallback]]
382
_DeepRoutes = Dict[str, Dict[str, Dict[Any, List[AsyncCallback]]]]
383
```