🐛 Source Github: validate input organizations and repositories (#15730)

Signed-off-by: Sergey Chvalyuk <[email protected]>

Author: grubberr

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -323,7 +323,7 @@
 - name: GitHub
   sourceDefinitionId: ef69ef6e-aa7f-4af1-a01d-ef775033524e
   dockerRepository: airbyte/source-github
-  dockerImageTag: 0.2.45
+  dockerImageTag: 0.2.46
   documentationUrl: https://docs.airbyte.io/integrations/sources/github
   icon: github.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -2701,7 +2701,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-github:0.2.45"
+- dockerImage: "airbyte/source-github:0.2.46"
   spec:
     documentationUrl: "https://docs.airbyte.com/integrations/sources/github"
     connectionSpecification:

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.45
+LABEL io.airbyte.version=0.2.46
 LABEL io.airbyte.name=airbyte/source-github

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

@@ -3,7 +3,6 @@
 #
 
 
-import re
 from typing import Any, Dict, List, Mapping, Tuple
 
 from airbyte_cdk import AirbyteLogger
@@ -50,43 +49,57 @@
     WorkflowRuns,
     Workflows,
 )
+from .utils import read_full_refresh
 
 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("^.*/\\*$")
 
 
 class SourceGithub(AbstractSource):
     @staticmethod
-    def _generate_repositories(config: Mapping[str, Any], authenticator: MultipleTokenAuthenticator) -> Tuple[List[str], List[str]]:
+    def _get_org_repositories(config: Mapping[str, Any], authenticator: MultipleTokenAuthenticator) -> Tuple[List[str], List[str]]:
         """
-        Parse repositories config line and produce two lists of repositories.
+        Parse config.repository and produce two lists: organizations, repositories.
         Args:
             config (dict): Dict representing connector's config
             authenticator(MultipleTokenAuthenticator): authenticator object
-        Returns:
-            Tuple[List[str], List[str]]: Tuple of two lists: first representing
-            repositories directly mentioned in config and second is
-            organization repositories from orgs/{org}/repos request.
         """
-        repositories = list(filter(None, config["repository"].split(" ")))
-
-        if not repositories:
+        config_repositories = set(filter(None, config["repository"].split(" ")))
+        if not config_repositories:
             raise Exception("Field `repository` required to be provided for connect to Github API")
 
-        repositories_list: set = {repo for repo in repositories if not ORGANIZATION_PATTERN.match(repo)}
-        organizations = [org.split("/")[0] for org in repositories if org not in repositories_list]
-        organisation_repos = set()
-        if organizations:
-            repos = Repositories(authenticator=authenticator, organizations=organizations)
-            for stream in repos.stream_slices(sync_mode=SyncMode.full_refresh):
-                organisation_repos = organisation_repos.union(
-                    {r["full_name"] for r in repos.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream)}
-                )
+        repositories = set()
+        organizations = set()
+        unchecked_repos = set()
+        unchecked_orgs = set()
+
+        for org_repos in config_repositories:
+            org, _, repos = org_repos.partition("/")
+            if repos == "*":
+                unchecked_orgs.add(org)
+            else:
+                unchecked_repos.add(org_repos)
+
+        if unchecked_orgs:
+            stream = Repositories(authenticator=authenticator, organizations=unchecked_orgs)
+            for record in read_full_refresh(stream):
+                repositories.add(record["full_name"])
+                organizations.add(record["organization"])
+
+        unchecked_repos = unchecked_repos - repositories
+        if unchecked_repos:
+            stream = RepositoryStats(
+                authenticator=authenticator,
+                repositories=unchecked_repos,
+                page_size_for_large_streams=config.get("page_size_for_large_streams", DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM),
+            )
+            for record in read_full_refresh(stream):
+                repositories.add(record["full_name"])
+                organization = record.get("organization", {}).get("login")
+                if organization:
+                    organizations.add(organization)
 
-        return list(repositories_list), list(organisation_repos)
+        return list(organizations), list(repositories)
 
     @staticmethod
     def _get_authenticator(config: Dict[str, Any]):
@@ -139,20 +152,9 @@ def _get_branches_data(selected_branches: str, full_refresh_args: Dict[str, Any]
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
         try:
             authenticator = self._get_authenticator(config)
-            # In case of getting repository list for given organization was
-            # successfull no need of checking stats for every repository within
-            # that organization.
-            # Since we have "repo" scope requested it should grant access to private repos as well:
-            # https://docs.github.com/en/developers/apps/building-oauth-apps/scopes-for-oauth-apps#available-scopes
-            repositories, _ = self._generate_repositories(config=config, authenticator=authenticator)
-
-            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)
+            _, repositories = self._get_org_repositories(config=config, authenticator=authenticator)
+            if not repositories:
+                return False, "no valid repositories found"
             return True, None
 
         except Exception as e:
@@ -172,10 +174,7 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
 
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         authenticator = self._get_authenticator(config)
-        repos, organization_repos = self._generate_repositories(config=config, authenticator=authenticator)
-        repositories = repos + organization_repos
-
-        organizations = list({org.split("/")[0] for org in repositories})
+        organizations, repositories = self._get_org_repositories(config=config, authenticator=authenticator)
         page_size = config.get("page_size_for_large_streams", DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM)
 
         organization_args = {"authenticator": authenticator, "organizations": organizations}

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

@@ -3,9 +3,21 @@
 #
 
 
+from airbyte_cdk.models import SyncMode
+from airbyte_cdk.sources.streams import Stream
+
+
 def getter(D: dict, key_or_keys):
     if not isinstance(key_or_keys, list):
         key_or_keys = [key_or_keys]
     for k in key_or_keys:
         D = D[k]
     return D
+
+
+def read_full_refresh(stream_instance: Stream):
+    slices = stream_instance.stream_slices(sync_mode=SyncMode.full_refresh)
+    for _slice in slices:
+        records = stream_instance.read_records(stream_slice=_slice, sync_mode=SyncMode.full_refresh)
+        for record in records:
+            yield record

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

@@ -19,7 +19,7 @@ def check_source(repo_line: str) -> AirbyteConnectionStatus:
 
 @responses.activate
 def test_check_connection_repos_only():
-    responses.add("GET", "https://api.github.com/repos/airbyte", json={})
+    responses.add("GET", "https://api.github.com/repos/airbyte", json={"full_name": "airbyte"})
 
     status = check_source("airbyte airbyte airbyte")
     assert not status.message
@@ -31,8 +31,12 @@ def test_check_connection_repos_only():
 @responses.activate
 def test_check_connection_repos_and_org_repos():
     repos = [{"name": f"name {i}", "full_name": f"full name {i}", "updated_at": "2020-01-01T00:00:00Z"} for i in range(1000)]
-    responses.add("GET", "https://api.github.com/repos/airbyte/test", json={})
-    responses.add("GET", "https://api.github.com/repos/airbyte/test2", json={})
+    responses.add(
+        "GET", "https://api.github.com/repos/airbyte/test", json={"full_name": "airbyte/test", "organization": {"login": "airbyte"}}
+    )
+    responses.add(
+        "GET", "https://api.github.com/repos/airbyte/test2", json={"full_name": "airbyte/test2", "organization": {"login": "airbyte"}}
+    )
     responses.add("GET", "https://api.github.com/orgs/airbytehq/repos", json=repos)
     responses.add("GET", "https://api.github.com/orgs/org/repos", json=repos)
 
@@ -95,13 +99,19 @@ def test_get_branches_data():
 
 
 @responses.activate
-def test_generate_repositories():
+def test_get_org_repositories():
 
     source = SourceGithub()
 
     with pytest.raises(Exception):
         config = {"repository": ""}
-        source._generate_repositories(config, authenticator=None)
+        source._get_org_repositories(config, authenticator=None)
+
+    responses.add(
+        "GET",
+        "https://api.github.com/repos/airbytehq/integration-test",
+        json={"full_name": "airbytehq/integration-test", "organization": {"login": "airbytehq"}},
+    )
 
     responses.add(
         "GET",
@@ -113,9 +123,7 @@ def test_generate_repositories():
     )
 
     config = {"repository": "airbytehq/integration-test docker/*"}
-    repositories_list, organisation_repos = source._generate_repositories(config, authenticator=None)
+    organisations, repositories = source._get_org_repositories(config, authenticator=None)
 
-    assert repositories_list == ["airbytehq/integration-test"]
-    assert len(organisation_repos) == 2
-    assert "docker/compose" in organisation_repos
-    assert "docker/docker-py" in organisation_repos
+    assert set(repositories) == {"airbytehq/integration-test", "docker/docker-py", "docker/compose"}
+    assert set(organisations) == {"airbytehq", "docker"}

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

@@ -43,8 +43,9 @@
     Users,
     WorkflowRuns,
 )
+from source_github.utils import read_full_refresh
 
-from .utils import ProjectsResponsesAPI, read_full_refresh, read_incremental
+from .utils import ProjectsResponsesAPI, read_incremental
 
 DEFAULT_BACKOFF_DELAYS = [5, 10, 20, 40, 80]
 
@@ -108,7 +109,7 @@ def request_callback(request):
     )
 
     stream = Organizations(organizations=["airbytehq"])
-    read_full_refresh(stream)
+    list(read_full_refresh(stream))
     assert len(responses.calls) == 2
     assert responses.calls[0].request.url == "https://api.github.com/orgs/airbytehq?per_page=100"
     assert responses.calls[1].request.url == "https://api.github.com/orgs/airbytehq?per_page=100"
@@ -139,7 +140,7 @@ def test_graphql_rate_limited(time_mock, sleep_mock):
     )
 
     stream = PullRequestStats(repositories=["airbytehq/airbyte"], page_size_for_large_streams=30)
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == []
     assert len(responses.calls) == 2
     assert responses.calls[0].request.url == "https://api.github.com/graphql"
@@ -159,7 +160,7 @@ def test_stream_teams_404():
         json={"message": "Not Found", "documentation_url": "https://docs.github.com/rest/reference/teams#list-teams"},
     )
 
-    assert read_full_refresh(stream) == []
+    assert list(read_full_refresh(stream)) == []
     assert len(responses.calls) == 1
     assert responses.calls[0].request.url == "https://api.github.com/orgs/org_name/teams?per_page=100"
 
@@ -170,7 +171,7 @@ def test_stream_organizations_read():
     stream = Organizations(**organization_args)
     responses.add("GET", "https://api.github.com/orgs/org1", json={"id": 1})
     responses.add("GET", "https://api.github.com/orgs/org2", json={"id": 2})
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [{"id": 1}, {"id": 2}]
 
 
@@ -180,7 +181,7 @@ def test_stream_teams_read():
     stream = Teams(**organization_args)
     responses.add("GET", "https://api.github.com/orgs/org1/teams", json=[{"id": 1}, {"id": 2}])
     responses.add("GET", "https://api.github.com/orgs/org2/teams", json=[{"id": 3}])
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [{"id": 1, "organization": "org1"}, {"id": 2, "organization": "org1"}, {"id": 3, "organization": "org2"}]
     assert len(responses.calls) == 2
     assert responses.calls[0].request.url == "https://api.github.com/orgs/org1/teams?per_page=100"
@@ -193,7 +194,7 @@ def test_stream_users_read():
     stream = Users(**organization_args)
     responses.add("GET", "https://api.github.com/orgs/org1/members", json=[{"id": 1}, {"id": 2}])
     responses.add("GET", "https://api.github.com/orgs/org2/members", json=[{"id": 3}])
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [{"id": 1, "organization": "org1"}, {"id": 2, "organization": "org1"}, {"id": 3, "organization": "org2"}]
     assert len(responses.calls) == 2
     assert responses.calls[0].request.url == "https://api.github.com/orgs/org1/members?per_page=100"
@@ -212,7 +213,7 @@ def test_stream_repositories_404():
         json={"message": "Not Found", "documentation_url": "https://docs.github.com/rest/reference/repos#list-organization-repositories"},
     )
 
-    assert read_full_refresh(stream) == []
+    assert list(read_full_refresh(stream)) == []
     assert len(responses.calls) == 1
     assert responses.calls[0].request.url == "https://api.github.com/orgs/org_name/repos?per_page=100&sort=updated&direction=desc"
 
@@ -226,7 +227,7 @@ def test_stream_repositories_read():
         "GET", "https://api.github.com/orgs/org1/repos", json=[{"id": 1, "updated_at": updated_at}, {"id": 2, "updated_at": updated_at}]
     )
     responses.add("GET", "https://api.github.com/orgs/org2/repos", json=[{"id": 3, "updated_at": updated_at}])
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [
         {"id": 1, "organization": "org1", "updated_at": updated_at},
         {"id": 2, "organization": "org1", "updated_at": updated_at},
@@ -250,7 +251,7 @@ def test_stream_projects_disabled():
         json={"message": "Projects are disabled for this repository", "documentation_url": "https://docs.github.com/v3/projects"},
     )
 
-    assert read_full_refresh(stream) == []
+    assert list(read_full_refresh(stream)) == []
     assert len(responses.calls) == 1
     assert responses.calls[0].request.url == "https://api.github.com/repos/test_repo/projects?per_page=100&state=all"
 
@@ -429,7 +430,7 @@ def test_stream_pull_request_commits():
         json=[{"sha": 3}, {"sha": 4}],
     )
 
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [
         {"sha": 1, "repository": "organization/repository", "pull_number": 2},
         {"sha": 2, "repository": "organization/repository", "pull_number": 2},
@@ -785,7 +786,7 @@ def get_records(cursor_field):
     ]:
         stream = cls(**repository_args_with_start_date)
         responses.add("GET", url, json=get_json_response(stream.cursor_field))
-        records = read_full_refresh(stream)
+        records = list(read_full_refresh(stream))
         assert records == get_records(stream.cursor_field)[1:2]
 
     for cls, url in [
@@ -796,7 +797,7 @@ def get_records(cursor_field):
     ]:
         stream = cls(**repository_args)
         responses.add("GET", url, json=get_json_response(stream.cursor_field))
-        records = read_full_refresh(stream)
+        records = list(read_full_refresh(stream))
         assert records == get_records(stream.cursor_field)
 
     responses.add(
@@ -809,7 +810,7 @@ def get_records(cursor_field):
     )
 
     stream = Stargazers(**repository_args_with_start_date)
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
     assert records == [{"repository": "organization/repository", "starred_at": "2022-02-02T00:00:00Z", "user": {"id": 2}, "user_id": 2}]
 
 
@@ -863,7 +864,7 @@ def test_stream_team_members_full_refresh():
     responses.add("GET", "https://api.github.com/orgs/org1/teams/team2/memberships/login2", json={"username": "login2"})
 
     stream = TeamMembers(parent=Teams(**organization_args), **repository_args)
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
 
     assert records == [
         {"login": "login1", "organization": "org1", "team_slug": "team1"},
@@ -872,7 +873,7 @@ def test_stream_team_members_full_refresh():
     ]
 
     stream = TeamMemberships(parent=stream, **repository_args)
-    records = read_full_refresh(stream)
+    records = list(read_full_refresh(stream))
 
     assert records == [
         {"username": "login1", "organization": "org1", "team_slug": "team1"},

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

@@ -9,14 +9,6 @@
 from airbyte_cdk.sources.streams import Stream
 
 
-def read_full_refresh(stream_instance: Stream):
-    records = []
-    slices = stream_instance.stream_slices(sync_mode=SyncMode.full_refresh)
-    for slice in slices:
-        records.extend(list(stream_instance.read_records(stream_slice=slice, sync_mode=SyncMode.full_refresh)))
-    return records
-
-
 def read_incremental(stream_instance: Stream, stream_state: MutableMapping[str, Any]):
     res = []
     slices = stream_instance.stream_slices(sync_mode=SyncMode.incremental, stream_state=stream_state)

docs/integrations/sources/github.md

@@ -141,6 +141,7 @@ The GitHub connector should not run into GitHub API limitations under normal usa
 
 | Version | Date       | Pull Request | Subject                                                                                                      |
 |:--------|:-----------| :--- |:-------------------------------------------------------------------------------------------------------------|
+| 0.2.46  | 2022-08-17 | [15730](https://github.com/airbytehq/airbyte/pull/15730) | Validate input organizations and repositories                                                              |
 | 0.2.45  | 2022-08-11 | [15420](https://github.com/airbytehq/airbyte/pull/15420) | "User" object can be "null"                                                                                |
 | 0.2.44  | 2022-08-01 | [14795](https://github.com/airbytehq/airbyte/pull/14795) | Use GraphQL for `pull_request_comment_reactions` stream                                                    |
 | 0.2.43  | 2022-07-26 | [15049](https://github.com/airbytehq/airbyte/pull/15049) | Bugfix schemas for streams `deployments`, `workflow_runs`, `teams`                                         |