decomposer: generate files for Build source connectors that retrieve content and source permissions from SCM and CMS systems and expose normalized content-change records to downstream ingestion workflows.

src/source_connectors/__init__.py

@@ -0,0 +1,11 @@
+from .auth import StaticTokenProvider, TokenProvider
+from .http import HttpClient, UrllibHttpClient
+from .models import ACLRecord, ChangeRecord, ContentRecord, Cursor, WebhookEvent
+from .github import GitHubConnector
+from .sharepoint import SharePointConnector
+
+__all__ = [
+    "ACLRecord", "ChangeRecord", "ContentRecord", "Cursor", "WebhookEvent",
+    "TokenProvider", "StaticTokenProvider", "HttpClient", "UrllibHttpClient",
+    "GitHubConnector", "SharePointConnector",
+]

src/source_connectors/auth.py

@@ -0,0 +1,13 @@
+from typing import Protocol
+
+class TokenProvider(Protocol):
+    """Returns a short-lived delegated OAuth access token for the configured source."""
+    def token(self) -> str: ...
+
+class StaticTokenProvider:
+    def __init__(self, value: str):
+        if not value:
+            raise ValueError("token must not be empty")
+        self.value = value
+    def token(self) -> str:
+        return self.value

src/source_connectors/base.py

@@ -0,0 +1,21 @@
+from abc import ABC, abstractmethod
+from collections.abc import Iterable
+from typing import Any
+from .auth import TokenProvider
+from .http import HttpClient
+from .models import ACLRecord, ChangeRecord, ContentRecord, Cursor, WebhookEvent
+
+class Connector(ABC):
+    source: str
+    def __init__(self, auth: TokenProvider, http: HttpClient):
+        self.auth, self.http = auth, http
+    def _get(self, url: str) -> dict[str, Any]:
+        return self.http.get(url, {"Authorization": f"Bearer {self.auth.token()}", "Accept": "application/json"})
+    @abstractmethod
+    def enumerate(self, scope: dict[str, str], cursor: Cursor | None = None) -> Iterable[ContentRecord]: ...
+    @abstractmethod
+    def changes(self, scope: dict[str, str], cursor: Cursor | None = None) -> Iterable[ChangeRecord]: ...
+    @abstractmethod
+    def normalize_webhook(self, headers: dict[str, str], payload: dict[str, Any]) -> WebhookEvent: ...
+    @abstractmethod
+    def acl(self, scope: dict[str, str]) -> Iterable[ACLRecord]: ...

src/source_connectors/github.py

@@ -0,0 +1,41 @@
+from collections.abc import Iterable
+from .base import Connector
+from .models import ACLRecord, ChangeRecord, ContentRecord, Cursor, WebhookEvent
+
+class GitHubConnector(Connector):
+    source = "github"
+    def _scope(self, scope):
+        owner, repo = scope.get("owner"), scope.get("repo")
+        if not owner or not repo or any(x in owner + repo for x in ("/", "..")):
+            raise ValueError("scope requires safe owner and repo")
+        return owner, repo
+    def enumerate(self, scope, cursor=None) -> Iterable[ContentRecord]:
+        owner, repo = self._scope(scope)
+        ref = scope.get("ref", "HEAD")
+        data = self._get(f"https://api.github.com/repos/{owner}/{repo}/git/trees/{ref}?recursive=1")
+        for item in data.get("tree", []):
+            if item.get("type") == "blob":
+                yield ContentRecord(item["sha"], item["path"], self.source, {"owner": owner, "repo": repo, "ref": ref}, item.get("url"), None, {"size": item.get("size")})
+    def changes(self, scope, cursor=None):
+        owner, repo = self._scope(scope)
+        params = f"?sha={scope.get('ref', 'HEAD')}"
+        if cursor and cursor.value: params += f"&since={cursor.value}"
+        data = self._get(f"https://api.github.com/repos/{owner}/{repo}/commits{params}")
+        latest = cursor.value if cursor else None
+        for commit in data:
+            latest = commit["sha"]
+            detail = self._get(commit["url"])
+            for file in detail.get("files", []):
+                operation = "delete" if file["status"] == "removed" else "upsert"
+                content = None if operation == "delete" else ContentRecord(file["sha"], file["filename"], self.source, {"owner": owner, "repo": repo}, file.get("raw_url"), commit.get("commit", {}).get("author", {}).get("date"), {"status": file["status"]})
+                yield ChangeRecord(f"{commit['sha']}:{file['filename']}", operation, content, self.source, Cursor(latest), commit.get("commit", {}).get("author", {}).get("date"))
+    def normalize_webhook(self, headers, payload):
+        action = payload.get("action", "modified")
+        op = "delete" if action in ("deleted", "removed") else "upsert"
+        repo = payload.get("repository", {})
+        return WebhookEvent(headers.get("X-GitHub-Delivery", ""), op, payload.get("after") or payload.get("ref", ""), self.source, {"owner": repo.get("owner", {}).get("login", ""), "repo": repo.get("name", "")}, Cursor(payload.get("after")), {"event": headers.get("X-GitHub-Event", "push")})
+    def acl(self, scope):
+        owner, repo = self._scope(scope)
+        data = self._get(f"https://api.github.com/repos/{owner}/{repo}/collaborators?affiliation=all&per_page=100")
+        for user in data:
+            yield ACLRecord(f"{owner}/{repo}", user["login"], "user", user.get("permissions", {}).get("push") and "write" or "read", False, {"id": user.get("id")})

src/source_connectors/http.py

@@ -0,0 +1,12 @@
+import json
+from typing import Any, Protocol
+from urllib.request import Request, urlopen
+
+class HttpClient(Protocol):
+    def get(self, url: str, headers: dict[str, str]) -> dict[str, Any]: ...
+
+class UrllibHttpClient:
+    def get(self, url: str, headers: dict[str, str]) -> dict[str, Any]:
+        request = Request(url, headers=headers, method="GET")
+        with urlopen(request, timeout=30) as response:
+            return json.loads(response.read())

src/source_connectors/models.py

@@ -0,0 +1,44 @@
+from dataclasses import dataclass, field
+from typing import Any
+
+@dataclass(frozen=True)
+class Cursor:
+    value: str | None = None
+
+@dataclass(frozen=True)
+class ContentRecord:
+    id: str
+    path: str
+    source: str
+    scope: dict[str, str]
+    content_url: str | None
+    modified_at: str | None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+@dataclass(frozen=True)
+class ChangeRecord:
+    id: str
+    operation: str  # upsert or delete
+    content: ContentRecord | None
+    source: str
+    cursor: Cursor
+    occurred_at: str | None = None
+
+@dataclass(frozen=True)
+class ACLRecord:
+    resource_id: str
+    subject_id: str
+    subject_type: str
+    role: str
+    inherited: bool = False
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+@dataclass(frozen=True)
+class WebhookEvent:
+    event_id: str
+    operation: str
+    resource_id: str
+    source: str
+    scope: dict[str, str]
+    cursor_hint: Cursor | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)

src/source_connectors/sharepoint.py

@@ -0,0 +1,42 @@
+from collections.abc import Iterable
+from .base import Connector
+from .models import ACLRecord, ChangeRecord, ContentRecord, Cursor, WebhookEvent
+
+class SharePointConnector(Connector):
+    source = "sharepoint"
+    base = "https://graph.microsoft.com/v1.0"
+    def _scope(self, scope):
+        site, drive = scope.get("site_id"), scope.get("drive_id")
+        if not site or not drive: raise ValueError("scope requires site_id and drive_id")
+        return site, drive
+    def _item(self, item, scope):
+        return ContentRecord(item["id"], item.get("name", ""), self.source, scope, item.get("webUrl"), item.get("lastModifiedDateTime"), {"folder": "folder" in item})
+    def enumerate(self, scope, cursor=None) -> Iterable[ContentRecord]:
+        site, drive = self._scope(scope)
+        url = cursor.value if cursor and cursor.value else f"{self.base}/sites/{site}/drives/{drive}/root/delta"
+        while url:
+            data = self._get(url)
+            for item in data.get("value", []):
+                if "file" in item: yield self._item(item, scope)
+            url = data.get("@odata.nextLink")
+    def changes(self, scope, cursor=None):
+        site, drive = self._scope(scope)
+        url = cursor.value if cursor and cursor.value else f"{self.base}/sites/{site}/drives/{drive}/root/delta"
+        while url:
+            data = self._get(url)
+            next_cursor = data.get("@odata.deltaLink") or data.get("@odata.nextLink") or url
+            for item in data.get("value", []):
+                deleted = "deleted" in item
+                yield ChangeRecord(item["id"], "delete" if deleted else "upsert", None if deleted else self._item(item, scope), self.source, Cursor(next_cursor), item.get("lastModifiedDateTime"))
+            url = data.get("@odata.nextLink")
+    def normalize_webhook(self, headers, payload):
+        resource = payload.get("resource", "")
+        return WebhookEvent(payload.get("subscriptionId", headers.get("X-Event-ID", "")), "upsert", resource, self.source, {"site_id": payload.get("siteId", ""), "drive_id": payload.get("driveId", "")}, None, {"changeType": payload.get("changeType", "updated")})
+    def acl(self, scope):
+        site, drive = self._scope(scope)
+        data = self._get(f"{self.base}/sites/{site}/drives/{drive}/root/permissions")
+        for permission in data.get("value", []):
+            for key in ("user", "group", "application"):
+                identity = permission.get("grantedToV2", {}).get(key, {})
+                if identity.get("id"):
+                    yield ACLRecord("root", identity["id"], key, ",".join(permission.get("roles", [])), False, {"permission_id": permission.get("id")})