Skip to content

Commit

Permalink
fix: Safely compare UUID replication keys with state bookmarks (#2756)
Browse files Browse the repository at this point in the history
  • Loading branch information
edgarrmondragon authored Nov 11, 2024
1 parent 3ebd4dd commit e264392
Show file tree
Hide file tree
Showing 3 changed files with 26 additions and 0 deletions.
4 changes: 4 additions & 0 deletions singer_sdk/helpers/_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,10 @@ def increment_state(
extra={"replication_key": replication_key},
)
progress_dict = stream_or_partition_state[PROGRESS_MARKERS]
# TODO: Instead of forcing all values to be JSON-compatible strings and hope
# we catch all cases, we should allow the stream to define how to
# the values from the state and the record should be pre-processed.
# https://github.com/meltano/sdk/issues/2753
old_rk_value = to_json_compatible(progress_dict.get("replication_key_value"))
new_rk_value = to_json_compatible(latest_record[replication_key])

Expand Down
3 changes: 3 additions & 0 deletions singer_sdk/helpers/_typing.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import datetime
import logging
import typing as t
import uuid
from enum import Enum
from functools import lru_cache

Expand Down Expand Up @@ -42,6 +43,8 @@ def to_json_compatible(val: t.Any) -> t.Any: # noqa: ANN401
if isinstance(val, (datetime.datetime,)):
# Make naive datetimes UTC
return (val.replace(tzinfo=UTC) if val.tzinfo is None else val).isoformat("T")
if isinstance(val, (uuid.UUID,)):
return str(val)
return val


Expand Down
19 changes: 19 additions & 0 deletions tests/core/test_state_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import logging
import uuid

import pytest

Expand Down Expand Up @@ -155,3 +156,21 @@ def test_null_replication_value(caplog):
), "State should not be updated."
assert caplog.records[0].levelname == "WARNING"
assert "is null" in caplog.records[0].message


def test_uuidv7_replication_value():
stream_state = {
"replication_key": "id",
"replication_key_value": "01931c63-b14e-7ff3-8621-e577ed392dc8",
}
new_string_val = "01931c63-b14e-7ff3-8621-e578edbca9a3"

_state.increment_state(
stream_state,
latest_record={"id": uuid.UUID(new_string_val)},
replication_key="id",
is_sorted=True,
check_sorted=True,
)

assert stream_state["replication_key_value"] == new_string_val

0 comments on commit e264392

Please sign in to comment.