Skip to content

github_notifications

github_notifications

GitHub Notifications connector — unread notifications via GitHub REST API.

Uses a Personal Access Token stored in the connector config dir. All API calls are in module-level functions for easy mocking in tests.

Classes

GitHubNotificationsConnector 

GitHubNotificationsConnector(
    *, token_path: str = _DEFAULT_TOKEN_PATH
)

Bases: BaseConnector

Sync unread notifications from GitHub.

Source code in src/diapason/connectors/github_notifications.py 
48
49
50
def __init__(self, *, token_path: str = _DEFAULT_TOKEN_PATH) -> None:
    self._token_path = Path(token_path)
    self._status = SyncStatus()
Methods:
sync 
sync(
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,
) -> Iterator[Document]

Yield Documents for each GitHub notification.

Source code in src/diapason/connectors/github_notifications.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def sync(
    self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
    """Yield Documents for each GitHub notification."""
    token = self._load_token()
    params: Dict[str, str] = {}
    if since is not None:
        params["since"] = f"{since.isoformat()}Z"

    notifications = _github_api_get(token, params=params)

    for notif in notifications:
        subject = notif.get("subject", {})
        repo = notif.get("repository", {}).get("full_name", "")
        reason = notif.get("reason", "")
        notif_type = subject.get("type", "")
        title = subject.get("title", "")
        notif_id = notif.get("id", "")
        updated_at = notif.get("updated_at", "")

        content = f"Reason: {reason}, Repository: {repo}"
        ts = datetime.now()
        if updated_at:
            try:
                ts = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
            except ValueError:
                pass

        yield Document(
            doc_id=f"github-notification-{notif_id}",
            source="github_notifications",
            doc_type="notification",
            content=content,
            title=title,
            timestamp=ts,
            url=subject.get("url"),
            metadata={
                "reason": reason,
                "repo": repo,
                "type": notif_type,
            },
        )

    self._status.state = "idle"
    self._status.last_sync = datetime.now()