feat(cdk): add schema filter to DynamicSchemaLoader #550
Conversation
|
/autofix
|
📝 WalkthroughWalkthroughA new optional Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant DeclarativeStream
participant ModelToComponentFactory
participant DynamicSchemaLoader
participant SchemaFilter
User->>DeclarativeStream: Define stream with DynamicSchemaLoader (with optional schema_filter)
DeclarativeStream->>ModelToComponentFactory: Create DynamicSchemaLoader
ModelToComponentFactory->>DynamicSchemaLoader: Instantiate (schema_filter passed if present)
DynamicSchemaLoader->>DynamicSchemaLoader: get_json_schema()
DynamicSchemaLoader->>DynamicSchemaLoader: Extract schema properties
alt schema_filter is set
DynamicSchemaLoader->>SchemaFilter: filter_records(properties)
SchemaFilter-->>DynamicSchemaLoader: Filtered properties
end
DynamicSchemaLoader->>DynamicSchemaLoader: Transform filtered properties
DynamicSchemaLoader-->>DeclarativeStream: Return filtered/transformed schema
Would you like me to help draft an example configuration snippet for the Note ⚡️ AI Code Reviews for VS Code, Cursor, WindsurfCodeRabbit now has a plugin for VS Code, Cursor and Windsurf. This brings AI code reviews directly in the code editor. Each commit is reviewed immediately, finding bugs before the PR is raised. Seamless context handoff to your AI code agent ensures that you can easily incorporate review feedback. Note ⚡️ Faster reviews with cachingCodeRabbit now supports caching for code and dependencies, helping speed up reviews. This means quicker feedback, reduced wait times, and a smoother review experience overall. Cached data is encrypted and stored securely. This feature will be automatically enabled for all accounts on May 16th. To opt out, configure 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
⏰ 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
🧹 Nitpick comments (2)
airbyte_cdk/sources/declarative/resolvers/components_resolver.py (1)
25-25: Introducecreate_or_updateflag in ComponentMappingDefinition
Great to see a control flag added for mapping behavior! Should we simplify the annotation tocreate_or_update: bool = Falseinstead ofOptional[bool], since the default is alwaysFalse? wdyt?airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
2417-2419: The schema_filter description needs improvement, wdyt?The description is currently just a placeholder ("placeholder"). Consider providing a more informative description that explains the purpose of this field, similar to how other filter fields are documented in this file.
- schema_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( - None, description="placeholder", title="Schema Filter" - ) + schema_filter: Optional[Union[RecordFilter, CustomRecordFilter]] = Field( + None, + description="Filter applied to schema properties during schema generation. Properties will be included only if they satisfy the filter condition.", + title="Schema Filter" + )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Cache: Disabled due to data retention organization setting
Knowledge Base: Disabled due to data retention organization setting
📒 Files selected for processing (5)
airbyte_cdk/sources/declarative/declarative_component_schema.yaml(1 hunks)airbyte_cdk/sources/declarative/models/declarative_component_schema.py(2 hunks)airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py(1 hunks)airbyte_cdk/sources/declarative/resolvers/components_resolver.py(2 hunks)airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py(4 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (2)
airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
airbyte_cdk/sources/declarative/extractors/record_filter.py (1)
RecordFilter(17-49)
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py (2)
airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py (1)
DynamicSchemaLoader(120-300)airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
DynamicSchemaLoader(2410-2439)
🪛 GitHub Actions: Linters
airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py
[error] 183-183: mypy error: Need type annotation for "filtered_properties" (hint: "filtered_properties: dict[, ] = ...") [var-annotated]
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
[error] 2436-2436: mypy error: Argument 1 to "_create_component_from_model" of "ModelToComponentFactory" has incompatible type "RecordFilter | CustomRecordFilter | None"; expected "BaseModel" [arg-type]
⏰ Context from checks skipped due to timeout of 90000ms (4)
- GitHub Check: Check: 'source-amplitude' (skip=false)
- GitHub Check: Check: 'source-shopify' (skip=false)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
🔇 Additional comments (7)
airbyte_cdk/sources/declarative/resolvers/components_resolver.py (1)
38-38: Mirrorcreate_or_updatein ResolvedComponentMappingDefinition
Nice consistency between the definition and its resolved counterpart. Could you verify that downstream logic (e.g., in the component factory or resolver implementations) properly consumes this new flag? wdyt?airbyte_cdk/sources/declarative/declarative_component_schema.yaml (1)
2321-2326: Add optionalschema_filterto DynamicSchemaLoader schema
This looks like a solid enhancement for selective field inclusion. Can you confirm that omittingschema_filterstill results in the expected “no filter” behavior and that it’s correctly wired in the loader implementation? wdyt?airbyte_cdk/sources/declarative/models/declarative_component_schema.py (1)
2167-2168: Good improvement to the schema loader descriptionThe updated description now clearly explains what happens when multiple schema loaders are used in a declarative stream, particularly that properties will be merged with precedence given to loaders defined earlier in the list.
airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py (4)
13-13: Appropriate import additionThe import for RecordFilter is correctly added and needed for the new schema filtering functionality.
130-130: Good addition of the schema_filter parameterThe optional schema_filter parameter is correctly added to the DynamicSchemaLoader class.
156-156: Correct application of filtering before transformationThe get_json_schema method now correctly applies filtering before transformation, which is the appropriate order.
183-189: Consider empty stream state for filter_records, wdyt?The filter_records method is called with an empty dictionary as the stream_state. Is this intentional? If a more appropriate state should be used, consider passing it.
Given that this is a schema filtering context rather than a normal data stream context, is passing an empty dictionary for stream_state appropriate for your use case? This might be fine if schema filtering doesn't need stream state information, but it's worth confirming.
🧰 Tools
🪛 GitHub Actions: Linters
[error] 183-183: mypy error: Need type annotation for "filtered_properties" (hint: "filtered_properties: dict[, ] = ...") [var-annotated]
airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py
Outdated
Show resolved
Hide resolved
airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
Outdated
Show resolved
Hide resolved
…byte-python-cdk into lazebnyi/add-schema-filter
|
/autofix
|
There was a problem hiding this comment.
Actionable comments posted: 0
🔭 Outside diff range comments (1)
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py (1)
1-1: 💡 Verification agent🧩 Analysis chain
Small formatting issue flagged by pipeline
The pipeline shows a Ruff formatting issue that needs to be fixed.
🏁 Script executed:
#!/bin/bash # Run ruff format check to see what needs to be fixed ruff format --check unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py --diffLength of output: 677
Add missing trailing comma in schema_filter condition
Could you add a comma at the end of the
"condition"line so that Ruff’s formatting check passes? wdyt?
- File:
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
Around line 70:- "condition": "{{ 'filtered_field' not in record }}" + "condition": "{{ 'filtered_field' not in record }}",🧰 Tools
🪛 GitHub Actions: Linters
[error] 1-1: Ruff formatting check failed. File would be reformatted. Run 'ruff format' to fix code style issues.
🧹 Nitpick comments (1)
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py (1)
338-355: Consider explicit verification of filtering behaviorThe test verifies the schema content implicitly (by checking the final expected schema doesn't contain 'filtered_field'). Would it be clearer to add an explicit assertion that 'filtered_field' is not in the properties?
Something like:
assert "filtered_field" not in actual_catalog.streams[0].json_schema["properties"]This would make the test's purpose more obvious and serve as documentation of what's being tested, wdyt?
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Cache: Disabled due to data retention organization setting
Knowledge Base: Disabled due to data retention organization setting
📒 Files selected for processing (4)
airbyte_cdk/sources/declarative/models/declarative_component_schema.py(2 hunks)airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py(1 hunks)airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py(4 hunks)unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py(2 hunks)
🚧 Files skipped from review as they are similar to previous changes (3)
- airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py
- airbyte_cdk/sources/declarative/models/declarative_component_schema.py
- airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
🧰 Additional context used
🪛 GitHub Actions: Linters
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
[error] 1-1: Ruff formatting check failed. File would be reformatted. Run 'ruff format' to fix code style issues.
⏰ Context from checks skipped due to timeout of 90000ms (9)
- GitHub Check: Check: 'source-pokeapi' (skip=false)
- GitHub Check: Check: 'source-hardcoded-records' (skip=false)
- GitHub Check: Check: 'source-amplitude' (skip=false)
- GitHub Check: Check: 'source-shopify' (skip=false)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (Fast)
- GitHub Check: SDM Docker Image Build
- GitHub Check: Analyze (python)
🔇 Additional comments (2)
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py (2)
71-74: Schema filter implementation looks good!The implementation adds a filter to exclude records with 'filtered_field' key, which aligns with the PR objective to filter properties in the dynamic schema loader. It's clear and concise.
Would adding a brief comment explaining the purpose of this schema filter make it more maintainable for future developers, wdyt?
397-397: Test data correctly updated to validate filteringThis new field entry is well placed to test if the schema_filter successfully excludes fields that shouldn't be included in the final schema. The test setup effectively validates that 'filtered_field' doesn't appear in the expected schema properties.
|
/autofix
|
What
Resolved: https://github.com/airbytehq/airbyte-internal-issues/issues/13045
In some cases, when using the dynamic schema loader endpoint, the provider may return all possible fields that could be present. To keep the schema up to date and ensure that the
propertiessection only includes the fields that were actually requested, we need to filter the returned values accordingly.How
Added filtering behavior to the schema loader using the
schema_filter.Summary by CodeRabbit
New Features
Documentation
Tests