Source Github: add custom pagination size for large streams

airbyte-integrations/connectors/source-github/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.2.13
+LABEL io.airbyte.version=0.2.14
 LABEL io.airbyte.name=airbyte/source-github

airbyte-integrations/connectors/source-github/source_github/source.py

@@ -44,6 +44,7 @@
 )
 
 TOKEN_SEPARATOR = ","
+DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM = 10
 # To scan all the repos within orgnaization, organization name could be
 # specified by using asteriks i.e. "airbytehq/*"
 ORGANIZATION_PATTERN = re.compile("^.*/\\*$")
@@ -140,6 +141,7 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
             repository_stats_stream = RepositoryStats(
                 authenticator=authenticator,
                 repositories=repositories,
+                page_size_for_large_streams=config.get("page_size_for_large_streams", DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM),
             )
             for stream_slice in repository_stats_stream.stream_slices(sync_mode=SyncMode.full_refresh):
                 next(repository_stats_stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream_slice), None)
@@ -153,9 +155,10 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         repositories = repos + organization_repos
 
         organizations = list({org.split("/")[0] for org in repositories})
+        page_size = config.get("page_size_for_large_streams", DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM)
 
         organization_args = {"authenticator": authenticator, "organizations": organizations}
-        repository_args = {"authenticator": authenticator, "repositories": repositories}
+        repository_args = {"authenticator": authenticator, "repositories": repositories, "page_size_for_large_streams": page_size}
         repository_args_with_start_date = {**repository_args, "start_date": config["start_date"]}
 
         default_branches, branches_to_pull = self._get_branches_data(config.get("branch", ""), repository_args)

airbyte-integrations/connectors/source-github/source_github/spec.json

@@ -52,7 +52,6 @@
           }
         ]
       },
-
       "repository": {
         "type": "string",
         "examples": ["airbytehq/airbyte", "airbytehq/*"],
@@ -71,6 +70,14 @@
         "title": "Branch",
         "examples": ["airbytehq/airbyte/master"],
         "description": "Space-delimited list of GitHub repository branches to pull commits for, e.g. `airbytehq/airbyte/master`. If no branches are specified for a repository, the default branch will be pulled."
+      },
+      "page_size_for_large_streams": {
+        "type": "integer",
+        "title": "Page size for large streams",
+        "minimum": 1,
+        "maximum": 100,
+        "default": 10,
+        "description": "The Github connector contains several streams with a large load. The page size of such streams depends on the size of your repository. Recommended to specify values between 10 and 30."
       }
     }
   },

airbyte-integrations/connectors/source-github/source_github/streams.py

@@ -13,22 +13,27 @@
 from airbyte_cdk.sources.streams.http import HttpStream, HttpSubStream
 from requests.exceptions import HTTPError
 
+DEFAULT_PAGE_SIZE = 100
+
 
 class GithubStream(HttpStream, ABC):
     url_base = "https://api.github.com/"
 
     primary_key = "id"
     use_cache = True
 
-    # GitHub pagination could be from 1 to 100.
-    page_size = 100
+    # Detect streams with high API load
+    large_stream = False
 
     stream_base_params = {}
 
-    def __init__(self, repositories: List[str], **kwargs):
+    def __init__(self, repositories: List[str], page_size_for_large_streams: int, **kwargs):
         super().__init__(**kwargs)
         self.repositories = repositories
 
+        # GitHub pagination could be from 1 to 100.
+        self.page_size = page_size_for_large_streams if self.large_stream else DEFAULT_PAGE_SIZE
+
         MAX_RETRIES = 3
         adapter = requests.adapters.HTTPAdapter(max_retries=MAX_RETRIES)
         self._session.mount("https://", adapter)
@@ -295,6 +300,9 @@ class Organizations(GithubStream):
     API docs: https://docs.github.com/en/rest/reference/orgs#get-an-organization
     """
 
+    # GitHub pagination could be from 1 to 100.
+    page_size = 100
+
     def __init__(self, organizations: List[str], **kwargs):
         super(GithubStream, self).__init__(**kwargs)
         self.organizations = organizations
@@ -394,7 +402,7 @@ class PullRequests(SemiIncrementalGithubStream):
     API docs: https://docs.github.com/en/rest/reference/pulls#list-pull-requests
     """
 
-    page_size = 50
+    large_stream = True
     first_read_override_key = "first_read_override"
 
     def __init__(self, **kwargs):
@@ -524,7 +532,7 @@ class Comments(IncrementalGithubStream):
     API docs: https://docs.github.com/en/rest/reference/issues#list-issue-comments-for-a-repository
     """
 
-    page_size = 30  # `comments` is a large stream so it's better to set smaller page size.
+    large_stream = True
 
     def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
         return f"repos/{stream_slice['repository']}/issues/comments"
@@ -637,7 +645,7 @@ class Issues(IncrementalGithubStream):
     API docs: https://docs.github.com/en/rest/reference/issues#list-repository-issues
     """
 
-    page_size = 50  # `issues` is a large stream so it's better to set smaller page size.
+    large_stream = True
 
     stream_base_params = {
         "state": "all",
@@ -651,7 +659,7 @@ class ReviewComments(IncrementalGithubStream):
     API docs: https://docs.github.com/en/rest/reference/pulls#list-review-comments-in-a-repository
     """
 
-    page_size = 30  # `review-comments` is a large stream so it's better to set smaller page size.
+    large_stream = True
 
     def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
         return f"repos/{stream_slice['repository']}/pulls/comments"

airbyte-integrations/connectors/source-github/unit_tests/test_stream.py

@@ -16,7 +16,7 @@
 @responses.activate
 @patch("time.sleep")
 def test_bad_gateway_retry(time_mock):
-    args = {"authenticator": None, "repositories": ["test_repo"], "start_date": "start_date"}
+    args = {"authenticator": None, "repositories": ["test_repo"], "start_date": "start_date", "page_size_for_large_streams": 30}
     stream = PullRequestCommentReactions(**args)
     stream_slice = {"repository": "test_repo", "id": "id"}
 

docs/integrations/sources/github.md

@@ -92,6 +92,7 @@ Your token should have at least the `repo` scope. Depending on which streams you
 
 | Version | Date | Pull Request | Subject |
 | :--- | :--- | :--- | :--- |
+| 0.2.14 | 2021-01-?? | [????](https://github.com/airbytehq/airbyte/pull/????) | Add custom pagination size for large streams |
 | 0.2.13 | 2021-01-20 | [9619](https://github.com/airbytehq/airbyte/pull/9619) | Fix logging for function `should_retry` |
 | 0.2.11 | 2021-01-17 | [9492](https://github.com/airbytehq/airbyte/pull/9492) | Remove optional parameter `Accept` for reaction`s streams to fix error with 502 HTTP status code in response |
 | 0.2.10 | 2021-01-03 | [7250](https://github.com/airbytehq/airbyte/pull/7250) | Use CDK caching and convert PR-related streams to incremental |