Welcome back! So far, you have learned how to set up a FastAPI backend, manage changesets, and handle errors and validation in your Code Review Assistant project. In this lesson, we will take the next step by connecting your application to GitHub using webhooks.

A webhook is a way for one application to send real-time data to another application when a specific event happens. In our case, we want GitHub to notify our Code Review Assistant whenever someone opens or updates a pull request. This allows us to automatically start a code review process as soon as new code is submitted.

By the end of this lesson, you will know how to receive GitHub webhook events, verify their authenticity, extract pull request data, and trigger your code review workflow — all automatically.

Before we dive in, let’s quickly remind ourselves how FastAPI handles HTTP requests. In previous lessons, you learned how to define endpoints using FastAPI and how to read data from incoming requests.

For example, to create a POST endpoint and read the request body, you might write:

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/example")
async def example_endpoint(request: Request):
    data = await request.json()
    return {"received": data}

Here, @app.post("/example") defines a POST endpoint. The request: Request parameter allows you to access the raw request, and await request.json() reads the JSON body sent by the client.

This pattern is the foundation for receiving webhook events, which are just HTTP POST requests sent by GitHub to your application.

Let’s start by creating a FastAPI endpoint that can receive webhook events from GitHub.

First, we need to define a POST endpoint. Since GitHub sends the webhook data as a raw payload, we will read the body as bytes:

from fastapi import APIRouter, Request

router = APIRouter()

@router.post("/webhook/github")
async def github_webhook(request: Request):
    payload = await request.body()
    # We will process the payload in the next steps
    return {"status": "received"}

• @router.post("/webhook/github") creates a new POST endpoint at /webhook/github.
• payload = await request.body() reads the raw bytes sent by GitHub. This is important because we may need the exact bytes for signature verification.

At this point, your application can receive webhook events from GitHub. However, we need to make sure these events are actually coming from GitHub and not from someone else.

Security is important. GitHub allows you to set a secret when configuring a webhook. When GitHub sends a webhook event, it includes a signature in the headers. Your application should verify this signature to make sure the request is genuine.

Here’s how you can verify the signature using HMAC and SHA-256:

import hmac
import hashlib
from fastapi import Header, HTTPException

def verify_github_signature(payload: bytes, signature: str, secret: str) -> bool:
    if not signature or not signature.startswith('sha256='):
        return False

    expected = 'sha256=' + hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, signature)

• payload is the raw request body.
• signature is the value from the X-Hub-Signature-256 header.
• secret is the webhook secret you set in GitHub.

To use this in your endpoint:

@router.post("/webhook/github")
async def github_webhook(
    request: Request,
    x_hub_signature_256: str = Header(None)
):
    payload = await request.body()
    GITHUB_SECRET = "your-secret"  # Replace with your actual secret

    if not verify_github_signature(payload, x_hub_signature_256, GITHUB_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Continue processing the payload
    return {"status": "received"}

If the signature is invalid, the endpoint returns a 401 Unauthorized error. This helps protect your application from unwanted or malicious requests.

Once you have verified the webhook, you need to extract the pull request information from the payload. GitHub sends the data as JSON, so you can parse it like this:

import json

@router.post("/webhook/github")
async def github_webhook(
    request: Request,
    x_hub_signature_256: str = Header(None)
):
    payload = await request.body()
    # Assume signature is already verified

    event = json.loads(payload)

    if event.get('action') in ['opened', 'synchronize']:
        pr = event['pull_request']
        title = pr['title']
        description = pr.get('body', '')
        author = pr['user']['login']
        diff_url = pr['diff_url']

        print("Pull Request Title:", title)
        print("Description:", description)
        print("Author:", author)
        print("Diff URL:", diff_url)

    return {"status": "received"}

• event = json.loads(payload) parses the JSON payload.
• We check if the action is 'opened' or 'synchronize', which are the events we care about for new or updated pull requests.
• We extract the pull request title, description, author, and the URL to the diff.

Example Output:

Pull Request Title: Add new feature
Description: This PR adds a new feature to the application
Author: developer123
Diff URL: https://github.com/repo/pull/123.diff

This information is essential for tracking changes and starting the review process.

Now that you have the pull request data, you need to save it in your database and schedule a code review. In previous lessons, you learned how to use SQLAlchemy models for changesets. Here’s how you might use them in this context:

from sqlalchemy.orm import Session
from fastapi import Depends, BackgroundTasks

# Dummy session and models for demonstration
def get_session():
    class DummySession:
        def add(self, obj): pass
        def flush(self): pass
        def commit(self): pass
        def close(self): pass
    return DummySession()

class Changeset:
    def __init__(self, title, description, author, status='pending'):
        self.title = title
        self.description = description
        self.author = author
        self.status = status
        self.id = 1  # Dummy ID

@router.post("/webhook/github")
async def github_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_hub_signature_256: str = Header(None),
    db: Session = Depends(get_session)
):
    payload = await request.body()
    # Assume signature is already verified
    event = json.loads(payload)

    if event.get('action') in ['opened', 'synchronize']:
        pr = event['pull_request']
        changeset = Changeset(
            title=pr['title'],
            description=pr.get('body', ''),
            author=pr['user']['login'],
            status='pending'
        )
        db.add(changeset)
        db.flush()
        db.commit()

        # Schedule background review
        background_tasks.add_task(
            process_pr_review,
            changeset.id,
            pr['number'],
            event['repository']['full_name']
        )

    return {"status": "received"}

• We create a new Changeset object with the pull request data.
• We add and commit it to the database.
• We use background_tasks.add_task() to schedule the review process without blocking the webhook response.

This approach ensures that your application quickly responds to GitHub and then processes the review in the background.

In this lesson, you learned how to connect your Code Review Assistant to GitHub using webhooks. You saw how to:

• Set up a FastAPI endpoint to receive webhook events.
• Verify webhook signatures for security.
• Parse pull request data from the webhook payload.
• Store changeset information and schedule background review tasks.

These steps allow your application to automatically react to new or updated pull requests, making your code review process faster and more reliable.

In the next practice exercises, you will get hands-on experience with each of these steps. You will implement and test webhook handling, signature verification, and changeset storage. This will help you solidify your understanding and prepare you for more advanced integrations in the future.