Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[low-code connectors] Get parent stream's full slice #15631

Merged
merged 5 commits into from
Aug 18, 2022
Merged

[low-code connectors] Get parent stream's full slice #15631

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
299c5a0
always access parent stream using full_refresh mode
girarda Aug 13, 2022
821fdfd
Update test
girarda Aug 13, 2022
ae2511d
fix substream slicer
girarda Aug 17, 2022
752ccee
Merge branch 'master' into alex/fixSubstreamIncremental
girarda Aug 17, 2022
b3685a1
bump
girarda Aug 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions airbyte-cdk/python/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
# Changelog

## 0.1.76
- Bugfix: Correctly set parent slice stream for sub-resource streams

## 0.1.75
- Improve `filter_secrets` skip empty secret

Expand Down
2 changes: 1 addition & 1 deletion airbyte-cdk/python/airbyte_cdk/sources/declarative/stream_slicers/substream_slicer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ def stream_slices(self, sync_mode: SyncMode, stream_state: StreamState) -> Itera
stream_state_field = parent_stream_config.stream_slice_field
for parent_stream_slice in parent_stream.stream_slices(sync_mode=sync_mode, cursor_field=None, stream_state=stream_state):
empty_parent_slice = True
parent_slice = parent_stream_slice.get("slice")
parent_slice = parent_stream_slice
Copy link
Contributor Author

@girarda girarda Aug 17, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is a problem for sub-resources with an incremental parent stream.
this bug sneaked in during the stream slicer refactor because the parent stream slices have a "slice" field in the tests https://github.com/airbytehq/airbyte/pull/15631/files#diff-e46344fad49c4a5f8e385cf45a805d7117107c4b0e1b806720a7416cf397e853L16-L17


for parent_record in parent_stream.read_records(
sync_mode=SyncMode.full_refresh, cursor_field=None, stream_slice=parent_stream_slice, stream_state=None
Expand Down
2 changes: 1 addition & 1 deletion airbyte-cdk/python/setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

setup(
name="airbyte-cdk",
version="0.1.75",
version="0.1.76",
description="A framework for writing Airbyte Connectors.",
long_description=README,
long_description_content_type="text/markdown",
Expand Down
26 changes: 14 additions & 12 deletions airbyte-cdk/python/unit_tests/sources/declarative/stream_slicers/test_substream_slicer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ def read_records(
stream_slice: Mapping[str, Any] = None,
stream_state: Mapping[str, Any] = None,
) -> Iterable[Mapping[str, Any]]:
# The parent stream's records should always be read as full refresh
assert sync_mode == SyncMode.full_refresh
if not stream_slice:
yield from self._records
else:
Expand All @@ -64,7 +66,7 @@ def read_records(
stream=MockStream([{}], [], "first_stream"), parent_key="id", stream_slice_field="first_stream_id", options={}
)
],
[{"first_stream_id": None, "parent_slice": None}],
[{"first_stream_id": None, "parent_slice": {}}],
),
(
"test_single_parent_slices_with_records",
Expand All @@ -76,7 +78,7 @@ def read_records(
options={},
)
],
[{"first_stream_id": 1, "parent_slice": None}, {"first_stream_id": 2, "parent_slice": None}],
[{"first_stream_id": 1, "parent_slice": {}}, {"first_stream_id": 2, "parent_slice": {}}],
),
(
"test_with_parent_slices_and_records",
Expand All @@ -89,10 +91,10 @@ def read_records(
)
],
[
{"parent_slice": "first", "first_stream_id": 0},
{"parent_slice": "first", "first_stream_id": 1},
{"parent_slice": "second", "first_stream_id": 2},
{"parent_slice": "third", "first_stream_id": None},
{"parent_slice": {"slice": "first"}, "first_stream_id": 0},
{"parent_slice": {"slice": "first"}, "first_stream_id": 1},
{"parent_slice": {"slice": "second"}, "first_stream_id": 2},
{"parent_slice": {"slice": "third"}, "first_stream_id": None},
],
),
(
Expand All @@ -112,12 +114,12 @@ def read_records(
),
],
[
{"parent_slice": "first", "first_stream_id": 0},
{"parent_slice": "first", "first_stream_id": 1},
{"parent_slice": "second", "first_stream_id": 2},
{"parent_slice": "third", "first_stream_id": None},
{"parent_slice": "second_parent", "second_stream_id": 10},
{"parent_slice": "second_parent", "second_stream_id": 20},
{"parent_slice": {"slice": "first"}, "first_stream_id": 0},
{"parent_slice": {"slice": "first"}, "first_stream_id": 1},
{"parent_slice": {"slice": "second"}, "first_stream_id": 2},
{"parent_slice": {"slice": "third"}, "first_stream_id": None},
{"parent_slice": {"slice": "second_parent"}, "second_stream_id": 10},
{"parent_slice": {"slice": "second_parent"}, "second_stream_id": 20},
],
),
],
Expand Down
2 changes: 1 addition & 1 deletion airbyte-integrations/connector-templates/source-configuration-based/setup.py.hbs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from setuptools import find_packages, setup

MAIN_REQUIREMENTS = [
"airbyte-cdk~=0.1.73",
"airbyte-cdk~=0.1.76",
]

TEST_REQUIREMENTS = [
Expand Down