a
adam_roderick
12/02/2021, 3:00 PM.
adam_roderick
12/02/2021, 3:07 PMThe situation is that led to this is: I have an API that cannot filter, but it can sort. So the tap requests that the API sorts descending. The strategy is to short circuit and end the processing when the replication key value matches or is lower than the stored state. In this case,
is_sortedprogress_trackingThis is used to track the max value seen for the replication_key during the current sync.adam_roderick
12/02/2021, 3:17 PM.
adam_roderick
12/02/2021, 3:17 PMBut for the descending sort order being the only option the API provides, any suggestions?
t
taylor
12/02/2021, 4:39 PMWas the original context of this thread deleted?
a
aaronsteers
12/02/2021, 4:49 PM@adam_roderick - This should be possible - but it is a different paradigm than the main use case.
a
adam_roderick
12/03/2021, 3:31 PM@taylor I conflated two different questions. One was a python syntax issue that didn't add anything, so I deleted it
adam_roderick
12/03/2021, 3:32 PM@aaronsteers I'm looking into overriding the logic that applies the maximum value to the replication key value. I don't have my mind wrapped around my approach just yet but will respond here when I do. Any guidance you might have in the meantime is definitely welcome
a
aaronsteers
12/03/2021, 6:07 PM@adam_roderick - After thinking over this more, I have a hunch that perhaps leaving is_sorted=false might just meet your requirements as long as you have a pagination strategy that still loops through the records in the descending manner that respects your desired query pattern. When is_sorted=false, this forces the incremental key tracking to be non-resumable "progress markers" until the successful end of the stream.
aaronsteers
12/03/2021, 6:12 PMIf you needed resumability after interruption, this would be much harder, since you've then got a descending tracker for each run but an ascending tracker for the load jobs overall. In theory it would be possible, but that resumability is what creates a lot more complexity. If you can tolerate that failed syncs are not resumable and only a successful sync will save the state bookmark, I think that makes this much easier to design for.
a
adam_roderick
12/03/2021, 7:06 PMyes that is tolerable
adam_roderick
12/03/2021, 7:06 PMThe remaining piece is to stop the pagination when the meets a certain timestamp from the past
adam_roderick
12/03/2021, 7:06 PMDo you have a suggestion for best place to override to achieve that?
a
aaronsteers
12/03/2021, 7:29 PMI think simplest way may be to compare the latest pagination pointer (descending) with the stop point provided from get_starting_timestamp(). And rather than calling that method on each pagination loop, you can optionally store both values in a
dict Copy code
next_page_token = { "stop_point": self.get_starting_timestamp(), "current_page": 1 }Noneaaronsteers
12/03/2021, 7:30 PMIf the final page also includes records you don't want to emit, then you may also need to use post_process() or parse_response() to filter those out based on the same comparison with the starting timestamp lower bound.
aaronsteers
12/03/2021, 7:30 PMDoes this help?
a
adam_roderick
12/03/2021, 8:46 PMYes! Some of those functions I hadn't seen before so this definitely points me in the right direction
adam_roderick
12/03/2021, 8:46 PMThank you
a
aaronsteers
12/03/2021, 11:43 PMFantastic! Good luck and keep us posted. 😄
Also - have a good weekend 🌅
a
adam_roderick
12/04/2021, 2:45 PMThink I'm getting close here, but getting a circular reference error that I can't pinpoint
adam_roderick
12/04/2021, 2:50 PMFrom the output below, you can see the last couple of iterations and then the error. Each page returns 2 records, then after page 12, I return
Noneget_next_page_tokenadam_roderick
12/04/2021, 2:50 PMAny ideas why I might be seeing a circular reference?
adam_roderick
12/04/2021, 2:51 PMI do have the branch available at https://github.com/Datateer/tap-krow/tree/pagination
adam_roderick
12/06/2021, 12:46 PMIf I comment out the line
previous_token = {"stop_point": self.get_starting_timestamp(self.stream_state), "current_page": 1}get_starting_timestampadam_roderick
12/09/2021, 11:47 AMThere is definitely a circular reference in the state object used to print the final state message. I can extend this indefinitely with nested context/partitions:
Copy code
print(
o["value"]["bookmarks"]["organizations"]["partitions"][0]["context"]["partitions"][0]["context"]["partitions"][0]['context']['partitions']
)adam_roderick
12/09/2021, 11:52 AMIf I make a call to
self.get_starting_timestamp(self.get_context_state(None)) Copy code
{
"type": "STATE",
"value": {
"bookmarks": {
"organizations": {
"replication_key": "updated_at",
"replication_key_value": "2020-11-30T21:44:28.839Z"
}
}
}
}adam_roderick
12/09/2021, 12:00 PMCalling
self.get_starting_timestamp(None)