PowerBI自定义查询中GitHub API限流问题的解决方案求助

一、切换到GitHub GraphQL API（大幅减少请求数） #

GraphQL允许单次请求获取多个仓库的多类数据（Issues、Pulls、Commits），比REST API的请求量至少减少70%以上，完美适配你的多仓库场景。以下是Power Query M语言的实现示例：

1. 复用已有的组织-仓库列表，批量构造GraphQL查询：

let
    // 复用你的表1数据
    OrgRepos = 表1,
    // 按组织分组生成查询片段
    GroupedOrgs = Table.Group(OrgRepos, {"组织名称"}, {
        {"QueryFragments", each 
            Text.Combine(List.Transform([仓库名称], (repo) => 
                "\"" & repo & "\": {
                    issues(states: [OPEN, CLOSED], since: \"#date#\") {
                        nodes { title, author { login }, createdAt, closedAt }
                    },
                    pullRequests(states: [OPEN, CLOSED, MERGED], since: \"#date#\") {
                        nodes { title, author { login }, mergedAt, createdAt }
                    },
                    defaultBranchRef {
                        target {
                            ... on Commit {
                                history(since: \"#date#\") {
                                    nodes { author { user { login } }, committedDate }
                                }
                            }
                        }
                    }
                }"
            ), ", ")
        }
    }),
    // 设置近2个月的起始日期
    StartDate = DateTime.ToText(DateTime.LocalNow() - #duration(60,0,0,0), "yyyy-MM-ddTHH:mm:ssZ"),
    // 替换日期变量，生成完整查询
    BuildQueries = Table.AddColumn(GroupedOrgs, "FullQuery", each 
        "query {
            organization(login: """ & [组织名称] & """) {
                " & Text.Replace([QueryFragments], "#date#", StartDate) & "#(lf)
            }
        }"
    ),
    // 调用GraphQL API（替换为你的个人访问令牌）
    Token = "ghp_yourpersonalaccesstoken",
    CallAPI = Table.AddColumn(BuildQueries, "APIResponse", each 
        Json.Document(Web.Contents("https://api.github.com/graphql", [
            Headers = [
                #"Authorization" = "Bearer " & Token,
                #"Content-Type" = "application/json"
            ],
            Content = Text.ToBinary(Json.FromValue([query = [FullQuery]]))
        ]))
    ),
    // 解析并展开数据（后续逻辑根据需求调整）
    ParseResponse = Table.ExpandRecordColumn(CallAPI, "APIResponse", {"data"}, {"data"}),
    ExpandOrgData = Table.ExpandRecordColumn(ParseResponse, "data", {"organization"}, {"organization"}),
    FinalTable = Table.ExpandRecordColumn(ExpandOrgData, "organization", 
        List.Distinct(List.Combine(List.Transform(ExpandOrgData[organization], Record.FieldNames))), 
        List.Distinct(List.Combine(List.Transform(ExpandOrgData[organization], Record.FieldNames)))
    )
in
    FinalTable

注：GraphQL和REST的小时限流都是5000次，但单次请求能获取的信息量是REST的数倍，实际覆盖的仓库数量会大幅提升。

二、优化Power Query REST API请求效率 #

如果暂时不想切换GraphQL，可通过以下方式减少调用次数：

• 批量请求复用：利用GitHub REST的批量接口（如批量获取仓库信息），避免逐个仓库调用。
• 添加指数退避重试：在M语言中处理403限流错误，自动重试：

// 带重试的API调用函数
RetryAPICall = (url as text, token as text) as any =>
    let
        Retry = (attempt as number) as any =>
            let
                Response = try Web.Contents(url, [Headers = [Authorization = "token " & token]]) otherwise null,
                Result = if Response = null then
                    if attempt < 3 then
                        let
                            Wait = Duration.FromSeconds(2^attempt),
                            WaitDone = List.Generate(() => DateTime.LocalNow(), each _ < DateTime.LocalNow() + Wait, each _)
                        in
                            Retry(attempt + 1)
                    else
                        error "API调用重试失败"
                    else
                        Json.Document(Response)
            in
                Result
    in
        Retry(0)

• 启用Power Query缓存：在选项中开启“允许数据预览缓存”，避免重复请求相同的仓库数据。

三、用Python做中间层处理请求与限流 #

Python的第三方库能更灵活处理限流和批量请求，处理完数据后导入PowerBI：

1. 编写Python脚本获取并整理数据：

import requests
from tenacity import retry, stop_after_attempt, wait_exponential
import pandas as pd

GITHUB_TOKEN = "ghp_yourpersonalaccesstoken"
HEADERS = {"Authorization": f"token {GITHUB_TOKEN}"}
START_DATE = (pd.Timestamp.now() - pd.Timedelta(days=60)).isoformat() + "Z"

# 读取组织-仓库列表（可直接从PowerBI导出为CSV）
org_repos = pd.read_csv("org_repos.csv")

# 带指数退避的请求函数
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_data(url):
    response = requests.get(url, headers=HEADERS)
    if response.status_code == 403:
        raise Exception("Rate limit exceeded")
    response.raise_for_status()
    return response.json()

# 批量拉取数据
all_records = []
for _, row in org_repos.iterrows():
    org, repo = row["组织名称"], row["仓库名称"]
    
    # 获取Issues
    issues = fetch_data(f"https://api.github.com/repos/{org}/{repo}/issues?state=all&since={START_DATE}")
    all_records.extend([{
        "组织": org, "仓库": repo, "类型": "Issue",
        "作者": issue["user"]["login"] if issue["user"] else None,
        "创建时间": issue["created_at"], "关闭时间": issue["closed_at"]
    } for issue in issues])
    
    # 获取Pulls
    pulls = fetch_data(f"https://api.github.com/repos/{org}/{repo}/pulls?state=all&since={START_DATE}")
    all_records.extend([{
        "组织": org, "仓库": repo, "类型": "PullRequest",
        "作者": pull["user"]["login"] if pull["user"] else None,
        "创建时间": pull["created_at"], "合并时间": pull["merged_at"]
    } for pull in pulls])
    
    # 获取Commits
    commits = fetch_data(f"https://api.github.com/repos/{org}/{repo}/commits?since={START_DATE}")
    all_records.extend([{
        "组织": org, "仓库": repo, "类型": "Commit",
        "作者": commit["commit"]["author"]["name"],
        "提交时间": commit["commit"]["committer"]["date"]
    } for commit in commits])

# 转为DataFrame供PowerBI读取
df = pd.DataFrame(all_records)
print(df)

2. 在PowerBI中通过「获取数据 > Python脚本」导入上述脚本，直接加载处理好的数据。

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