fix: revert builder concurrency and pagination url changes #694
Conversation
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. Testing This CDK VersionYou can test this version of the CDK using the following: # Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@lmossman/revert-cdk-changes#egg=airbyte-python-cdk[dev]' --help
# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch lmossman/revert-cdk-changesHelpful ResourcesPR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
|
There was a problem hiding this comment.
Pull Request Overview
This PR reverts builder concurrency and pagination URL changes by removing the concurrent framework components from test files and connector builder handling, reverting to the previous synchronous implementation for testing purposes.
- Removes concurrent processing components including
ConcurrentMessageRepository,PartitionLogger, and related concurrency infrastructure - Simplifies PartitionReader to no longer handle cursor operations during partition processing
- Reverts connector builder to use
ManifestDeclarativeSourceinstead ofConcurrentDeclarativeSourcefor test operations
Reviewed Changes
Copilot reviewed 20 out of 20 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| test_partition_reader.py | Removes cursor-related test cases and simplifies partition processing tests |
| test_concurrent_read_processor.py | Adds cursor close_partition calls and error handling tests |
| stream_facade_builder.py | Simplifies ConcurrentSource constructor call to use positional arguments |
| test_dynamic_schema_loader.py | Replaces TestLimits with ModelToComponentFactory for cache disabling |
| test_http_requester.py | Removes test cases for URL path handling with different URL patterns |
| test_connector_builder_handler.py | Reverts from ConcurrentDeclarativeSource to ManifestDeclarativeSource usage |
| slice_logger.py | Removes comment about concurrent CDK cleanup |
| partition_reader.py | Removes PartitionLogger class and cursor operations from partition processing |
| concurrent_repository.py | Completely removes ConcurrentMessageRepository implementation |
| stream_slicer_test_read_decorator.py | Adds missing imports for proper type handling |
| declarative_partition_generator.py | Removes slice_limit parameter and StreamSlicerTestReadDecorator usage |
| http_requester.py | Simplifies URL joining logic |
| model_to_component_factory.py | Removes MAX_SLICES constant |
| concurrent_declarative_source.py | Removes TestLimits class and queue-based messaging infrastructure |
| concurrent_source.py | Moves queue creation to read method and removes partition logger |
| concurrent_read_processor.py | Adds cursor close_partition handling in partition completion |
| helpers.py | Removes state blob conversion utility functions |
| main.py | Simplifies source creation call |
| connector_builder_handler.py | Moves TestLimits class here and reverts to ManifestDeclarativeSource |
Comments suppressed due to low confidence (2)
unit_tests/connector_builder/test_connector_builder_handler.py:1269
- The expected exception type has been changed from 'StreamThreadException' to 'AirbyteTracedException'. This change should be verified to ensure it aligns with the actual exception behavior after removing concurrent processing.
"AirbyteTracedException",
unit_tests/connector_builder/test_connector_builder_handler.py:1275
- The expected exception type has been changed from 'StreamThreadException' to 'InvalidSchema'. This change should be verified to ensure it aligns with the actual exception behavior after removing concurrent processing.
"InvalidSchema",
airbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.py
Show resolved
Hide resolved
📝 WalkthroughWalkthroughThis change removes concurrency and queue management from the connector builder and related sources. It replaces Changes
Sequence Diagram(s)sequenceDiagram
participant Main as main.py
participant Handler as connector_builder_handler.py
participant Source as ManifestDeclarativeSource
participant Factory as ModelToComponentFactory
Main->>Handler: handle_request(config, limits)
Handler->>Factory: Create ModelToComponentFactory(limits)
Handler->>Source: Instantiate ManifestDeclarativeSource(factory)
Source-->>Handler: Source instance
Handler-->>Main: Serialized result
Estimated code review effort🎯 4 (Complex) | ⏱️ ~40 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Would you like to review any particular area in more detail, or should I focus on the concurrency removal and test updates, wdyt? Note ⚡️ Unit Test Generation is now available in beta!Learn more here, or try it out under "Finishing Touches" below. ✨ Finishing Touches
🧪 Generate unit tests
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. 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 (1)
airbyte_cdk/sources/declarative/concurrent_declarative_source.py (1)
68-93: Excellent constructor refactoring!The optional
component_factoryparameter provides great flexibility while maintaining sensible defaults. The comment about disabling RFR with synthetic cursors helps explain the design decision.One observation - the default factory disables resumable full refresh. Should we document this behavior change more prominently for users who might be expecting RFR to work with concurrent sources, wdyt?
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (20)
airbyte_cdk/connector_builder/connector_builder_handler.py(3 hunks)airbyte_cdk/connector_builder/main.py(1 hunks)airbyte_cdk/connector_builder/test_reader/helpers.py(2 hunks)airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py(2 hunks)airbyte_cdk/sources/concurrent_source/concurrent_source.py(5 hunks)airbyte_cdk/sources/declarative/concurrent_declarative_source.py(2 hunks)airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py(0 hunks)airbyte_cdk/sources/declarative/requesters/http_requester.py(1 hunks)airbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.py(2 hunks)airbyte_cdk/sources/declarative/stream_slicers/stream_slicer_test_read_decorator.py(1 hunks)airbyte_cdk/sources/message/concurrent_repository.py(0 hunks)airbyte_cdk/sources/streams/concurrent/partition_reader.py(2 hunks)airbyte_cdk/sources/streams/concurrent/partitions/types.py(1 hunks)airbyte_cdk/sources/utils/slice_logger.py(0 hunks)unit_tests/connector_builder/test_connector_builder_handler.py(17 hunks)unit_tests/sources/declarative/requesters/test_http_requester.py(0 hunks)unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py(2 hunks)unit_tests/sources/streams/concurrent/scenarios/stream_facade_builder.py(1 hunks)unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py(5 hunks)unit_tests/sources/streams/concurrent/test_partition_reader.py(3 hunks)
💤 Files with no reviewable changes (4)
- airbyte_cdk/sources/utils/slice_logger.py
- airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py
- airbyte_cdk/sources/message/concurrent_repository.py
- unit_tests/sources/declarative/requesters/test_http_requester.py
🧰 Additional context used
🧠 Learnings (7)
📚 Learning: the files in `airbyte_cdk/cli/source_declarative_manifest/`, including `_run.py`, are imported from ...
Learnt from: aaronsteers
PR: airbytehq/airbyte-python-cdk#58
File: airbyte_cdk/cli/source_declarative_manifest/_run.py:62-65
Timestamp: 2024-11-15T01:04:21.272Z
Learning: The files in `airbyte_cdk/cli/source_declarative_manifest/`, including `_run.py`, are imported from another repository, and changes to these files should be minimized or avoided when possible to maintain consistency.
Applied to files:
airbyte_cdk/sources/declarative/stream_slicers/stream_slicer_test_read_decorator.pyunit_tests/sources/declarative/schema/test_dynamic_schema_loader.pyairbyte_cdk/sources/streams/concurrent/partitions/types.pyairbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.pyairbyte_cdk/connector_builder/test_reader/helpers.pyunit_tests/sources/streams/concurrent/test_partition_reader.pyairbyte_cdk/sources/concurrent_source/concurrent_source.pyairbyte_cdk/sources/declarative/concurrent_declarative_source.pyunit_tests/connector_builder/test_connector_builder_handler.pyairbyte_cdk/connector_builder/connector_builder_handler.py
📚 Learning: when code in `airbyte_cdk/cli/source_declarative_manifest/` is being imported from another repositor...
Learnt from: aaronsteers
PR: airbytehq/airbyte-python-cdk#58
File: airbyte_cdk/cli/source_declarative_manifest/spec.json:9-15
Timestamp: 2024-11-15T00:59:08.154Z
Learning: When code in `airbyte_cdk/cli/source_declarative_manifest/` is being imported from another repository, avoid suggesting modifications to it during the import process.
Applied to files:
airbyte_cdk/sources/declarative/stream_slicers/stream_slicer_test_read_decorator.pyunit_tests/sources/declarative/schema/test_dynamic_schema_loader.pyairbyte_cdk/sources/streams/concurrent/partitions/types.pyairbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.pyairbyte_cdk/connector_builder/test_reader/helpers.pyairbyte_cdk/sources/concurrent_source/concurrent_source.pyairbyte_cdk/sources/declarative/concurrent_declarative_source.pyunit_tests/connector_builder/test_connector_builder_handler.pyairbyte_cdk/connector_builder/connector_builder_handler.py
📚 Learning: when modifying the `yamldeclarativesource` class in `airbyte_cdk/sources/declarative/yaml_declarativ...
Learnt from: ChristoGrab
PR: airbytehq/airbyte-python-cdk#58
File: airbyte_cdk/sources/declarative/yaml_declarative_source.py:0-0
Timestamp: 2024-11-18T23:40:06.391Z
Learning: When modifying the `YamlDeclarativeSource` class in `airbyte_cdk/sources/declarative/yaml_declarative_source.py`, avoid introducing breaking changes like altering method signatures within the scope of unrelated PRs. Such changes should be addressed separately to minimize impact on existing implementations.
Applied to files:
airbyte_cdk/sources/declarative/stream_slicers/stream_slicer_test_read_decorator.pyunit_tests/sources/declarative/schema/test_dynamic_schema_loader.pyunit_tests/sources/streams/concurrent/scenarios/stream_facade_builder.pyairbyte_cdk/connector_builder/main.pyairbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.pyairbyte_cdk/connector_builder/test_reader/helpers.pyairbyte_cdk/sources/concurrent_source/concurrent_read_processor.pyairbyte_cdk/sources/concurrent_source/concurrent_source.pyairbyte_cdk/sources/declarative/concurrent_declarative_source.pyunit_tests/connector_builder/test_connector_builder_handler.pyairbyte_cdk/sources/streams/concurrent/partition_reader.pyairbyte_cdk/connector_builder/connector_builder_handler.py
📚 Learning: the custompageincrement class in unit_tests/source_declarative_manifest/resources/source_the_guardia...
Learnt from: aaronsteers
PR: airbytehq/airbyte-python-cdk#174
File: unit_tests/source_declarative_manifest/resources/source_the_guardian_api/components.py:21-29
Timestamp: 2025-01-13T23:39:15.457Z
Learning: The CustomPageIncrement class in unit_tests/source_declarative_manifest/resources/source_the_guardian_api/components.py is imported from another connector definition and should not be modified in this context.
Applied to files:
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.pyunit_tests/connector_builder/test_connector_builder_handler.pyairbyte_cdk/connector_builder/connector_builder_handler.py
📚 Learning: in the `airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py` file, the strict modu...
Learnt from: aaronsteers
PR: airbytehq/airbyte-python-cdk#174
File: airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py:1093-1102
Timestamp: 2025-01-14T00:20:32.310Z
Learning: In the `airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py` file, the strict module name checks in `_get_class_from_fully_qualified_class_name` (requiring `module_name` to be "components" and `module_name_full` to be "source_declarative_manifest.components") are intentionally designed to provide early, clear feedback when class declarations won't be found later in execution. These restrictions may be loosened in the future if the requirements for class definition locations change.
Applied to files:
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.pyunit_tests/connector_builder/test_connector_builder_handler.pyairbyte_cdk/connector_builder/connector_builder_handler.py
📚 Learning: in the airbytehq/airbyte-python-cdk repository, the `declarative_component_schema.py` file is auto-g...
Learnt from: pnilan
PR: airbytehq/airbyte-python-cdk#0
File: :0-0
Timestamp: 2024-12-11T16:34:46.319Z
Learning: In the airbytehq/airbyte-python-cdk repository, the `declarative_component_schema.py` file is auto-generated from `declarative_component_schema.yaml` and should be ignored in the recommended reviewing order.
Applied to files:
unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py
📚 Learning: in the typetransformer class, the data being transformed comes from api responses or source systems,...
Learnt from: ChristoGrab
PR: airbytehq/airbyte-python-cdk#221
File: airbyte_cdk/sources/utils/transform.py:0-0
Timestamp: 2025-01-16T00:50:39.069Z
Learning: In the TypeTransformer class, the data being transformed comes from API responses or source systems, so only standard JSON-serializable types are expected. The python_to_json mapping covers all expected types, and it's designed to fail fast (KeyError) on unexpected custom types rather than providing fallbacks.
Applied to files:
airbyte_cdk/connector_builder/test_reader/helpers.py
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (10)
- GitHub Check: Check: destination-motherduck
- GitHub Check: Check: source-intercom
- GitHub Check: Check: source-pokeapi
- GitHub Check: Check: source-shopify
- GitHub Check: Check: source-hardcoded-records
- GitHub Check: Pytest (Fast)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: SDM Docker Image Build
- GitHub Check: Analyze (python)
🔇 Additional comments (36)
airbyte_cdk/sources/streams/concurrent/partitions/types.py (1)
37-37: LGTM! Simplification aligns with the revert.Removing
AirbyteMessagefrom theQueueItemunion makes sense for reverting the concurrency changes. This simplifies the queue to handle only partition-related items and exceptions, which seems cleaner than mixing in raw Airbyte messages.unit_tests/sources/streams/concurrent/scenarios/stream_facade_builder.py (1)
53-53: Constructor calling convention updated - looks good!The change from keyword to positional arguments for the
ConcurrentSourceconstructor maintains the same argument order and values. This aligns well with simplifying the constructor interface as part of the revert. Nice and clean!unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py (2)
14-16: Import change looks good!Switching from
TestLimitstoModelToComponentFactoryaligns well with the broader refactor to move away from internal limit management inConcurrentDeclarativeSource.
361-363: Good test isolation improvement!Using
ModelToComponentFactory(disable_cache=True)instead ofTestLimits()not only aligns with the refactor but also explicitly addresses the caching concern mentioned in your comment. This should help prevent test interference - nice attention to detail! wdyt about this approach vs other test isolation strategies?airbyte_cdk/connector_builder/main.py (2)
94-94: Function call simplification looks clean!The
create_sourcecall now only takesconfigandlimitsas positional arguments, which suggests a nice simplification in the function signature. This aligns well with the broader refactor toManifestDeclarativeSource.
95-99: Minor formatting improvement!Moving the type ignore comment to the specific
.decode()call is more precise than applying it to the entire expression. Good attention to detail!airbyte_cdk/sources/declarative/stream_slicers/declarative_partition_generator.py (1)
85-89: The removal ofslice_limitis safe – no callers pass that parameterI checked all instantiations of
StreamSlicerPartitionGeneratorinconcurrent_declarative_source.pyand found none include aslice_limitargument (and no other references toslice_limitexist). This means we won’t break any callers by dropping it. LGTM – thanks for cleaning this up! wdyt?airbyte_cdk/connector_builder/test_reader/helpers.py (2)
8-8: Good cleanup of unused import.Removing
AirbyteStateBlobfrom the imports makes sense since the conversion logic was removed. Clean!
469-469: Redundant state conversion safely removedWe ran a full search and confirmed there are no remaining calls to
convert_state_blob_to_mapping. Since this helper already receives anAirbyteStateMessage, usinglatest_state_messagedirectly doesn’t change how state is passed in—it simply drops an unnecessary blob-to-mapping step.• No
convert_state_blob_to_mappingreferences found
• AllAirbyteStateBlobusages remain valid and untouched elsewhereLooks good to merge as-is—wdyt?
unit_tests/sources/streams/concurrent/test_partition_reader.py (2)
29-29: Nice simplification of the PartitionReader constructor.Removing the cursor and logger parameters makes the API much cleaner. The test setup is more straightforward now.
32-32: Test method calls properly updated.All the
process_partitioncalls have been correctly updated to remove the cursor parameter. The test coverage still validates the essential behaviors - empty partitions, successful processing, and exception handling.Also applies to: 43-43, 55-55
airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py (2)
102-102: Process partition call correctly simplified.The removal of the cursor parameter from
process_partitionaligns perfectly with the PartitionReader simplification. Clean change!
115-134: Enhanced error handling around cursor closing looks solid.The addition of proper exception handling around
cursor.close_partitionis a great improvement. The try/catch/finally structure ensures:
- Exceptions during cursor closing are properly flagged and reported
- Partition cleanup happens regardless of cursor errors
- Stream completion logic still executes properly
The use of
AirbyteTracedException.from_exception(...).as_sanitized_airbyte_message()for error reporting is the right approach here. Nice defensive coding! wdyt?unit_tests/connector_builder/test_connector_builder_handler.py (7)
20-22: LGTM on importing the default constants!Nice improvement to centralize these default values from the handler module instead of relying on
TestLimitsdefaults. This makes the tests more consistent with the actual implementation, wdyt?
60-60: Source import updated correctly.The switch from
ConcurrentDeclarativeSourcetoManifestDeclarativeSourcealigns perfectly with the PR's goal of reverting concurrency changes.
533-533: All source instantiations updated consistently.Great job consistently replacing
ConcurrentDeclarativeSourcewithManifestDeclarativeSourcethroughout the test file. The removal ofcatalogandstateparameters aligns with the simplified function signatures mentioned in the summary.Also applies to: 682-687, 694-694, 783-783, 820-842, 884-884, 948-948, 1010-1012, 1104-1104, 1152-1152, 1239-1239, 1335-1335, 1392-1392
973-975: Test parameters updated to use actual defaults.Excellent improvement! Using the actual default constants instead of
TestLimitsdefaults makes these parameterized tests more accurate and aligned with the real implementation behavior.Also applies to: 980-982
891-891: Simplified assertion looks good.The switch to
assert_called_once()is cleaner than checkingcall_count == 1. Minor but nice improvement in test readability!Also applies to: 896-896
1269-1270: Error expectations updated correctly.Good catch updating the expected error strings to match the new error handling behavior. This ensures the tests remain accurate after the error handling changes.
Also applies to: 1275-1276
307-354: Great addition of error handling test coverage!This new test method effectively covers the edge case where
close_partitionraises an exception. The test structure is solid with proper assertions for both the error trace and incomplete stream status. Nice improvement to the test coverage, wdyt?airbyte_cdk/sources/concurrent_source/concurrent_source.py (5)
7-7: Import cleanup looks good.Removing
Optionalfrom the typing imports and simplifying thePartitionReaderimport makes sense given the parameter removals. Clean and consistent!Also applies to: 19-19
62-68: Queue parameter removal centralizes queue management.The removal of the optional
queueparameter simplifies the factory method and centralizes queue creation to thereadmethod scope. This trades some injection flexibility for simpler queue management - seems like a reasonable design choice for this refactor, wdyt?
100-104: Well-reasoned local queue creation.Creating the queue locally with a fixed maxsize of 10,000 is a good approach to prevent unbounded growth. The detailed comment explaining the rationale and assumptions is particularly helpful. Nice documentation of the design decision!
107-107: PartitionReader instantiation simplified correctly.The removal of
PartitionLoggerfromPartitionReaderand using the local queue forPartitionEnqueueraligns well with the broader refactor to simplify partition processing. Clean simplification!Also applies to: 112-112
120-120: Queue consumption logic improved.Passing the queue explicitly is necessary now that it's local, and reordering the condition to check
is_done()first beforequeue.empty()is a nice logical optimization. Good attention to efficiency!Also applies to: 144-144
unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py (3)
182-182: Process partition calls updated correctly.Good job updating the
process_partitioncalls to remove theexpected_cursorargument. This keeps the tests aligned with the simplifiedPartitionReaderinterface after cursor handling was removed.Also applies to: 207-207
256-256: Cursor closure assertions add valuable coverage.Adding
assert_called_once()forclose_partitionin these partition completion tests provides good coverage of the cursor closure behavior. Nice improvement to verify this critical step happens correctly!Also applies to: 305-305, 382-382
307-354: Excellent error handling test addition!This new test method provides comprehensive coverage of the error scenario when
close_partitionraises an exception. The test structure is well-designed with proper mocking, execution, and verification of both error trace and stream status messages. Really strengthens the error handling coverage, wdyt?airbyte_cdk/sources/streams/concurrent/partition_reader.py (3)
1-11: LGTM! Clean import removal.The removal of the PartitionLogger import aligns well with the simplified partition processing approach.
21-25: Constructor simplification looks good!The removal of the
partition_loggerparameter streamlines the class interface nicely.
27-45: Excellent simplification of partition processing!The removal of cursor handling from this method creates a cleaner separation of concerns. The preserved exception handling ensures thread safety is maintained. This aligns perfectly with the PR's goal of removing concurrency management complexity.
airbyte_cdk/connector_builder/connector_builder_handler.py (4)
27-31: Nice use of constants for default values!Moving these to module-level constants improves maintainability. Have you considered if these values might need to be configurable via environment variables in the future, wdyt?
38-46: Great dataclass conversion!The transition to dataclass is clean and the
__test__ClassVar is a thoughtful addition to prevent pytest confusion. The field defaults align perfectly with the module constants.
48-56: Clean refactoring of get_limits!The use of
orfor default fallback values is more pythonic and concise.
78-93: Allcreate_sourcecallers now aligned with new signatureI’ve verified that every invocation of
create_sourcehas been updated to usecreate_source(config, limits)only:
- airbyte_cdk/connector_builder/main.py
- unit_tests/connector_builder/test_connector_builder_handler.py
No leftover calls to the old signature remain. Looks good to me—shall we merge? wdyt?
airbyte_cdk/sources/declarative/concurrent_declarative_source.py (1)
6-6: Clean import simplification!Removing
Queuefrom imports aligns with the removal of internal queue management.
airbyte_cdk/sources/declarative/stream_slicers/stream_slicer_test_read_decorator.py
Show resolved
Hide resolved
lmossman
changed the title
PyTest Results (Fast)3 695 tests - 4 3 684 ✅ - 4 6m 29s ⏱️ ±0s Results for commit 62c5eab. ± Comparison against base commit db1e4c0. This pull request removes 6 and adds 2 tests. Note that renamed tests count towards both. |
PyTest Results (Full)3 698 tests - 4 3 687 ✅ - 4 11m 35s ⏱️ -5s Results for commit 62c5eab. ± Comparison against base commit db1e4c0. This pull request removes 6 and adds 2 tests. Note that renamed tests count towards both. |
2 participants
Reverts the following 3 PRs, since they seem to be causing issues with builder connectors (see slack thread)
Summary by CodeRabbit
Refactor
Bug Fixes
Tests
Chores