Function github_webhook

app/routers/github.py:437–609

(
    request: Request,
    webhook_data: tuple[dict, str] = Depends(_verify_github_webhook),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client),
    queue: ArqRedis = Depends(get_queue),
)

@router.post("/webhook", name="github_webhook")
async def github_webhook(
    request: Request,
    webhook_data: tuple[dict, str] = Depends(_verify_github_webhook),
    db: AsyncSession = Depends(get_db),
    redis_client: Redis = Depends(get_redis_client),
    queue: ArqRedis = Depends(get_queue),
):
    try:
        data, event = webhook_data

        logger.info(f"Received GitHub webhook event: {event}")

        match event:
            case "installation":
                if data["action"] in ["deleted", "suspended", "unsuspended"]:
                    # App uninstalled or suspended
                    status = (
                        "active" if data["action"] == "unsuspended" else data["action"]
                    )
                    await db.execute(
                        update(GithubInstallation)
                        .where(
                            GithubInstallation.installation_id
                            == data["installation"]["id"]
                        )
                        .values(status=status)
                    )
                    await db.commit()
                    logger.info(
                        f"Installation {data['installation']['id']} for {data['installation']['account']['login']} is {data['action']}"
                    )

                elif data["action"] == "created":
                    # App installed
                    installation_id = data["installation"]["id"]
                    github_service = get_github_service()
                    token_data = await github_service.get_installation_access_token(
                        installation_id
                    )
                    installation = GithubInstallation(
                        installation_id=installation_id,
                        token=token_data["token"],
                        token_expires_at=datetime.fromisoformat(
                            token_data["expires_at"].replace("Z", "+00:00")
                        ),
                    )
                    await db.merge(installation)
                    await db.commit()
                    logger.info(
                        f"Installation {installation_id} for {data['installation']['account']['login']} created"
                    )

            case "installation_target":
                if data["action"] == "renamed":
                    # Installation account is renamed (not used)
                    pass

            case "installation_repositories":