GitHub

pip install "flyteplugins-github[app]"

import flyte
from flyte.extras.webhooks import WebhookAppEnvironment, WebhookEvent, run_once
from flyteplugins.github import GitHubProvider, events

# Serves the receiver at /webhook/github and a setup dashboard at /.
# GITHUB_WEBHOOK_SECRET is mounted automatically.
#
# `scopes` lists the repositories to act on. Deliveries from other
# repositories, or with no repository, are acknowledged but not dispatched.
app_env = WebhookAppEnvironment(
    name="github-webhooks",
    providers=[GitHubProvider()],
    scopes=["octo/repo"],
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-github[app]"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@app_env.on_event(events.PullRequest.OPENED)
async def on_pull_request_opened(event: WebhookEvent) -> dict:
    """Launch triage once per pull request.

    Retried and resent deliveries have the same `dedupe_key()`, so `run_once`
    launches only one run for them. Use `await run_once.aio(...)`: the blocking
    form stalls the app's event loop, and GitHub times out a delivery after
    ten seconds.
    """
    import flyte.remote as remote

    task = remote.Task.get(name="github-triage.triage_pr", auto_version="latest")
    result = await run_once.aio(
        task,
        key=event.dedupe_key(),
        repo=event.scope,
        number=event.payload["pull_request"]["number"],
    )
    if not result.created:
        # An earlier delivery of this event already launched a run.
        return {"skipped": result.run.name, "url": result.run.url}
    return {"run": result.run.name, "url": result.run.url}

if __name__ == "__main__":
    flyte.init_from_config(root_dir=pathlib.Path(__file__).parent)
    deployment = flyte.serve(app_env)
    # The dashboard shows the payload URL to enter in GitHub under
    # Settings -> Webhooks -> Add webhook.
    print(f"Setup dashboard: {deployment.url}")

env = flyte.TaskEnvironment(
    name="github-triage",
    image=flyte.Image.from_debian_base().with_pip_packages("PyGithub"),
    # Token for reading the pull request. The webhook secret is mounted on
    # the receiver app, not here.
    secrets=[flyte.Secret(key="github-token", as_env_var="GITHUB_TOKEN")],
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@env.task
async def triage_pr(repo: str, number: int) -> str:
    """Label a pull request by size."""
    import os

    from github import Auth, Github

    client = Github(auth=Auth.Token(os.environ["GITHUB_TOKEN"]))
    pull = client.get_repo(repo).get_pull(number)
    changed = pull.additions + pull.deletions
    label = "size/s" if changed < 50 else "size/m" if changed < 500 else "size/l"
    pull.add_to_labels(label)
    return label

review_env = flyte.TaskEnvironment(
    name="github-review",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-github[review]"),
    secrets=[flyte.Secret(key="github-token", as_env_var="GITHUB_TOKEN")],
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@review_env.task
async def gated_merge(repo: str, number: int) -> str:
    """Wait for a review decision in the Flyte UI, then act on it.

    `review_pr` creates a condition carrying the pull request's metadata and
    waits for a reviewer to answer it. The condition is stored on the backend,
    so the run survives restarts while it waits.
    """
    from flyteplugins.github import review_pr

    decision = await review_pr(repo, number, instructions="Block on missing tests.")
    if decision.is_approved:
        return f"approved by {decision.reviewer}: {decision.summary}"
    blocking = ", ".join(c.path for c in decision.blocking_comments)
    return f"{decision.verdict}: {decision.summary} ({blocking})"

agent_env = flyte.TaskEnvironment(
    name="github-agent",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-github[auth]"),
    # Each key maps to the environment variable `mint_installation_token`
    # reads (github-app-id -> GITHUB_APP_ID), so `as_env_var=` isn't needed.
    secrets=[
        flyte.Secret(key="github-app-id"),
        flyte.Secret(key="github-app-installation-id"),
        flyte.Secret(key="github-app-private-key"),
    ],
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@agent_env.task
async def clone_at_head(repo: str) -> str:
    """Build an authenticated clone URL with a GitHub App installation token.

    The token expires after one hour.
    """
    import asyncio

    from flyteplugins.github import clone_url, mint_installation_token

    # mint_installation_token is synchronous; run it off the event loop.
    # It returns None if the App credentials are missing.
    token = await asyncio.to_thread(mint_installation_token)
    url = clone_url(repo, token)
    # Never return or log the token itself.
    return url.replace(token, "***") if token else url

flyte create secret github-app-id --value 1234567
flyte create secret github-app-installation-id --value 87654321
flyte create secret github-app-private-key --from-file ~/Downloads/app.private-key.pem

# A separate environment with no secrets, so the replay runs before any
# secret is created.
replay_env = flyte.TaskEnvironment(
    name="github-replay",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-github"),
    resources=flyte.Resources(cpu=1, memory="512Mi"),
)

@replay_env.task
async def replay_sample_delivery() -> dict[str, str]:
    """Verify and parse the sample delivery bundled with the plugin.

    `SAMPLE_DELIVERY` is a recorded GitHub payload and a function that signs
    it. The body is real; the signature header is generated by the plugin.
    """
    import flyteplugins.github as plugin

    secret = "a-test-signing-secret"
    sign, body = plugin.SAMPLE_DELIVERY
    headers = sign(body, secret)

    assert plugin.verify(body, headers, secret), "a correctly signed delivery must verify"
    assert not plugin.verify(body, headers, "wrong-secret"), "a bad signature must not"

    event = plugin.parse(headers, body)
    return {
        # Matched by `on_event`. `events.PullRequest.OPENED` equals this string.
        "qualified_type": event.qualified_type,
        # Matched against `scopes`.
        "scope": event.scope or "",
        "title": event.title or "",
        "actor": event.actor or "",
        # The key `run_once` deduplicates on.
        "dedupe_key": event.dedupe_key(),
    }

flyte run --local github_tasks.py replay_sample_delivery