Merge branch 'master' into ddavydov/#744-oncall-fb-marketing-modify-s…

…tream-schema

.github/workflows/publish-helm-charts.yml

@@ -8,14 +8,17 @@ on:
       - 'master'
     paths:
       - 'charts/**'
+      - '!charts/**/Chart.yaml'
+      - '!charts/**/Chart.lock'
   workflow_dispatch:
 
 jobs:
   generate-semantic-version:
     name: Generate semantic version
     runs-on: ubuntu-22.04
     outputs:
-      next-version: ${{ steps.sem-ver.outputs.version }}
+      next-version: ${{ steps.ver.outputs.fragment }}
+      tag: ${{ steps.sem-ver.outputs.version_tag }}
     steps:
       - uses: actions/checkout@v2
         with:
@@ -30,11 +33,21 @@ jobs:
           format: "${major}.${minor}.${patch}"
           change_path: "./charts"
           bump_each_commit: true
+          namespace: helm
+
+      - name: "Remove -helm from ver number"
+        shell: bash
+        env:
+          VERSION: ${{ steps.sem-ver.outputs.version }}
+        id: ver
+        run: echo "::set-output name=fragment::${VERSION%%-*}"
+
 
   release-chart:
     name: Chart release
     runs-on: ubuntu-22.04
     needs: ["generate-semantic-version"]
+    permissions: write-all
     steps:
       - uses: actions/checkout@v3
         with:
@@ -51,14 +64,17 @@ jobs:
         shell: bash
         working-directory: ./airbyte/charts
         run: |
-          sed -i "s/    version: placeholder/    version: ${{ needs.generate-semantic-version.outputs.next-version }}/g" airbyte/Chart.yaml
+          sed -i -E "s/    version: [[:digit:]].[[:digit:]].[[:digit:]]/    version: ${{ needs.generate-semantic-version.outputs.next-version }}/g" airbyte/Chart.yaml
+          sed -i -E 's/version: [0-9]+\.[0-9]+\.[0-9]+/version: ${{ needs.generate-semantic-version.outputs.next-version }}/' airbyte/Chart.yaml
+
 
       - name: "Helm package"
         shell: bash
         run: |
           declare -a StringArray=("airbyte-bootloader" "airbyte-server" "airbyte-temporal" "airbyte-webapp" "airbyte-pod-sweeper" "airbyte-worker" "airbyte-metrics")
           for val in ${StringArray[@]}; do
             cd ./airbyte/charts/${val} && helm dep update && cd $GITHUB_WORKSPACE
+            sed -i -E  's/version: \"[0-9]+\.[0-9]+\.[0-9]+\"/version: \"${{ needs.generate-semantic-version.outputs.next-version }}\"/' ./airbyte/charts/${val}/Chart.yaml
             helm package ./airbyte/charts/${val} -d airbyte-oss  --version ${{ needs.generate-semantic-version.outputs.next-version }}
           done
           helm repo index airbyte-oss/
@@ -89,3 +105,33 @@ jobs:
           add: '.'
           cwd: './airbyte-oss/'
 
+
+      - name: "Generate changelog"
+        shell: bash
+        id: changelog
+        run: |
+          cd ./airbyte/
+          echo "::set-output name=changelog::$(PAGER=cat git log $(git describe --tags --match "*-helm" $(git rev-list --tags --max-count=1))..HEAD --oneline --decorate=no)"
+
+
+      - name: Create Pull Request
+        uses: peter-evans/create-pull-request@v4
+        with:
+          path: ./airbyte/
+          branch: update-helm-chart-version-ref
+          branch-suffix: random
+          title: Bump helm chart version reference to ${{ needs.generate-semantic-version.outputs.next-version }}
+          body: |
+            ## What
+            Bump version reference in all Chart.yaml files to ${{ needs.generate-semantic-version.outputs.next-version }}
+            CHANGELOG:
+            ${{ steps.changelog.outputs.changelog }}
+          commit-message: Bump helm chart version reference to ${{ needs.generate-semantic-version.outputs.next-version }}
+          delete-branch: true
+
+      - name: Create tag
+        shell: bash
+        run: |
+          cd ./airbyte/
+          git tag ${{ needs.generate-semantic-version.outputs.tag }}
+          git push origin ${{ needs.generate-semantic-version.outputs.tag }}

airbyte-cdk/python/CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.1.100
+
+- Low-code: Pass stream_slice to read_records when reading from CheckStream
+
 ## 0.1.99
 
 - Low-code: Fix default stream schema loader

airbyte-cdk/python/airbyte_cdk/sources/declarative/checks/check_stream.py

@@ -36,10 +36,22 @@ def check_connection(self, source: Source, logger: logging.Logger, config: Mappi
             if stream_name in stream_name_to_stream.keys():
                 stream = stream_name_to_stream[stream_name]
                 try:
-                    records = stream.read_records(sync_mode=SyncMode.full_refresh)
+                    # Some streams need a stream slice to read records (eg if they have a SubstreamSlicer)
+                    stream_slice = self._get_stream_slice(stream)
+                    records = stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream_slice)
                     next(records)
                 except Exception as error:
                     return False, f"Unable to connect to stream {stream_name} - {error}"
             else:
                 raise ValueError(f"{stream_name} is not part of the catalog. Expected one of {stream_name_to_stream.keys()}")
         return True, None
+
+    def _get_stream_slice(self, stream):
+        slices = stream.stream_slices(
+            cursor_field=stream.cursor_field,
+            sync_mode=SyncMode.full_refresh,
+        )
+        try:
+            return next(slices)
+        except StopIteration:
+            return {}

airbyte-cdk/python/setup.py

@@ -15,7 +15,7 @@
 
 setup(
     name="airbyte-cdk",
-    version="0.1.99",
+    version="0.1.100",
     description="A framework for writing Airbyte Connectors.",
     long_description=README,
     long_description_content_type="text/markdown",

airbyte-cdk/python/unit_tests/sources/declarative/checks/test_check_stream.py

@@ -15,17 +15,19 @@
 
 
 @pytest.mark.parametrize(
-    "test_name, record, streams_to_check, expectation",
+    "test_name, record, streams_to_check, stream_slice, expectation",
     [
-        ("test success check", record, stream_names, (True, None)),
-        ("test fail check", None, stream_names, (True, None)),
-        ("test try to check invalid stream", record, ["invalid_stream_name"], None),
+        ("test_success_check", record, stream_names, {}, (True, None)),
+        ("test_success_check_stream_slice", record, stream_names, {"slice": "slice_value"}, (True, None)),
+        ("test_fail_check", None, stream_names, {}, (True, None)),
+        ("test_try_to_check_invalid stream", record, ["invalid_stream_name"], {}, None),
     ],
 )
-def test_check_stream(test_name, record, streams_to_check, expectation):
+def test_check_stream(test_name, record, streams_to_check, stream_slice, expectation):
     stream = MagicMock()
     stream.name = "s1"
-    stream.read_records.return_value = iter([record])
+    stream.stream_slices.return_value = iter([stream_slice])
+    stream.read_records.side_effect = mock_read_records({frozenset(stream_slice): iter([record])})
 
     source = MagicMock()
     source.streams.return_value = [stream]
@@ -38,3 +40,7 @@ def test_check_stream(test_name, record, streams_to_check, expectation):
     else:
         with pytest.raises(ValueError):
             check_stream.check_connection(source, logger, config)
+
+
+def mock_read_records(responses, default_response=None, **kwargs):
+    return lambda stream_slice, sync_mode: responses[frozenset(stream_slice)] if frozenset(stream_slice) in responses else default_response