feat: add CommitDiffsStream, PullRequestCommitDiffsStream, refactor diff logic into GitHubDiffStream (#372)

This PR adds child streams for fetching diffs of the respective parent
streams. Referenced [Parent-Child
Streams](https://gitlab.com/meltano/sdk/-/blob/main/docs/parent_streams.md)
and followed the same format as changes in #345.
- `CommitDiffsStream` -> `CommitsStream` as parent
- `PullRequestCommitDiffsStream` -> `PullRequestCommitsStream` as parent
- renamed `PullRequestCommits` to `PullRequestCommitsStream` to match
convention

Like #345, parent context is also added to relevant streams to give
users more flexibility in using or joining their tables.
I've tested changes on my end successfully and as always, please advise
if I missed anything. Thank you!

Latest commit refactors the repetition between the 3 diff streams into a
reusable block (`GithubDiffStream`)

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Edgar Ramírez Mondragón <[email protected]>

Author: 3 people

tap_github/client.py

@@ -306,6 +306,52 @@ def calculate_sync_cost(
         return {"rest": 1, "graphql": 0, "search": 0}
 
 
+class GitHubDiffStream(GitHubRestStream):
+    """Base class for GitHub diff streams."""
+
+    @property
+    def http_headers(self) -> dict:
+        """Return the http headers needed for diff requests."""
+        headers = super().http_headers
+        headers["Accept"] = "application/vnd.github.v3.diff"
+        return headers
+
+    def parse_response(self, response: requests.Response) -> Iterable[dict]:
+        """Parse the response to yield the diff text instead of an object
+        and prevent buffer overflow."""
+        if response.status_code != 200:
+            contents = response.json()
+            self.logger.info(
+                "Skipping %s due to %d error: %s",
+                self.name.replace("_", " "),
+                response.status_code,
+                contents["message"],
+            )
+            yield {
+                "success": False,
+                "error_message": contents["message"],
+            }
+            return
+
+        if content_length_str := response.headers.get("Content-Length"):
+            content_length = int(content_length_str)
+            max_size = 41_943_040  # 40 MiB
+            if content_length > max_size:
+                self.logger.info(
+                    "Skipping %s. The diff size (%.2f MiB) exceeded the maximum"
+                    " size limit of 40 MiB.",
+                    self.name.replace("_", " "),
+                    content_length / 1024 / 1024,
+                )
+                yield {
+                    "success": False,
+                    "error_message": "Diff exceeded the maximum size limit of 40 MiB.",
+                }
+                return
+
+        yield {"diff": response.text, "success": True}
+
+
 class GitHubGraphqlStream(GraphQLStream, GitHubRestStream):
     """GitHub Graphql stream class."""
 

tap_github/repository_streams.py

@@ -10,7 +10,7 @@
 from singer_sdk.exceptions import FatalAPIError
 from singer_sdk.helpers.jsonpath import extract_jsonpath
 
-from tap_github.client import GitHubGraphqlStream, GitHubRestStream
+from tap_github.client import GitHubDiffStream, GitHubGraphqlStream, GitHubRestStream
 from tap_github.schema_objects import (
     files_object,
     label_object,
@@ -1079,6 +1079,14 @@ def post_process(self, row: dict, context: dict | None = None) -> dict:
         row["commit_timestamp"] = row["commit"]["committer"]["date"]
         return row
 
+    def get_child_context(self, record: dict, context: dict | None) -> dict:
+        return {
+            "org": context["org"] if context else None,
+            "repo": context["repo"] if context else None,
+            "repo_id": context["repo_id"] if context else None,
+            "commit_id": record["sha"],
+        }
+
     schema = th.PropertiesList(
         th.Property("org", th.StringType),
         th.Property("repo", th.StringType),
@@ -1162,6 +1170,37 @@ class CommitCommentsStream(GitHubRestStream):
     ).to_dict()
 
 
+class CommitDiffsStream(GitHubDiffStream):
+    name = "commit_diffs"
+    path = "/repos/{org}/{repo}/commits/{commit_id}"
+    primary_keys: ClassVar[list[str]] = ["commit_id"]
+    parent_stream_type = CommitsStream
+    ignore_parent_replication_key = False
+    state_partitioning_keys: ClassVar[list[str]] = ["repo", "org"]
+
+    def post_process(self, row: dict, context: dict[str, str] | None = None) -> dict:
+        row = super().post_process(row, context)
+        if context is not None:
+            # Get commit ID (sha) from context
+            row["org"] = context["org"]
+            row["repo"] = context["repo"]
+            row["repo_id"] = context["repo_id"]
+            row["commit_id"] = context["commit_id"]
+        return row
+
+    schema = th.PropertiesList(
+        # Parent keys
+        th.Property("org", th.StringType),
+        th.Property("repo", th.StringType),
+        th.Property("repo_id", th.IntegerType),
+        th.Property("commit_id", th.StringType),
+        # Rest
+        th.Property("diff", th.StringType),
+        th.Property("success", th.BooleanType),
+        th.Property("error_message", th.StringType),
+    ).to_dict()
+
+
 class LabelsStream(GitHubRestStream):
     """Defines 'labels' stream."""
 
@@ -1355,14 +1394,23 @@ def get_child_context(self, record: dict, context: dict | None) -> dict:
     ).to_dict()
 
 
-class PullRequestCommits(GitHubRestStream):
+class PullRequestCommitsStream(GitHubRestStream):
     name = "pull_request_commits"
     path = "/repos/{org}/{repo}/pulls/{pull_number}/commits"
     ignore_parent_replication_key = False
     primary_keys: ClassVar[list[str]] = ["node_id"]
     parent_stream_type = PullRequestsStream
     state_partitioning_keys: ClassVar[list[str]] = ["repo", "org"]
 
+    def get_child_context(self, record: dict, context: dict | None) -> dict:
+        return {
+            "org": context["org"] if context else None,
+            "repo": context["repo"] if context else None,
+            "repo_id": context["repo_id"] if context else None,
+            "pull_number": context["pull_number"] if context else None,
+            "commit_id": record["sha"],
+        }
+
     schema = th.PropertiesList(
         # Parent keys
         th.Property("org", th.StringType),
@@ -1443,7 +1491,7 @@ def post_process(self, row: dict, context: dict[str, str] | None = None) -> dict
         return row
 
 
-class PullRequestDiffsStream(GitHubRestStream):
+class PullRequestDiffsStream(GitHubDiffStream):
     name = "pull_request_diffs"
     path = "/repos/{org}/{repo}/pulls/{pull_number}"
     primary_keys: ClassVar[list[str]] = ["pull_id"]
@@ -1453,53 +1501,48 @@ class PullRequestDiffsStream(GitHubRestStream):
     # Known Github API errors
     tolerated_http_errors: ClassVar[list[int]] = [404, 406, 422, 502]
 
-    @property
-    def http_headers(self) -> dict:
-        headers = super().http_headers
-        headers["Accept"] = "application/vnd.github.v3.diff"
-        return headers
+    def post_process(self, row: dict, context: dict[str, str] | None = None) -> dict:
+        row = super().post_process(row, context)
+        if context is not None:
+            # Get PR ID from context
+            row["org"] = context["org"]
+            row["repo"] = context["repo"]
+            row["repo_id"] = context["repo_id"]
+            row["pull_number"] = context["pull_number"]
+            row["pull_id"] = context["pull_id"]
+        return row
 
-    def parse_response(self, response: requests.Response) -> Iterable[dict]:
-        """Parse the response to yield the diff text instead of an object and prevent buffer overflow."""  # noqa: E501
-        if response.status_code != 200:
-            contents = response.json()
-            self.logger.info(
-                "Skipping PR due to %d error: %s",
-                response.status_code,
-                contents["message"],
-            )
-            yield {
-                "success": False,
-                "error_message": contents["message"],
-            }
-            return
+    schema = th.PropertiesList(
+        # Parent keys
+        th.Property("org", th.StringType),
+        th.Property("repo", th.StringType),
+        th.Property("repo_id", th.IntegerType),
+        th.Property("pull_number", th.IntegerType),
+        th.Property("pull_id", th.IntegerType),
+        # Rest
+        th.Property("diff", th.StringType),
+        th.Property("success", th.BooleanType),
+        th.Property("error_message", th.StringType),
+    ).to_dict()
 
-        if content_length_str := response.headers.get("Content-Length"):
-            content_length = int(content_length_str)
-            max_size = 41_943_040  # 40 MiB
-            if content_length > max_size:
-                self.logger.info(
-                    "Skipping PR. The diff size (%.2f MiB) exceeded the maximum size "
-                    "limit of 40 MiB.",
-                    content_length / 1024 / 1024,
-                )
-                yield {
-                    "success": False,
-                    "error_message": "Diff exceeded the maximum size limit of 40 MiB.",
-                }
-                return
 
-        yield {"diff": response.text, "success": True}
+class PullRequestCommitDiffsStream(GitHubDiffStream):
+    name = "pull_request_commit_diffs"
+    path = "/repos/{org}/{repo}/commits/{commit_id}"
+    primary_keys: ClassVar[list[str]] = ["commit_id"]
+    parent_stream_type = PullRequestCommitsStream
+    ignore_parent_replication_key = False
+    state_partitioning_keys: ClassVar[list[str]] = ["repo", "org"]
 
     def post_process(self, row: dict, context: dict[str, str] | None = None) -> dict:
         row = super().post_process(row, context)
         if context is not None:
-            # Get PR ID from context
+            # Get commit ID (sha) from context
             row["org"] = context["org"]
             row["repo"] = context["repo"]
             row["repo_id"] = context["repo_id"]
             row["pull_number"] = context["pull_number"]
-            row["pull_id"] = context["pull_id"]
+            row["commit_id"] = context["commit_id"]
         return row
 
     schema = th.PropertiesList(
@@ -1508,7 +1551,7 @@ def post_process(self, row: dict, context: dict[str, str] | None = None) -> dict
         th.Property("repo", th.StringType),
         th.Property("repo_id", th.IntegerType),
         th.Property("pull_number", th.IntegerType),
-        th.Property("pull_id", th.IntegerType),
+        th.Property("commit_id", th.StringType),
         # Rest
         th.Property("diff", th.StringType),
         th.Property("success", th.BooleanType),

tap_github/streams.py

@@ -16,6 +16,7 @@
     BranchesStream,
     CollaboratorsStream,
     CommitCommentsStream,
+    CommitDiffsStream,
     CommitsStream,
     CommunityProfileStream,
     ContributorsStream,
@@ -34,7 +35,8 @@
     ProjectCardsStream,
     ProjectColumnsStream,
     ProjectsStream,
-    PullRequestCommits,
+    PullRequestCommitDiffsStream,
+    PullRequestCommitsStream,
     PullRequestDiffsStream,
     PullRequestsStream,
     ReadmeHtmlStream,
@@ -82,6 +84,7 @@ def __init__(self, valid_queries: set[str], streams: list[type[Stream]]) -> None
             CollaboratorsStream,
             CommitCommentsStream,
             CommitsStream,
+            CommitDiffsStream,
             CommunityProfileStream,
             ContributorsStream,
             DependenciesStream,
@@ -98,7 +101,8 @@ def __init__(self, valid_queries: set[str], streams: list[type[Stream]]) -> None
             ProjectCardsStream,
             ProjectColumnsStream,
             ProjectsStream,
-            PullRequestCommits,
+            PullRequestCommitsStream,
+            PullRequestCommitDiffsStream,
             PullRequestDiffsStream,
             PullRequestsStream,
             ReadmeHtmlStream,