feat: support nested replication keys via dotted path#3690
Open
daria-hrabar wants to merge 19 commits into
Open
feat: support nested replication keys via dotted path#3690daria-hrabar wants to merge 19 commits into
daria-hrabar wants to merge 19 commits into
Conversation
Add a get_nested_value() utility function to singer_sdk/helpers/_util.py that resolves dotted key paths (e.g. 'attributes.updated') against a nested dictionary by traversing each level in sequence.
…ation_key Update the is_timestamp_replication_key property in core.py to traverse >> nested schema properties using a dotted key path instead of performing a single-level dictionary lookup.
…ream_state Update _increment_stream_state() in core.py to extract nested replication key values before passing the record to the state manager.
…key fix Add verifying_fix_1198.py to confirm that the fix for issue meltano#1198 works correctly. The script uses get_nested_value() directly to traverse a dotted replication key path against fake records with nested timestamps.
Contributor
Reviewer's GuideAdds nested replication key support using dotted paths by introducing a generic nested-dict accessor, wiring it into stream state incrementing and replication key type detection, and covering the behavior with tests plus example scripts that reproduce and verify Issue #1198. Sequence diagram for nested replication key state updatesequenceDiagram
participant Stream
participant get_nested_value
participant StateManager
Stream->>Stream: _increment_stream_state(latest_record, context)
alt replication_method == REPLICATION_INCREMENTAL and '.' in replication_key
Stream->>get_nested_value: get_nested_value(latest_record, replication_key)
get_nested_value-->>Stream: value
Stream->>StateManager: increment_state({replication_key: value}, context, replication_key)
else replication_method == REPLICATION_INCREMENTAL and no '.' in replication_key
Stream->>StateManager: increment_state(latest_record, context, replication_key)
end
File-Level Changes
Assessment against linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues, and left some high level feedback:
- The standalone scripts
reproducing_issue_1198.pyandverifying_fix_1198.pylook like ad‑hoc debugging helpers; consider moving them under an examples/tests directory or omitting them from the library package to avoid shipping them as top-level modules. - The nested schema traversal in
is_timestamp_replication_keyassumes each intermediate key corresponds to an object withproperties; if your JSON Schemas can use patterns likeadditionalPropertiesor arrays of objects, it may be worth hardening this traversal or clearly constraining which schema shapes are supported for nested replication keys.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The standalone scripts `reproducing_issue_1198.py` and `verifying_fix_1198.py` look like ad‑hoc debugging helpers; consider moving them under an examples/tests directory or omitting them from the library package to avoid shipping them as top-level modules.
- The nested schema traversal in `is_timestamp_replication_key` assumes each intermediate key corresponds to an object with `properties`; if your JSON Schemas can use patterns like `additionalProperties` or arrays of objects, it may be worth hardening this traversal or clearly constraining which schema shapes are supported for nested replication keys.
## Individual Comments
### Comment 1
<location path="singer_sdk/helpers/_util.py" line_range="78-87" />
<code_context>
return datetime.datetime.now(datetime.timezone.utc)
+
+
+def get_nested_value(record: dict, dotted_key: str) -> t.Any | None: # noqa: ANN401
+ """Retrieve a value from a nested dictionary using a dotted key path.
+
+ Args:
+ record: The record dictionary to traverse.
+ dotted_key: A dot-separated key path (e.g. ``"attributes.updated"``).
+
+ Returns:
+ The value at the nested path, or ``None`` if any level is missing.
+
+ Examples:
+ >>> get_nested_value({"a": {"b": 1}}, "a.b")
+ 1
+ >>> get_nested_value({"a": 1}, "a.b")
+ None
+ """
+ keys = dotted_key.split(".")
+ current: t.Any = record
+ for key in keys:
+ if not isinstance(current, dict):
+ return None
+ current = current.get(key)
+ if current is None:
+ return None
+ return current
</code_context>
<issue_to_address>
**suggestion:** Broaden `get_nested_value` input type and consider clearer semantics for `None` values.
Right now this helper only accepts `record: dict`, but many SDK records are `Mapping[str, Any]`. Widening the type to `t.Mapping[str, t.Any]` (or `MutableMapping`) would make it usable in more places.
Also, the function returns `None` both when the path is missing and when the leaf value is explicitly `None`, which differs from the docstring’s “None if any level is missing”. If callers need to distinguish these cases, consider a dedicated sentinel return value or an optional flag to control the behavior.
Suggested implementation:
```python
def get_nested_value(
record: t.Mapping[str, t.Any],
dotted_key: str,
*,
missing: object | None = None,
) -> t.Any | None: # noqa: ANN401
"""Retrieve a value from a nested mapping using a dotted key path.
Args:
record: The record mapping to traverse.
dotted_key: A dot-separated key path (e.g. ``"attributes.updated"``).
missing: Sentinel value to return when any level in the path is missing.
Defaults to ``None`` for backward-compatible behavior.
Returns:
The value at the nested path if it exists. If any level in the path is
missing, returns ``missing`` instead. This allows callers to distinguish
between an explicitly stored ``None`` value and a missing path by passing
a dedicated sentinel (e.g. ``missing=object()``).
Examples:
>>> get_nested_value({"a": {"b": 1}}, "a.b")
1
>>> get_nested_value({"a": 1}, "a.b")
None
>>> sentinel = object()
>>> get_nested_value({"a": 1}, "a.b", missing=sentinel) is sentinel
True
>>> get_nested_value({"a": {"b": None}}, "a.b", missing=sentinel) is None
True
"""
keys = dotted_key.split(".")
current: t.Any = record
for key in keys:
# Stop if the current value is not a mapping.
if not isinstance(current, Mapping):
return missing
# Distinguish between a missing key and a present key with value None.
if key not in current:
return missing
current = current[key]
return current
```
1. Ensure `Mapping` is imported from `collections.abc` at the top of the module, e.g.:
`from collections.abc import Mapping`.
2. If the project uses strict type checking, you may want to refine the return type to
reflect that it can be either the stored value or the sentinel type (e.g. using
`t.Any` for simplicity or a type variable constrained to `object` for `missing`).
3. Call sites that need to distinguish between missing paths and explicit `None` values
should be updated to pass a dedicated sentinel object to the `missing` parameter and
compare against it (e.g. `sentinel = object(); value = get_nested_value(..., missing=sentinel)`).
</issue_to_address>
### Comment 2
<location path="tests/core/test_streams.py" line_range="888-891" />
<code_context>
+}
+
+
+def test_flat_replication_key_unchanged() -> None:
+ """Flat keys must behave exactly as before — no regression."""
+ result = get_nested_value(GITHUB_RECORD, "updated_at")
+ assert result == "2024-06-15T14:22:10Z"
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Extend tests to cover stream state increment behavior for flat vs nested replication keys
These tests cover `get_nested_value`, but they don’t verify `_increment_stream_state`, which flattens nested values before calling the state manager. Please add tests that:
1. For a stream with a flat `replication_key` (e.g. `updated_at`), assert `_increment_stream_state` passes the original record unchanged to `state_manager.increment_state`.
2. For a stream with a nested `replication_key` (e.g. `attributes.updated`), assert `_increment_stream_state` passes a flattened record like `{"attributes.updated": <timestamp>}` to `state_manager.increment_state`.
A fake/stub `state_manager` or inspection of the resulting state is sufficient to validate this integration without tightly coupling to the state implementation.
Suggested implementation:
```python
CONFIG_START_DATE = "2021-01-01"
class _FakeStateManager:
"""Simple state manager stub that records increment_state calls."""
def __init__(self) -> None:
self.calls: t.List[t.Tuple[dict, dict]] = []
def increment_state(
self,
stream_name: str,
partition_state: dict,
replication_key_value: dict,
) -> None:
# Store the arguments so tests can assert on them.
self.calls.append(
(
{
"stream_name": stream_name,
"partition_state": partition_state,
},
replication_key_value,
)
)
def test_increment_stream_state_flat_replication_key(
stream_with_flat_replication_key: "Stream",
) -> None:
"""For flat replication keys, the original record must be passed unchanged."""
state_manager = _FakeStateManager()
record = GITHUB_RECORD # uses flat `updated_at` key
stream_with_flat_replication_key._increment_stream_state(
state_manager=state_manager,
context=None,
record=record,
)
assert len(state_manager.calls) == 1
call_metadata, replication_key_value = state_manager.calls[0]
# The stream metadata is not the focus here; we only care about the record
# passed to increment_state for the replication key.
assert replication_key_value == {"updated_at": record["updated_at"]}
def test_increment_stream_state_nested_replication_key(
stream_with_nested_replication_key: "Stream",
) -> None:
"""For nested replication keys, a flattened record must be passed."""
state_manager = _FakeStateManager()
record = DEEP_NESTED_RECORD
stream_with_nested_replication_key._increment_stream_state(
state_manager=state_manager,
context=None,
record=record,
)
assert len(state_manager.calls) == 1
call_metadata, replication_key_value = state_manager.calls[0]
# Nested value must be flattened before passing to the state manager.
expected_value = get_nested_value(record, "meta.audit.last_modified")
assert replication_key_value == {"meta.audit.last_modified": expected_value}
```
1. Ensure `t` (typing) is imported at the top of `tests/core/test_streams.py` (e.g. `import typing as t`) if not already present.
2. Provide fixtures `stream_with_flat_replication_key` and `stream_with_nested_replication_key` in this test module or a shared `conftest.py`. They should:
- Construct concrete `Stream` instances (or test doubles) whose `replication_key` is `"updated_at"` for the flat case and `"meta.audit.last_modified"` (or your actual nested key string) for the nested case.
- Be wired so that `_increment_stream_state` uses the provided `state_manager`’s `increment_state` method. Adapt the argument names and count in `_FakeStateManager.increment_state` to exactly match what `_increment_stream_state` calls in your codebase.
3. If `_increment_stream_state` has a different signature (e.g. additional parameters like `partition` or `bookmark`), update the test calls accordingly and capture those extra arguments in `_FakeStateManager.increment_state` as needed, keeping the assertions focused on the shape of `replication_key_value`.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+888
to
+891
| def test_flat_replication_key_unchanged() -> None: | ||
| """Flat keys must behave exactly as before — no regression.""" | ||
| result = get_nested_value(GITHUB_RECORD, "updated_at") | ||
| assert result == "2024-06-15T14:22:10Z" |
Contributor
There was a problem hiding this comment.
suggestion (testing): Extend tests to cover stream state increment behavior for flat vs nested replication keys
These tests cover get_nested_value, but they don’t verify _increment_stream_state, which flattens nested values before calling the state manager. Please add tests that:
- For a stream with a flat
replication_key(e.g.updated_at), assert_increment_stream_statepasses the original record unchanged tostate_manager.increment_state. - For a stream with a nested
replication_key(e.g.attributes.updated), assert_increment_stream_statepasses a flattened record like{"attributes.updated": <timestamp>}tostate_manager.increment_state.
A fake/stub state_manager or inspection of the resulting state is sufficient to validate this integration without tightly coupling to the state implementation.
Suggested implementation:
CONFIG_START_DATE = "2021-01-01"
class _FakeStateManager:
"""Simple state manager stub that records increment_state calls."""
def __init__(self) -> None:
self.calls: t.List[t.Tuple[dict, dict]] = []
def increment_state(
self,
stream_name: str,
partition_state: dict,
replication_key_value: dict,
) -> None:
# Store the arguments so tests can assert on them.
self.calls.append(
(
{
"stream_name": stream_name,
"partition_state": partition_state,
},
replication_key_value,
)
)
def test_increment_stream_state_flat_replication_key(
stream_with_flat_replication_key: "Stream",
) -> None:
"""For flat replication keys, the original record must be passed unchanged."""
state_manager = _FakeStateManager()
record = GITHUB_RECORD # uses flat `updated_at` key
stream_with_flat_replication_key._increment_stream_state(
state_manager=state_manager,
context=None,
record=record,
)
assert len(state_manager.calls) == 1
call_metadata, replication_key_value = state_manager.calls[0]
# The stream metadata is not the focus here; we only care about the record
# passed to increment_state for the replication key.
assert replication_key_value == {"updated_at": record["updated_at"]}
def test_increment_stream_state_nested_replication_key(
stream_with_nested_replication_key: "Stream",
) -> None:
"""For nested replication keys, a flattened record must be passed."""
state_manager = _FakeStateManager()
record = DEEP_NESTED_RECORD
stream_with_nested_replication_key._increment_stream_state(
state_manager=state_manager,
context=None,
record=record,
)
assert len(state_manager.calls) == 1
call_metadata, replication_key_value = state_manager.calls[0]
# Nested value must be flattened before passing to the state manager.
expected_value = get_nested_value(record, "meta.audit.last_modified")
assert replication_key_value == {"meta.audit.last_modified": expected_value}- Ensure
t(typing) is imported at the top oftests/core/test_streams.py(e.g.import typing as t) if not already present. - Provide fixtures
stream_with_flat_replication_keyandstream_with_nested_replication_keyin this test module or a sharedconftest.py. They should:- Construct concrete
Streaminstances (or test doubles) whosereplication_keyis"updated_at"for the flat case and"meta.audit.last_modified"(or your actual nested key string) for the nested case. - Be wired so that
_increment_stream_stateuses the providedstate_manager’sincrement_statemethod. Adapt the argument names and count in_FakeStateManager.increment_stateto exactly match what_increment_stream_statecalls in your codebase.
- Construct concrete
- If
_increment_stream_statehas a different signature (e.g. additional parameters likepartitionorbookmark), update the test calls accordingly and capture those extra arguments in_FakeStateManager.increment_stateas needed, keeping the assertions focused on the shape ofreplication_key_value.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #3690 +/- ##
==========================================
+ Coverage 94.21% 94.23% +0.02%
==========================================
Files 73 73
Lines 6203 6227 +24
Branches 761 768 +7
==========================================
+ Hits 5844 5868 +24
Misses 267 267
Partials 92 92
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Documentation build overview
|
…d fix context parameter name
daria-hrabar
changed the title
Author
|
@edgarrmondragon I've addressed the Sourcery bot feedback, improved test coverage, and updated the PR title to follow the semantic PR conventions. Would appreciate your review and feedback when you have a chance. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
Adds support for nested replication keys in the Singer SDK by introducing a
get_nested_value()helper insinger_sdk/helpers/_util.pythat traverses a record dictionary level by level using a dotted key path (e.g.,"attributes.updated"). The_increment_stream_state()method andis_timestamp_replication_keyproperty insinger_sdk/streams/core.pyare updated to use this traversal instead of a flat dictionary lookup, enabling tap developers to point replication keys at timestamp fields nested inside sub-objects rather than only at top-level fields.Why was this PR needed?
Many real-world APIs — including HubSpot, Salesforce, and GitHub — return timestamps nested inside sub-objects. Before this fix, setting
replication_key = "attributes.updated"caused the SDK to perform a flat lookup (record.get("attributes.updated")), which returnedNonebecause no top-level key with that literal name exists. As a result, the state was never updated, and the tap silently re-synced all records on every run.Investigation traced the root cause to two locations in
singer_sdk/streams/core.py:_increment_stream_state()passed the raw record directly toincrement_state(), which performed a flat key lookup returningNonefor dotted pathsis_timestamp_replication_keycalledschema.get("properties", {}).get(replication_key)which only checked the schema's top level, incorrectly raisingInvalidReplicationKeyExceptioneven when the field existed at a deeper levelThe fix keeps
singer_sdk/streams/_state.pyunchanged by flattening the extracted value into a flat record before passing it to the state manager, limiting the scope of the change and reducing regression risk.What are the relevant issue numbers?
Closes #1198
Does this PR meet the acceptance criteria?
Summary by Sourcery
Support incremental replication using dotted-path replication keys that point to nested timestamp fields, updating schema validation, state management, and tests to reliably handle nested structures.
New Features:
Bug Fixes:
Enhancements:
Tests: