如何通过AWS Lambda调用GitHub API提交PR更新仓库代号文件？

Absolutely, you can pull this off seamlessly with the GitHub API—let’s break this down step by step, since I’ve built similar sync workflows before.

Prerequisites First #

Before writing any Lambda code, lock in these foundational pieces:

• GitHub Personal Access Token (PAT): Create a PAT with the repo scope (this lets it create branches, commit changes, and open PRs). Store this securely in AWS Secrets Manager (never hardcode it!) and grant your Lambda execution role permission to retrieve it.
• Lambda Execution Role: Attach policies that allow:
  • Read access to Secrets Manager (to fetch the GitHub PAT)
  • Read/write access to your target S3 bucket
  • Network access (if your API is private, configure VPC access or a NAT gateway)
• GitHub Repo Access: Ensure the PAT’s linked account has write access to the repo (or fork it if you don’t own the original).

Lambda Function Core Workflow #

Split your Lambda into 4 distinct, error-handled steps to avoid half-completed workflows:

Step 1: Fetch Latest Codenames from the API #

Use a HTTP client (like Python’s requests library) to pull and parse the API response. Add retries for transient failures (e.g., network blips):

import requests
import json

def fetch_latest_codenames(api_url):
    try:
        response = requests.get(api_url, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        # Log to CloudWatch and raise to trigger Lambda retry
        print(f"Failed to fetch codenames: {str(e)}")
        raise

Step 2: Save Updated List to S3 #

Write the fetched JSON to your S3 bucket’s secret-codenames.json file with the correct content type:

import boto3

def save_to_s3(bucket_name, codenames_data):
    s3 = boto3.client('s3')
    try:
        s3.put_object(
            Bucket=bucket_name,
            Key='secret-codenames.json',
            Body=json.dumps(codenames_data, indent=2),
            ContentType='application/json'
        )
        print("Successfully updated S3 file")
    except Exception as e:
        print(f"Failed to write to S3: {str(e)}")
        raise

Step 3: Sync Changes to GitHub via API #

This is the most involved part—use the GitHub API to create a branch, update the file, and open a PR. Here’s a simplified code snippet for this flow:

import os
from datetime import datetime

def update_github_repo(pat, repo_owner, repo_name, codenames_data):
    github_api_base = "https://api.github.com"
    headers = {
        "Authorization": f"token {pat}",
        "Accept": "application/vnd.github.v3+json"
    }
    branch_suffix = datetime.now().strftime("%Y%m%d%H%M%S")
    new_branch_name = f"update-secret-codenames-{branch_suffix}"

    # Get master branch's latest commit SHA
    master_branch_url = f"{github_api_base}/repos/{repo_owner}/{repo_name}/branches/master"
    master_resp = requests.get(master_branch_url, headers=headers)
    master_resp.raise_for_status()
    master_commit_sha = master_resp.json()["commit"]["sha"]

    # Create new feature branch
    create_branch_url = f"{github_api_base}/repos/{repo_owner}/{repo_name}/git/refs"
    create_branch_data = {
        "ref": f"refs/heads/{new_branch_name}",
        "sha": master_commit_sha
    }
    branch_resp = requests.post(create_branch_url, headers=headers, json=create_branch_data)
    branch_resp.raise_for_status()

    # Get current file's blob SHA (required for updates)
    file_url = f"{github_api_base}/repos/{repo_owner}/{repo_name}/contents/restricted/secret-codenames.json?ref=master"
    file_resp = requests.get(file_url, headers=headers)
    file_resp.raise_for_status()
    current_blob_sha = file_resp.json()["sha"]

    # Update the file in the new branch
    update_file_url = f"{github_api_base}/repos/{repo_owner}/{repo_name}/contents/restricted/secret-codenames.json"
    update_file_data = {
        "message": "Update secret codenames from latest API data",
        "content": json.dumps(codenames_data, indent=2).encode("base64").decode("utf-8"),
        "sha": current_blob_sha,
        "branch": new_branch_name
    }
    update_resp = requests.put(update_file_url, headers=headers, json=update_file_data)
    update_resp.raise_for_status()

    # Open the Pull Request
    pr_url = f"{github_api_base}/repos/{repo_owner}/{repo_name}/pulls"
    pr_data = {
        "title": f"Update secret codenames - {datetime.now().strftime('%Y-%m-%d')}",
        "body": "Automated update from AWS Lambda: Fetched latest codenames from external API.",
        "head": new_branch_name,
        "base": "master"
    }
    pr_resp = requests.post(pr_url, headers=headers, json=pr_data)
    pr_resp.raise_for_status()
    print(f"Successfully created PR: {pr_resp.json()['html_url']}")

Step 4: Add Robustness Checks #

• Skip duplicate updates: Compare the fetched codenames with the current GitHub file content. If they’re identical, skip the GitHub sync to avoid noise.
• Handle rate limits: Check the X-RateLimit-Remaining header in GitHub API responses and add exponential backoff if you hit limits.
• Log everything: Use CloudWatch Logs to track each step’s success/failure for easy debugging.

Testing & Validation #

• Manual trigger first: Test your Lambda manually to verify the S3 file updates, branch gets created, and PR appears in GitHub.
• Set up alerts: Use CloudWatch Alarms to notify you if the Lambda fails to execute, so you can fix issues before stale data becomes a problem.

内容的提问来源于stack exchange，提问作者larrydalmeida