feat(low-code CDK): Add ConditionalStreams component to low-code framework#592
feat(low-code CDK): Add ConditionalStreams component to low-code framework#592Brian Lai (brianjlai) merged 3 commits intomainfrom
Conversation
Brian Lai (brianjlai)
requested review from
David Gold (dbgold17) and
Patrick Nilan (pnilan)
📝 WalkthroughWalkthroughThis update introduces a new Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant ManifestDeclarativeSource
participant ConditionalStreams
participant DeclarativeStream
User->>ManifestDeclarativeSource: streams(config)
ManifestDeclarativeSource->>ManifestDeclarativeSource: _stream_configs(manifest, config)
alt Stream is ConditionalStreams
ManifestDeclarativeSource->>ConditionalStreams: Evaluate condition with config
alt Condition is True
ConditionalStreams->>ManifestDeclarativeSource: Return nested streams
else Condition is False
ConditionalStreams->>ManifestDeclarativeSource: Return empty
end
else Stream is DeclarativeStream
ManifestDeclarativeSource->>DeclarativeStream: Add to stream list
end
ManifestDeclarativeSource-->>User: List of enabled streams
Suggested reviewers
Would you like to add more sample manifests demonstrating 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
⏰ Context from checks skipped due to timeout of 90000ms (9)
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🔭 Outside diff range comments (1)
unit_tests/sources/declarative/parsers/test_model_to_component_factory.py (1)
4806-4911:⚠️ Potential issueBroken YAML → the test will explode before it even runs
The YAML embedded in
test_conditional_streamsis not valid:
- Several nested keys are indented at the same level as their parent (
type,page_size_option,record_selector, …), so the parser will raise aScannerError.- The
retrieverblock is missing a requiredtype: SimpleRetriever.requesteris missing its owntype: HttpRequester.record_selector/extractorare mis-nested underrequester.Because
_parse()usesyaml.safe_load, the test will fail long before the CDK logic is exercised.Could we patch the snippet like this, wdyt?
- retriever: - paginator: - type: "DefaultPaginator" + retriever: + type: SimpleRetriever + paginator: + type: "DefaultPaginator" page_size_option: type: RequestOption inject_into: request_parameter field_name: page_size @@ - requester: - url_base: "https://testing.com" - path: "/api/v1/lists" + requester: + type: HttpRequester + url_base: "https://testing.com" + path: "/api/v1/lists" @@ - record_selector: - extractor: - field_path: ["result"] + record_selector: + type: RecordSelector + extractor: + type: DpathExtractor + field_path: ["result"]The same fixes apply to the
conditions_streamdefinition further down. Once the YAML parses, the rest of the assertions should execute as intended.
♻️ Duplicate comments (1)
airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
2206-2208: Same parity question forDeclarativeSource2Same comment as above: aligning allowed inner types would keep the two “streams” shapes interchangeable no matter the nesting depth, wdyt?
🧹 Nitpick comments (10)
.pre-commit-config.yaml (1)
77-77: Great addition of type stub packages!Adding
types-requestsandtypes-PyYAMLwill improve mypy coverage for those libraries. Could we alphabetize the list ("types-PyYAML", "types-requests") for consistency as we grow this list? wdyt?airbyte_cdk/sources/declarative/declarative_component_schema.yaml (1)
428-457: Align sub-schemastreamsitems with rootstreamsanyOf?
Currently,ConditionalStreams.streams.itemsonly referencesDeclarativeStream. Should we expand this to ananyOfmatching the rootstreams(e.g., includingStateDelegatingStreamor even nestedConditionalStreams) to maintain consistency? wdyt?airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (4)
159-161: Minor ordering nit — keep import list alphabetised?The bulk of this file keeps the
from … importblocks alphabetised. PlacingConditionalStreamsafterConcurrencyLevelbreaks that visual order slightly. Would you consider moving it to preserve the alphabetical grouping, or is there a reason to keep it here, wdyt?
662-663: Mapping entry positionSame nitpick as above – inserting the new entry right after
ConcurrencyLevelModelworks, but it slightly disturbs the (mostly) alphabetic order of the long mapping. Re-ordering keeps future merges cleaner. Up to you if you find that worthwhile, wdyt?
1626-1641: Consider propagating$parametersto nested streams
create_conditional_streamsis straightforward – it avoids instantiating nested streams when the condition is false. One thought: when the condition is true you delegate to_create_component_from_model(stream, …, **kwargs), but you don’t forwardmodel.parameters. If a parentConditionalStreamsblock is meant to share parameters with all its child streams, passing them might be helpful:- self._create_component_from_model(stream, config=config, **kwargs) + self._create_component_from_model( + stream, + config=config, + parameters={**(model.parameters or {}), **kwargs.get("parameters", {})}, + **{k: v for k, v in kwargs.items() if k != "parameters"}, + )That keeps child-stream behaviour consistent with other composite components (e.g.,
AddFields). What do you think?
3173-3182: Type annotations could beOptional[str]
model.requester.urlorurl_basecan beNone, so annotating_url/_url_baseas plainstrwill still silence mypy but doesn’t convey potentialNone. Would switching toOptional[str](orstr | Nonein 3.10+) be clearer?- _url: str = ( + _url: Optional[str] = ( ... - _url_base: str = ( + _url_base: Optional[str] = ((Mypy won’t complain because you immediately coalesce, but the annotation better matches reality.)
unit_tests/sources/declarative/test_manifest_declarative_source.py (3)
1149-1274: Could we factor the gigantic manifest into a small helper or reuse the existing_declarative_streamfixture to stay DRY?The in-line manifest repeats >200 lines of boiler-plate that already live in helpers above. Extracting only the few differences (
studentsvsclassrooms/clubs, plus theConditionalStreamswrapper) into a local fixture or helper would dramatically shrink the test and make the intent pop out, while reducing maintenance burden when the default stream template changes – wdyt?
1278-1282: Are the repeated “module is insys.modules” assertions still useful here?These four asserts were valuable in an earlier test to prove the monkey-patching worked, but they don’t add new coverage in this scenario and make the test longer. Removing them would keep the focus on the conditional-stream logic – what do you think?
1283-1300: Shall we assert absence of conditional streams whenis_sandboxisFalse?For the
Falsecase we only check the list length. Addingassert all(s.name not in {"classrooms", "clubs"} for s in actual_streams)would protect against an ordering bug accidentally giving length 1 but still returning a wrong stream, strengthening the test with minimal noise – wdyt?
airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
2287-2301: Consider supporting nestedConditionalStreamsorStateDelegatingStreamNice addition! Two small follow-ups you may want to ponder:
Nested conditionals – allowing
streams: List[Union[ConditionalStreams, DeclarativeStream, StateDelegatingStream]]would let authors build more complex tree-like conditions without extra boilerplate.
streamsdescription says “during an operation”, maybe change to “during a sync” for consistency with other docs.Would expanding the union and tweaking the phrasing make sense here, wdyt?
🧰 Tools
🪛 Pylint (3.3.7)
[refactor] 2287-2287: Too few public methods (0/2)
(R0903)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
.pre-commit-config.yaml(1 hunks)airbyte_cdk/sources/declarative/declarative_component_schema.yaml(2 hunks)airbyte_cdk/sources/declarative/manifest_declarative_source.py(4 hunks)airbyte_cdk/sources/declarative/models/declarative_component_schema.py(5 hunks)airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py(5 hunks)unit_tests/sources/declarative/parsers/test_model_to_component_factory.py(2 hunks)unit_tests/sources/declarative/test_manifest_declarative_source.py(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (5)
airbyte_cdk/sources/declarative/interpolation/interpolated_boolean.py (2)
InterpolatedBoolean(29-66)eval(45-66)airbyte_cdk/sources/declarative/interpolation/interpolated_string.py (1)
eval(35-55)airbyte_cdk/sources/declarative/models/declarative_component_schema.py (18)
ConditionalStreams(2287-2300)Config(136-137)Config(150-151)Config(164-165)Config(178-179)Config(196-197)Config(210-211)Config(224-225)Config(238-239)Config(252-253)Config(266-267)Config(280-281)Config(294-295)Config(310-311)Config(324-325)Config(338-339)Config(372-373)DeclarativeStream(2328-2397)airbyte_cdk/sources/declarative/declarative_stream.py (1)
DeclarativeStream(32-241)airbyte_cdk/sources/declarative/manifest_declarative_source.py (1)
streams(295-343)
🪛 Pylint (3.3.7)
airbyte_cdk/sources/declarative/models/declarative_component_schema.py
[refactor] 2287-2287: Too few public methods (0/2)
(R0903)
🔇 Additional comments (5)
airbyte_cdk/sources/declarative/declarative_component_schema.yaml (1)
27-31: Great addition ofConditionalStreamssupport in rootstreams
TheanyOffor top-levelstreamsnow includes the newConditionalStreamsschema, enabling dynamic stream grouping based on config, wdyt?airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (1)
108-108: Import looks goodNice addition – the factory now has direct access to
InterpolatedBoolean.unit_tests/sources/declarative/parsers/test_model_to_component_factory.py (1)
62-62: Import looks fine 👍The extra import of
ConditionalStreamsis in-line with the new feature and keeps the import grouping intact; nothing to change here.airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
2959-2960:update_forward_refsorder looks safe
ConditionalStreams.update_forward_refs()is invoked after all model classes are defined, so forward references resolve fine. Looks good to me.airbyte_cdk/sources/declarative/manifest_declarative_source.py (1)
386-406: Good handling of ConditionalStreamsModel!The implementation correctly extracts nested streams from ConditionalStreamsModel and applies cache configuration to them. The use of an empty list as default for
stream_config.get("streams") or []is a nice defensive programming touch.
2 participants
Resolves https://github.com/airbytehq/airbyte-internal-issues/issues/13285
What
source-tiktok-marketinghas a special piece of required functionality where only a smaller subset of streams are exposed to customers when they are using a sandbox account.There is currently no way to do this in the CDK or using custom components, so this change allows for certain streams to be emitted based on a condition of the incoming config.
How
I introduced a new component called
ConditionalStreamswhich is comprised of aconditionand a set ofstreamsthat will be available to users if the condition is met. And now the originalstreamsfield at the top level of the manifest takes in a list of streams orConditionalStreams.Summary by CodeRabbit
New Features
Tests
Chores