airbyte_cdk.legacy.sources.declarative.incremental

20class CursorFactory:
21    def __init__(self, create_function: Callable[[], DeclarativeCursor]):
22        self._create_function = create_function
23
24    def create(self) -> DeclarativeCursor:
25        return self._create_function()

 28@dataclass
 29class DatetimeBasedCursor(DeclarativeCursor):
 30    """
 31    Slices the stream over a datetime range and create a state with format {<cursor_field>: <datetime> }
 32
 33    Given a start time, end time, a step function, and an optional lookback window,
 34    the stream slicer will partition the date range from start time - lookback window to end time.
 35
 36    The step function is defined as a string of the form ISO8601 duration
 37
 38    The timestamp format accepts the same format codes as datetime.strfptime, which are
 39    all the format codes required by the 1989 C standard.
 40    Full list of accepted format codes: https://man7.org/linux/man-pages/man3/strftime.3.html
 41
 42    Attributes:
 43        start_datetime (Union[MinMaxDatetime, str]): the datetime that determines the earliest record that should be synced
 44        end_datetime (Optional[Union[MinMaxDatetime, str]]): the datetime that determines the last record that should be synced
 45        cursor_field (Union[InterpolatedString, str]): record's cursor field
 46        datetime_format (str): format of the datetime
 47        step (Optional[str]): size of the timewindow (ISO8601 duration)
 48        cursor_granularity (Optional[str]): smallest increment the datetime_format has (ISO 8601 duration) that will be used to ensure that the start of a slice does not overlap with the end of the previous one
 49        config (Config): connection config
 50        start_time_option (Optional[RequestOption]): request option for start time
 51        end_time_option (Optional[RequestOption]): request option for end time
 52        partition_field_start (Optional[str]): partition start time field
 53        partition_field_end (Optional[str]): stream slice end time field
 54        lookback_window (Optional[InterpolatedString]): how many days before start_datetime to read data for (ISO8601 duration)
 55    """
 56
 57    start_datetime: Union[MinMaxDatetime, str]
 58    cursor_field: Union[InterpolatedString, str]
 59    datetime_format: str
 60    config: Config
 61    parameters: InitVar[Mapping[str, Any]]
 62    _highest_observed_cursor_field_value: Optional[str] = field(
 63        repr=False, default=None
 64    )  # tracks the latest observed datetime, which may not be safe to emit in the case of out-of-order records
 65    _cursor: Optional[str] = field(
 66        repr=False, default=None
 67    )  # tracks the latest observed datetime that is appropriate to emit as stream state
 68    end_datetime: Optional[Union[MinMaxDatetime, str]] = None
 69    step: Optional[Union[InterpolatedString, str]] = None
 70    cursor_granularity: Optional[str] = None
 71    start_time_option: Optional[RequestOption] = None
 72    end_time_option: Optional[RequestOption] = None
 73    partition_field_start: Optional[str] = None
 74    partition_field_end: Optional[str] = None
 75    lookback_window: Optional[Union[InterpolatedString, str]] = None
 76    message_repository: Optional[MessageRepository] = None
 77    is_compare_strictly: Optional[bool] = False
 78    cursor_datetime_formats: List[str] = field(default_factory=lambda: [])
 79
 80    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 81        if (self.step and not self.cursor_granularity) or (
 82            not self.step and self.cursor_granularity
 83        ):
 84            raise ValueError(
 85                f"If step is defined, cursor_granularity should be as well and vice-versa. "
 86                f"Right now, step is `{self.step}` and cursor_granularity is `{self.cursor_granularity}`"
 87            )
 88        self._start_datetime = MinMaxDatetime.create(self.start_datetime, parameters)
 89        self._end_datetime = (
 90            None if not self.end_datetime else MinMaxDatetime.create(self.end_datetime, parameters)
 91        )
 92
 93        self._timezone = datetime.timezone.utc
 94        self._interpolation = JinjaInterpolation()
 95
 96        self._step = (
 97            self._parse_timedelta(
 98                InterpolatedString.create(self.step, parameters=parameters).eval(self.config)
 99            )
100            if self.step
101            else datetime.timedelta.max
102        )
103        self._cursor_granularity = self._parse_timedelta(self.cursor_granularity)
104        self.cursor_field = InterpolatedString.create(self.cursor_field, parameters=parameters)
105        self._lookback_window = (
106            InterpolatedString.create(self.lookback_window, parameters=parameters)
107            if self.lookback_window
108            else None
109        )
110        self._partition_field_start = InterpolatedString.create(
111            self.partition_field_start or "start_time", parameters=parameters
112        )
113        self._partition_field_end = InterpolatedString.create(
114            self.partition_field_end or "end_time", parameters=parameters
115        )
116        self._parser = DatetimeParser()
117
118        # If datetime format is not specified then start/end datetime should inherit it from the stream slicer
119        if not self._start_datetime.datetime_format:
120            self._start_datetime.datetime_format = self.datetime_format
121        if self._end_datetime and not self._end_datetime.datetime_format:
122            self._end_datetime.datetime_format = self.datetime_format
123
124        if not self.cursor_datetime_formats:
125            self.cursor_datetime_formats = [self.datetime_format]
126
127        _validate_component_request_option_paths(
128            self.config, self.start_time_option, self.end_time_option
129        )
130
131    def get_stream_state(self) -> StreamState:
132        return {self.cursor_field.eval(self.config): self._cursor} if self._cursor else {}  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
133
134    def set_initial_state(self, stream_state: StreamState) -> None:
135        """
136        Cursors are not initialized with their state. As state is needed in order to function properly, this method should be called
137        before calling anything else
138
139        :param stream_state: The state of the stream as returned by get_stream_state
140        """
141        self._cursor = (
142            stream_state.get(self.cursor_field.eval(self.config)) if stream_state else None  # type: ignore [union-attr]
143        )
144
145    def observe(self, stream_slice: StreamSlice, record: Record) -> None:
146        """
147        Register a record with the cursor; the cursor instance can then use it to manage the state of the in-progress stream read.
148
149        :param stream_slice: The current slice, which may or may not contain the most recently observed record
150        :param record: the most recently-read record, which the cursor can use to update the stream state. Outwardly-visible changes to the
151          stream state may need to be deferred depending on whether the source reliably orders records by the cursor field.
152        """
153        record_cursor_value = record.get(self.cursor_field.eval(self.config))  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
154        # if the current record has no cursor value, we cannot meaningfully update the state based on it, so there is nothing more to do
155        if not record_cursor_value:
156            return
157
158        start_field = self._partition_field_start.eval(self.config)
159        end_field = self._partition_field_end.eval(self.config)
160        is_highest_observed_cursor_value = (
161            not self._highest_observed_cursor_field_value
162            or self.parse_date(record_cursor_value)
163            > self.parse_date(self._highest_observed_cursor_field_value)
164        )
165        if (
166            self._is_within_daterange_boundaries(
167                record,
168                stream_slice.get(start_field),  # type: ignore [arg-type]
169                stream_slice.get(end_field),  # type: ignore [arg-type]
170            )
171            and is_highest_observed_cursor_value
172        ):
173            self._highest_observed_cursor_field_value = record_cursor_value
174
175    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
176        if stream_slice.partition:
177            raise ValueError(
178                f"Stream slice {stream_slice} should not have a partition. Got {stream_slice.partition}."
179            )
180        cursor_value_str_by_cursor_value_datetime = dict(
181            map(
182                # we need to ensure the cursor value is preserved as is in the state else the CATs might complain of something like
183                # 2023-01-04T17:30:19.000Z' <= '2023-01-04T17:30:19.000000Z'
184                lambda datetime_str: (self.parse_date(datetime_str), datetime_str),  # type: ignore # because of the filter on the next line, this will only be called with a str
185                filter(
186                    lambda item: item, [self._cursor, self._highest_observed_cursor_field_value]
187                ),
188            )
189        )
190        self._cursor = (
191            cursor_value_str_by_cursor_value_datetime[
192                max(cursor_value_str_by_cursor_value_datetime.keys())
193            ]
194            if cursor_value_str_by_cursor_value_datetime
195            else None
196        )
197
198    def stream_slices(self) -> Iterable[StreamSlice]:
199        """
200        Partition the daterange into slices of size = step.
201
202        The start of the window is the minimum datetime between start_datetime - lookback_window and the stream_state's datetime
203        The end of the window is the minimum datetime between the start of the window and end_datetime.
204
205        :return:
206        """
207        end_datetime = self.select_best_end_datetime()
208        start_datetime = self._calculate_earliest_possible_value(self.select_best_end_datetime())
209        return self._partition_daterange(start_datetime, end_datetime, self._step)
210
211    def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
212        # Datetime based cursors operate over slices made up of datetime ranges. Stream state is based on the progress
213        # through each slice and does not belong to a specific slice. We just return stream state as it is.
214        return self.get_stream_state()
215
216    def _calculate_earliest_possible_value(
217        self, end_datetime: datetime.datetime
218    ) -> datetime.datetime:
219        lookback_delta = self._parse_timedelta(
220            self._lookback_window.eval(self.config) if self._lookback_window else "P0D"
221        )
222        earliest_possible_start_datetime = min(
223            self._start_datetime.get_datetime(self.config), end_datetime
224        )
225        try:
226            cursor_datetime = (
227                self._calculate_cursor_datetime_from_state(self.get_stream_state()) - lookback_delta
228            )
229        except OverflowError:
230            # cursor_datetime defers to the minimum date if it does not exist in the state. Trying to subtract
231            # a timedelta from the minimum datetime results in an OverflowError
232            cursor_datetime = self._calculate_cursor_datetime_from_state(self.get_stream_state())
233        return max(earliest_possible_start_datetime, cursor_datetime)
234
235    def select_best_end_datetime(self) -> datetime.datetime:
236        """
237        Returns the optimal end datetime.
238        This method compares the current datetime with a pre-configured end datetime
239        and returns the earlier of the two. If no pre-configured end datetime is set,
240        the current datetime is returned.
241
242        :return datetime.datetime: The best end datetime, which is either the current datetime or the pre-configured end datetime, whichever is earlier.
243        """
244        now = datetime.datetime.now(tz=self._timezone)
245        if not self._end_datetime:
246            return now
247        return min(self._end_datetime.get_datetime(self.config), now)
248
249    def _calculate_cursor_datetime_from_state(
250        self, stream_state: Mapping[str, Any]
251    ) -> datetime.datetime:
252        if self.cursor_field.eval(self.config, stream_state=stream_state) in stream_state:  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
253            return self.parse_date(stream_state[self.cursor_field.eval(self.config)])  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
254        return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
255
256    def _format_datetime(self, dt: datetime.datetime) -> str:
257        return self._parser.format(dt, self.datetime_format)
258
259    def _partition_daterange(
260        self,
261        start: datetime.datetime,
262        end: datetime.datetime,
263        step: Union[datetime.timedelta, Duration],
264    ) -> List[StreamSlice]:
265        start_field = self._partition_field_start.eval(self.config)
266        end_field = self._partition_field_end.eval(self.config)
267        dates = []
268
269        while self._is_within_date_range(start, end):
270            next_start = self._evaluate_next_start_date_safely(start, step)
271            end_date = self._get_date(next_start - self._cursor_granularity, end, min)
272            dates.append(
273                StreamSlice(
274                    partition={},
275                    cursor_slice={
276                        start_field: self._format_datetime(start),
277                        end_field: self._format_datetime(end_date),
278                    },
279                )
280            )
281            start = next_start
282        return dates
283
284    def _is_within_date_range(self, start: datetime.datetime, end: datetime.datetime) -> bool:
285        if self.is_compare_strictly:
286            return start < end
287        return start <= end
288
289    def _evaluate_next_start_date_safely(
290        self, start: datetime.datetime, step: datetime.timedelta
291    ) -> datetime.datetime:
292        """
293        Given that we set the default step at datetime.timedelta.max, we will generate an OverflowError when evaluating the next start_date
294        This method assumes that users would never enter a step that would generate an overflow. Given that would be the case, the code
295        would have broken anyway.
296        """
297        try:
298            return start + step
299        except OverflowError:
300            return datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
301
302    def _get_date(
303        self,
304        cursor_value: datetime.datetime,
305        default_date: datetime.datetime,
306        comparator: Callable[[datetime.datetime, datetime.datetime], datetime.datetime],
307    ) -> datetime.datetime:
308        cursor_date = cursor_value or default_date
309        return comparator(cursor_date, default_date)
310
311    def parse_date(self, date: str) -> datetime.datetime:
312        for datetime_format in self.cursor_datetime_formats + [self.datetime_format]:
313            try:
314                return self._parser.parse(date, datetime_format)
315            except ValueError:
316                pass
317        raise ValueError(f"No format in {self.cursor_datetime_formats} matching {date}")
318
319    @classmethod
320    def _parse_timedelta(cls, time_str: Optional[str]) -> Union[datetime.timedelta, Duration]:
321        """
322        :return Parses an ISO 8601 durations into datetime.timedelta or Duration objects.
323        """
324        if not time_str:
325            return datetime.timedelta(0)
326        return parse_duration(time_str)
327
328    def get_request_params(
329        self,
330        *,
331        stream_state: Optional[StreamState] = None,
332        stream_slice: Optional[StreamSlice] = None,
333        next_page_token: Optional[Mapping[str, Any]] = None,
334    ) -> Mapping[str, Any]:
335        return self._get_request_options(RequestOptionType.request_parameter, stream_slice)
336
337    def get_request_headers(
338        self,
339        *,
340        stream_state: Optional[StreamState] = None,
341        stream_slice: Optional[StreamSlice] = None,
342        next_page_token: Optional[Mapping[str, Any]] = None,
343    ) -> Mapping[str, Any]:
344        return self._get_request_options(RequestOptionType.header, stream_slice)
345
346    def get_request_body_data(
347        self,
348        *,
349        stream_state: Optional[StreamState] = None,
350        stream_slice: Optional[StreamSlice] = None,
351        next_page_token: Optional[Mapping[str, Any]] = None,
352    ) -> Mapping[str, Any]:
353        return self._get_request_options(RequestOptionType.body_data, stream_slice)
354
355    def get_request_body_json(
356        self,
357        *,
358        stream_state: Optional[StreamState] = None,
359        stream_slice: Optional[StreamSlice] = None,
360        next_page_token: Optional[Mapping[str, Any]] = None,
361    ) -> Mapping[str, Any]:
362        return self._get_request_options(RequestOptionType.body_json, stream_slice)
363
364    def request_kwargs(self) -> Mapping[str, Any]:
365        # Never update kwargs
366        return {}
367
368    def _get_request_options(
369        self, option_type: RequestOptionType, stream_slice: Optional[StreamSlice]
370    ) -> Mapping[str, Any]:
371        options: MutableMapping[str, Any] = {}
372        if not stream_slice:
373            return options
374
375        if self.start_time_option and self.start_time_option.inject_into == option_type:
376            start_time_value = stream_slice.get(self._partition_field_start.eval(self.config))
377            self.start_time_option.inject_into_request(options, start_time_value, self.config)
378
379        if self.end_time_option and self.end_time_option.inject_into == option_type:
380            end_time_value = stream_slice.get(self._partition_field_end.eval(self.config))
381            self.end_time_option.inject_into_request(options, end_time_value, self.config)
382
383        return options
384
385    def should_be_synced(self, record: Record) -> bool:
386        cursor_field = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
387        record_cursor_value = record.get(cursor_field)
388        if not record_cursor_value:
389            self._send_log(
390                Level.WARN,
391                f"Could not find cursor field `{cursor_field}` in record. The incremental sync will assume it needs to be synced",
392            )
393            return True
394        latest_possible_cursor_value = self.select_best_end_datetime()
395        earliest_possible_cursor_value = self._calculate_earliest_possible_value(
396            latest_possible_cursor_value
397        )
398        return self._is_within_daterange_boundaries(
399            record, earliest_possible_cursor_value, latest_possible_cursor_value
400        )
401
402    def _is_within_daterange_boundaries(
403        self,
404        record: Record,
405        start_datetime_boundary: Union[datetime.datetime, str],
406        end_datetime_boundary: Union[datetime.datetime, str],
407    ) -> bool:
408        cursor_field = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
409        record_cursor_value = record.get(cursor_field)
410        if not record_cursor_value:
411            self._send_log(
412                Level.WARN,
413                f"Could not find cursor field `{cursor_field}` in record. The record will not be considered when emitting sync state",
414            )
415            return False
416        if isinstance(start_datetime_boundary, str):
417            start_datetime_boundary = self.parse_date(start_datetime_boundary)
418        if isinstance(end_datetime_boundary, str):
419            end_datetime_boundary = self.parse_date(end_datetime_boundary)
420        return (
421            start_datetime_boundary <= self.parse_date(record_cursor_value) <= end_datetime_boundary
422        )
423
424    def _send_log(self, level: Level, message: str) -> None:
425        if self.message_repository:
426            self.message_repository.emit_message(
427                AirbyteMessage(
428                    type=Type.LOG,
429                    log=AirbyteLogMessage(level=level, message=message),
430                )
431            )
432
433    def set_runtime_lookback_window(self, lookback_window_in_seconds: int) -> None:
434        """
435        Updates the lookback window based on a given number of seconds if the new duration
436        is greater than the currently configured lookback window.
437
438        :param lookback_window_in_seconds: The lookback duration in seconds to potentially update to.
439        """
440        runtime_lookback_window = duration_isoformat(timedelta(seconds=lookback_window_in_seconds))
441        config_lookback = parse_duration(
442            self._lookback_window.eval(self.config) if self._lookback_window else "P0D"
443        )
444
445        # Check if the new runtime lookback window is greater than the current config lookback
446        if parse_duration(runtime_lookback_window) > config_lookback:
447            self._lookback_window = InterpolatedString.create(
448                runtime_lookback_window, parameters={}
449            )

10class DeclarativeCursor(Cursor, StreamSlicer, ABC):
11    """
12    DeclarativeCursors are components that allow for checkpointing syncs. In addition to managing the fetching and updating of
13    state, declarative cursors also manage stream slicing and injecting slice values into outbound requests.
14    """

 74class GlobalSubstreamCursor(DeclarativeCursor):
 75    """
 76    The GlobalSubstreamCursor is designed to track the state of substreams using a single global cursor.
 77    This class is beneficial for streams with many partitions, as it allows the state to be managed globally
 78    instead of per partition, simplifying state management and reducing the size of state messages.
 79
 80    This cursor is activated by setting the `global_substream_cursor` parameter for incremental sync.
 81
 82    Warnings:
 83    - This class enforces a minimal lookback window for substream based on the duration of the previous sync to avoid losing records. This lookback ensures that any records added or updated during the sync are captured in subsequent syncs.
 84    - The global cursor is updated only at the end of the sync. If the sync ends prematurely (e.g., due to an exception), the state will not be updated.
 85    - When using the `incremental_dependency` option, the sync will progress through parent records, preventing the sync from getting infinitely stuck. However, it is crucial to understand the requirements for both the `global_substream_cursor` and `incremental_dependency` options to avoid data loss.
 86    """
 87
 88    def __init__(self, stream_cursor: DatetimeBasedCursor, partition_router: PartitionRouter):
 89        self._stream_cursor = stream_cursor
 90        self._partition_router = partition_router
 91        self._timer = Timer()
 92        self._lock = threading.Lock()
 93        self._slice_semaphore = threading.Semaphore(
 94            0
 95        )  # Start with 0, indicating no slices being tracked
 96        self._all_slices_yielded = False
 97        self._lookback_window: Optional[int] = None
 98        self._current_partition: Optional[Mapping[str, Any]] = None
 99        self._last_slice: bool = False
100        self._parent_state: Optional[Mapping[str, Any]] = None
101
102    def start_slices_generation(self) -> None:
103        self._timer.start()
104
105    def stream_slices(self) -> Iterable[StreamSlice]:
106        """
107        Generates stream slices, ensuring the last slice is properly flagged and processed.
108
109        This method creates a sequence of stream slices by iterating over partitions and cursor slices.
110        It holds onto one slice in memory to set `_all_slices_yielded` to `True` before yielding the
111        final slice. A semaphore is used to track the processing of slices, ensuring that `close_slice`
112        is called only after all slices have been processed.
113
114        We expect the following events:
115        * Yields all the slices except the last one. At this point, `close_slice` won't actually close the global slice as `self._all_slices_yielded == False`
116        * Release the semaphore one last time before setting `self._all_slices_yielded = True`. This will cause `close_slice` to know about all the slices before we indicate that all slices have been yielded so the left side of `if self._all_slices_yielded and self._slice_semaphore._value == 0` will be false if not everything is closed
117        * Setting `self._all_slices_yielded = True`. We do that before actually yielding the last slice as the caller of `stream_slices` might stop iterating at any point and hence the code after `yield` might not be executed
118        * Yield the last slice. At that point, once there are as many slices yielded as closes, the global slice will be closed too
119        """
120        slice_generator = (
121            StreamSlice(
122                partition=partition, cursor_slice=cursor_slice, extra_fields=partition.extra_fields
123            )
124            for partition in self._partition_router.stream_slices()
125            for cursor_slice in self._stream_cursor.stream_slices()
126        )
127
128        self.start_slices_generation()
129        for slice, last, state in iterate_with_last_flag_and_state(
130            slice_generator, self._partition_router.get_stream_state
131        ):
132            self._parent_state = state
133            self.register_slice(last)
134            yield slice
135        self._parent_state = self._partition_router.get_stream_state()
136
137    def generate_slices_from_partition(self, partition: StreamSlice) -> Iterable[StreamSlice]:
138        slice_generator = (
139            StreamSlice(
140                partition=partition, cursor_slice=cursor_slice, extra_fields=partition.extra_fields
141            )
142            for cursor_slice in self._stream_cursor.stream_slices()
143        )
144
145        yield from slice_generator
146
147    def register_slice(self, last: bool) -> None:
148        """
149        Tracks the processing of a stream slice.
150
151        Releases the semaphore for each slice. If it's the last slice (`last=True`),
152        sets `_all_slices_yielded` to `True` to indicate no more slices will be processed.
153
154        Args:
155            last (bool): True if the current slice is the last in the sequence.
156        """
157        self._slice_semaphore.release()
158        if last:
159            self._all_slices_yielded = True
160
161    def set_initial_state(self, stream_state: StreamState) -> None:
162        """
163        Set the initial state for the cursors.
164
165        This method initializes the state for the global cursor using the provided stream state.
166
167        Additionally, it sets the parent state for partition routers that are based on parent streams. If a partition router
168        does not have parent streams, this step will be skipped due to the default PartitionRouter implementation.
169
170        Args:
171            stream_state (StreamState): The state of the streams to be set. The format of the stream state should be:
172                {
173                    "state": {
174                        "last_updated": "2023-05-27T00:00:00Z"
175                    },
176                    "parent_state": {
177                        "parent_stream_name": {
178                            "last_updated": "2023-05-27T00:00:00Z"
179                        }
180                    },
181                    "lookback_window": 132
182                }
183        """
184        if not stream_state:
185            return
186
187        if "lookback_window" in stream_state:
188            self._lookback_window = stream_state["lookback_window"]
189            self._inject_lookback_into_stream_cursor(stream_state["lookback_window"])
190
191        if "state" in stream_state:
192            self._stream_cursor.set_initial_state(stream_state["state"])
193        elif "states" not in stream_state:
194            # We assume that `stream_state` is in the old global format
195            # Example: {"global_state_format_key": "global_state_format_value"}
196            self._stream_cursor.set_initial_state(stream_state)
197
198        # We used to set the parent state through this method but since moving the SubstreamPartitionRouter to the
199        # Concurrent CDK/AbstractStream, the state is passed at the __init__ stage and this does not need to be called.
200        # We are still keeping this line as a comment to be explicit about the past behavior.
201        # self._partition_router.set_initial_state(stream_state)
202
203    def _inject_lookback_into_stream_cursor(self, lookback_window: int) -> None:
204        """
205        Modifies the stream cursor's lookback window based on the duration of the previous sync.
206        This adjustment ensures the cursor is set to the minimal lookback window necessary for
207        avoiding missing data.
208
209        Parameters:
210            lookback_window (int): The lookback duration in seconds to be set, derived from
211                                   the previous sync.
212
213        Raises:
214            ValueError: If the cursor does not support dynamic lookback window adjustments.
215        """
216        if hasattr(self._stream_cursor, "set_runtime_lookback_window"):
217            self._stream_cursor.set_runtime_lookback_window(lookback_window)
218        else:
219            raise ValueError(
220                "The cursor class for Global Substream Cursor does not have a set_runtime_lookback_window method"
221            )
222
223    def observe(self, stream_slice: StreamSlice, record: Record) -> None:
224        self._stream_cursor.observe(
225            StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), record
226        )
227
228    def close_slice(self, stream_slice: StreamSlice, *args: Any) -> None:
229        """
230        Close the current stream slice.
231
232        This method is called when a stream slice is completed. For the global parent cursor, we close the child cursor
233        only after reading all slices. This ensures that we do not miss any child records from a later parent record
234        if the child cursor is earlier than a record from the first parent record.
235
236        Args:
237            stream_slice (StreamSlice): The stream slice to be closed.
238            *args (Any): Additional arguments.
239        """
240        with self._lock:
241            self._slice_semaphore.acquire()
242            if self._all_slices_yielded and self._slice_semaphore._value == 0:
243                self._lookback_window = self._timer.finish()
244                self._stream_cursor.close_slice(
245                    StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice), *args
246                )
247
248    def get_stream_state(self) -> StreamState:
249        state: dict[str, Any] = {"state": self._stream_cursor.get_stream_state()}
250
251        if self._parent_state:
252            state["parent_state"] = self._parent_state
253
254        if self._lookback_window is not None:
255            state["lookback_window"] = self._lookback_window
256
257        return state
258
259    def select_state(self, stream_slice: Optional[StreamSlice] = None) -> Optional[StreamState]:
260        # stream_slice is ignored as cursor is global
261        return self._stream_cursor.get_stream_state()
262
263    def get_request_params(
264        self,
265        *,
266        stream_state: Optional[StreamState] = None,
267        stream_slice: Optional[StreamSlice] = None,
268        next_page_token: Optional[Mapping[str, Any]] = None,
269    ) -> Mapping[str, Any]:
270        if stream_slice:
271            return self._partition_router.get_request_params(  # type: ignore # this always returns a mapping
272                stream_state=stream_state,
273                stream_slice=StreamSlice(partition=stream_slice.partition, cursor_slice={}),
274                next_page_token=next_page_token,
275            ) | self._stream_cursor.get_request_params(
276                stream_state=stream_state,
277                stream_slice=StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice),
278                next_page_token=next_page_token,
279            )
280        else:
281            raise ValueError("A partition needs to be provided in order to get request params")
282
283    def get_request_headers(
284        self,
285        *,
286        stream_state: Optional[StreamState] = None,
287        stream_slice: Optional[StreamSlice] = None,
288        next_page_token: Optional[Mapping[str, Any]] = None,
289    ) -> Mapping[str, Any]:
290        if stream_slice:
291            return self._partition_router.get_request_headers(  # type: ignore # this always returns a mapping
292                stream_state=stream_state,
293                stream_slice=StreamSlice(partition=stream_slice.partition, cursor_slice={}),
294                next_page_token=next_page_token,
295            ) | self._stream_cursor.get_request_headers(
296                stream_state=stream_state,
297                stream_slice=StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice),
298                next_page_token=next_page_token,
299            )
300        else:
301            raise ValueError("A partition needs to be provided in order to get request headers")
302
303    def get_request_body_data(
304        self,
305        *,
306        stream_state: Optional[StreamState] = None,
307        stream_slice: Optional[StreamSlice] = None,
308        next_page_token: Optional[Mapping[str, Any]] = None,
309    ) -> Union[Mapping[str, Any], str]:
310        if stream_slice:
311            return self._partition_router.get_request_body_data(  # type: ignore # this always returns a mapping
312                stream_state=stream_state,
313                stream_slice=StreamSlice(partition=stream_slice.partition, cursor_slice={}),
314                next_page_token=next_page_token,
315            ) | self._stream_cursor.get_request_body_data(
316                stream_state=stream_state,
317                stream_slice=StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice),
318                next_page_token=next_page_token,
319            )
320        else:
321            raise ValueError("A partition needs to be provided in order to get request body data")
322
323    def get_request_body_json(
324        self,
325        *,
326        stream_state: Optional[StreamState] = None,
327        stream_slice: Optional[StreamSlice] = None,
328        next_page_token: Optional[Mapping[str, Any]] = None,
329    ) -> Mapping[str, Any]:
330        if stream_slice:
331            return self._partition_router.get_request_body_json(  # type: ignore # this always returns a mapping
332                stream_state=stream_state,
333                stream_slice=StreamSlice(partition=stream_slice.partition, cursor_slice={}),
334                next_page_token=next_page_token,
335            ) | self._stream_cursor.get_request_body_json(
336                stream_state=stream_state,
337                stream_slice=StreamSlice(partition={}, cursor_slice=stream_slice.cursor_slice),
338                next_page_token=next_page_token,
339            )
340        else:
341            raise ValueError("A partition needs to be provided in order to get request body json")
342
343    def should_be_synced(self, record: Record) -> bool:
344        return self._stream_cursor.should_be_synced(self._convert_record_to_cursor_record(record))
345
346    @staticmethod
347    def _convert_record_to_cursor_record(record: Record) -> Record:
348        return Record(
349            data=record.data,
350            stream_name=record.stream_name,
351            associated_slice=StreamSlice(
352                partition={}, cursor_slice=record.associated_slice.cursor_slice
353            )
354            if record.associated_slice
355            else None,
356        )