feat(concurrent): do not raise when record does not have cursor value in concurrent cursor #96
Conversation
… concurrent cursor
📝 Walkthrough📝 WalkthroughWalkthroughThe changes in this pull request focus on enhancing error handling and logging mechanisms within the Changes
Possibly related PRs
Suggested reviewers
What do you think about the suggested reviewers? Do they align well with the changes made? 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
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: 0
🧹 Outside diff range and nitpick comments (3)
airbyte_cdk/sources/streams/concurrent/cursor.py (2)
219-225: LGTM! Consider being more specific with the exception handlingThe error handling looks good and aligns well with the PR objective. Would you consider catching
ValueErrorspecifically from extract_value to be more precise? Something like this, wdyt?try: cursor_value = self._extract_cursor_value(record) if most_recent_cursor_value is None or most_recent_cursor_value < cursor_value: self._most_recent_cursor_value_per_partition[record.associated_slice] = cursor_value -except ValueError: +except ValueError as e: + if "Could not find cursor field" not in str(e): + raise self._log_for_record_without_cursor_value()
469-474: LGTM! Consider log level adjustmentGreat implementation of the logging mechanism! The flag to prevent duplicate warnings is a nice touch. Would you consider using INFO level instead of WARNING since this is now an expected behavior rather than a warning condition? Just a thought, wdyt?
- LOGGER.warning( + LOGGER.info(unit_tests/sources/streams/concurrent/test_cursor.py (1)
104-117: Test looks good! Consider enhancing the coverageThe test effectively verifies that no exception is raised, but we could make it even more robust. Would you consider adding these assertions, wdyt?
cursor.observe( Record( data={"record_with_A_CURSOR_FIELD_KEY": "any value"}, associated_slice=partition.to_slice(), stream_name=_A_STREAM_NAME, ) ) - # did not raise + # Verify logger was triggered + assert cursor._should_be_synced_logger_triggered is True + + # Verify no cursor value was stored + assert partition.to_slice() not in cursor._most_recent_cursor_value_per_partition
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (2)
airbyte_cdk/sources/streams/concurrent/cursor.py(2 hunks)unit_tests/sources/streams/concurrent/test_cursor.py(1 hunks)
🔇 Additional comments (1)
airbyte_cdk/sources/streams/concurrent/cursor.py (1)
465-467: LGTM! Consistent error handling approach
The error handling here mirrors the approach in observe(), maintaining consistency in how missing cursor values are handled. The decision to sync records without cursor values makes sense for backward compatibility.
maxi297
changed the title
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Outside diff range and nitpick comments (3)
airbyte_cdk/sources/streams/concurrent/cursor.py (3)
230-236: Consider adding return type hint to improve type safety.The error handling looks good! Would you consider adding a return type hint to make the method signature more explicit? wdyt?
- def observe(self, record: Record) -> None: + def observe(self, record: Record) -> None:
476-478: Consider updating the docstring to reflect new behavior.The method now returns
Truewhen a cursor value is missing. Would you consider updating the docstring to document this important behavior? Something like:def should_be_synced(self, record: Record) -> bool: """ Determines if a record should be synced based on its cursor value. :param record: The record to evaluate - - :return: True if the record's cursor value falls within the sync boundaries + :return: True if the record's cursor value falls within the sync boundaries or if the cursor value is missing """
480-485: Consider adding a docstring to the new method.The implementation looks good! Would you consider adding a docstring to explain the purpose of the flag and the logging behavior? Something like:
def _log_for_record_without_cursor_value(self) -> None: + """ + Logs a warning when a cursor value is missing in a record. + The warning is logged only once per stream to prevent log spam. + """ if not self._should_be_synced_logger_triggered:
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (1)
airbyte_cdk/sources/streams/concurrent/cursor.py(4 hunks)
🔇 Additional comments (1)
airbyte_cdk/sources/streams/concurrent/cursor.py (1)
189-189: LGTM! Type annotation enhancement looks good.
The expanded type annotation better reflects the possible key types and aligns with the PR's goal of handling records without cursor values.
maxi297
changed the title
maxi297
deleted the
issue-10550/do-not-break-when-record-does-not-have-cursor-value-on-concurrent
branch
aaronsteers
changed the title
Partially resolves:
What
Following https://github.com/airbytehq/airbyte-internal-issues/issues/10550 change on source-pipedrive, we have seen issues when the cursor field was not provided.
For now, we will assume that we need to sync those records but they won't influence the
most_recent_cursor_valueHow
By not raising if the cursor value is not provided
Summary by CodeRabbit
New Features
ConcurrentCursorclass.Bug Fixes
Tests
observemethod behaves correctly when no cursor value is provided, along with updates to existing tests for comprehensive coverage.