Hey everyone Is it possible to increment state even if no re Meltano #singer-tap-development

Hey everyone! Is it possible to increment state e...

Conner Panarella

09/09/2024, 9:01 PM

Hey everyone! Is it possible to increment state even if no records were emitted using the SDK? Example scenario: I have an API that allows me to query for posts in the past 7 days. A record is grabbed on 2024-09-01, the state reflects this. There are no posts for 7+ days, so no new records are grabbed during this time. If, on 2024-09-09 I initiate a run the API will throw an error since I am pulling from 2024-09-01 which is out of bounds. I'd like to increment the state to the last run, whether or not there were any records.

visch

09/09/2024, 11:51 PM

You should be able to emit a state message at the end of your run regardless. I don't remember offhand but I think there's a clean_up function you can call, in that increment the state to the date and emit the state message. Personally what I'd do instead is figure out what is "out of bounds" for the API call (if it's 7 days back from today then I'd just calculate that to be my date, and if the state is older then that I'd just bump the date up to 7 days from that state message)

Edgar Ramírez (Arch.dev)

09/09/2024, 11:58 PM

There's isn't a documented or "official" way to do this atm and I've not implemented something like it, so I'm curious about what options you've tried so far to see if we can find a workaround. Also: issues and PRs, always welcome! I presume for your use case, in a normal execution the state is updated based on a record's field, but you want to update it regardless when no records are returned?

Conner Panarella

09/10/2024, 2:01 AM

Thanks for the tips! I totally agree with that approach, but unfortunately it won’t work in my case. What I’m actually trying to do is implement change tracking for a MSSQL tap. It’s not a typical method of incremental replication, so I’ve been trying to draw inspiration from the official Postgres tap. I’ll look into the clean up method to see if I can add it there!

Conner Panarella

09/10/2024, 2:31 PM

I was able to just override _sync_records:

Copy code

def _sync_records(  # noqa: C901
        self,
        context: types.Context | None = None,
        *,
        write_messages: bool = True,
) -> t.Generator[dict, t.Any, t.Any]:

    yield from super()._sync_records(context=context, write_messages=write_messages)

    if write_messages and self.selected:
        state_dict = self.get_context_state(context)
        increment_state(
            state_dict,
            replication_key=self.replication_key,
            latest_record={self.replication_key: self.change_tracking_current_version},
            is_sorted=self.is_sorted,
            check_sorted=self.check_sorted,
        )

Edgar Ramírez (Arch.dev)

09/10/2024, 4:43 PM

https://github.com/meltano/sdk/pull/2620 is a PR that does something similar: it emits a final state message before terminating a tap. Perhaps the clean up method is the way to go for this sort of thing in general, though.

Open in Slack

Previous Next