如何用Python获取GitHub仓库最新活动并实现事件监听装饰器

你要做的装饰器本质是回调注册器，核心逻辑是把被装饰的函数注册为对应事件的回调函数，当监听到对应事件触发时，自动把事件数据传给回调执行。
两种主流的事件监听方案可选：

• 主动轮询：不需要公网服务器，本地就能跑，适合小体量、实时性要求不高（分钟级）的场景，通过定期调用GitHub API对比本地缓存的最新状态判断有没有新增事件
• Webhook推送：实时性高（秒级），需要你有公网可访问的服务器接收GitHub的POST请求，适合生产环境使用

以下是轮询模式的可运行示例，依赖PyGitHub封装GitHub API调用，你也可以替换为原生REST API实现：

from github import Github
import time
from typing import Callable, List, Dict
from dataclasses import dataclass

# 定义事件类型枚举
@dataclass
class EventType:
    NEW_COMMIT = "new_commit"
    NEW_ISSUE = "new_issue"
    NEW_ISSUE_COMMENT = "new_issue_comment"
    NEW_PR = "new_pull_request"
    NEW_DISCUSSION = "new_discussion"
    ALL_UPDATES = "all_updates"

class Repo:
    def __init__(self, repo_name: str, github_token: str):
        self.client = Github(github_token)
        self.repo = self.client.get_repo(repo_name)
        # 存储各事件的回调函数列表
        self.callbacks: Dict[str, List[Callable]] = {k: [] for k in EventType.__dict__.values() if not k.startswith('_')}
        # 存储各事件的最新状态缓存，用来判断有没有新事件
        self.latest_state = {
            "last_commit_sha": None,
            "last_issue_number": 0,
            "last_issue_comment_id": 0,
            "last_pr_number": 0,
            "last_discussion_id": 0
        }
        # 初始化缓存状态
        self._init_cache()

    def _init_cache(self):
        # 初始化时拉取当前最新的状态，避免启动时把历史事件全部推送一遍
        latest_commit = self.repo.get_commits()[0]
        self.latest_state["last_commit_sha"] = latest_commit.sha
        issues = self.repo.get_issues(state="all", sort="created", direction="desc")
        if issues.totalCount > 0:
            self.latest_state["last_issue_number"] = issues[0].number
        # 其他事件的初始化逻辑同理
        ...

    # 新commit事件装饰器
    def on_new_commit(self) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.callbacks[EventType.NEW_COMMIT].append(func)
            return func
        return decorator

    # 新issue评论事件装饰器
    def on_new_issue_comment(self) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.callbacks[EventType.NEW_ISSUE_COMMENT].append(func)
            return func
        return decorator

    # 全量更新事件装饰器
    def on_update(self) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.callbacks[EventType.ALL_UPDATES].append(func)
            return func
        return decorator

    # 轮询检测新事件
    def _poll_events(self):
        all_new_events = []
        # 检测新commit
        latest_commit = self.repo.get_commits()[0]
        if latest_commit.sha != self.latest_state["last_commit_sha"]:
            self.latest_state["last_commit_sha"] = latest_commit.sha
            for cb in self.callbacks[EventType.NEW_COMMIT]:
                cb(latest_commit)
            all_new_events.append({"type": EventType.NEW_COMMIT, "data": latest_commit})
        
        # 检测新issue评论，其他事件检测逻辑同理
        latest_comments = self.repo.get_issue_comments(sort="created", direction="desc", per_page=5)
        for comment in latest_comments:
            if comment.id > self.latest_state["last_issue_comment_id"]:
                for cb in self.callbacks[EventType.NEW_ISSUE_COMMENT]:
                    cb(comment)
                all_new_events.append({"type": EventType.NEW_ISSUE_COMMENT, "data": comment})
                if comment.id > self.latest_state["last_issue_comment_id"]:
                    self.latest_state["last_issue_comment_id"] = comment.id

        # 触发全局更新回调
        if all_new_events:
            for cb in self.callbacks[EventType.ALL_UPDATES]:
                cb(all_new_events)

    # 启动监听
    def run(self, poll_interval: int = 60):
        # poll_interval是轮询间隔，单位秒，不要设太短避免触发GitHub API限流
        while True:
            self._poll_events()
            time.sleep(poll_interval)

# 初始化Repo对象
repo = Repo(repo_name="用户名/仓库名", github_token="你的GitHub Personal Access Token")

# 注册新commit回调
@repo.on_new_commit()
def handle_new_commit(commit):
    print(f"收到新commit: {commit.sha}, 提交者: {commit.commit.author.name}")

# 注册全局更新回调
@repo.on_update()
def handle_all_updates(events):
    print(f"收到{len(events)}条新变动")
    for event in events:
        print(f"变动类型: {event['type']}")

# 启动监听
repo.run(poll_interval=30)

• 轮询模式下注意控制请求频率，GitHub公开API无token时限流是每小时60次，带token时是每小时5000次，合理设置轮询间隔避免被限流
• 如果需要更高的实时性，换成Webhook模式：在Repo类里加一个接口接收GitHub的Webhook推送，解析请求的事件类型直接触发对应回调即可，不需要轮询
• 可以加错误重试、断点续传逻辑，程序重启后不需要重新拉取所有历史事件，从缓存的最新状态开始监听即可
• 如果需要监听discussion事件，PyGitHub目前原生支持不好，可以直接调用GitHub GraphQL API获取discussion数据

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