airbyte_cdk.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()

35class ConcurrentCursorFactory:
36    def __init__(self, create_function: Callable[..., ConcurrentCursor]):
37        self._create_function = create_function
38
39    def create(
40        self, stream_state: Mapping[str, Any], runtime_lookback_window: Optional[timedelta]
41    ) -> ConcurrentCursor:
42        return self._create_function(
43            stream_state=stream_state, runtime_lookback_window=runtime_lookback_window
44        )

 47class ConcurrentPerPartitionCursor(Cursor):
 48    """
 49    Manages state per partition when a stream has many partitions, preventing data loss or duplication.
 50
 51    Attributes:
 52        DEFAULT_MAX_PARTITIONS_NUMBER (int): Maximum number of partitions to retain in memory (default is 10,000).
 53
 54    - **Partition Limitation Logic**
 55      Ensures the number of tracked partitions does not exceed the specified limit to prevent memory overuse. Oldest partitions are removed when the limit is reached.
 56
 57    - **Global Cursor Fallback**
 58      New partitions use global state as the initial state to progress the state for deleted or new partitions. The history data added after the initial sync will be missing.
 59
 60    CurrentPerPartitionCursor expects the state of the ConcurrentCursor to follow the format {cursor_field: cursor_value}.
 61    """
 62
 63    DEFAULT_MAX_PARTITIONS_NUMBER = 25_000
 64    SWITCH_TO_GLOBAL_LIMIT = 10_000
 65    _NO_STATE: Mapping[str, Any] = {}
 66    _NO_CURSOR_STATE: Mapping[str, Any] = {}
 67    _GLOBAL_STATE_KEY = "state"
 68    _PERPARTITION_STATE_KEY = "states"
 69    _KEY = 0
 70    _VALUE = 1
 71
 72    def __init__(
 73        self,
 74        cursor_factory: ConcurrentCursorFactory,
 75        partition_router: PartitionRouter,
 76        stream_name: str,
 77        stream_namespace: Optional[str],
 78        stream_state: Any,
 79        message_repository: MessageRepository,
 80        connector_state_manager: ConnectorStateManager,
 81        connector_state_converter: AbstractStreamStateConverter,
 82        cursor_field: CursorField,
 83    ) -> None:
 84        self._global_cursor: Optional[StreamState] = {}
 85        self._stream_name = stream_name
 86        self._stream_namespace = stream_namespace
 87        self._message_repository = message_repository
 88        self._connector_state_manager = connector_state_manager
 89        self._connector_state_converter = connector_state_converter
 90        self._cursor_field = cursor_field
 91
 92        self._cursor_factory = cursor_factory
 93        self._partition_router = partition_router
 94
 95        # The dict is ordered to ensure that once the maximum number of partitions is reached,
 96        # the oldest partitions can be efficiently removed, maintaining the most recent partitions.
 97        self._cursor_per_partition: OrderedDict[str, ConcurrentCursor] = OrderedDict()
 98        self._semaphore_per_partition: OrderedDict[str, threading.Semaphore] = OrderedDict()
 99
100        # Parent-state tracking: store each partition’s parent state in creation order
101        self._partition_parent_state_map: OrderedDict[str, Mapping[str, Any]] = OrderedDict()
102
103        self._finished_partitions: set[str] = set()
104        self._lock = threading.Lock()
105        self._timer = Timer()
106        self._new_global_cursor: Optional[StreamState] = None
107        self._lookback_window: int = 0
108        self._parent_state: Optional[StreamState] = None
109        self._number_of_partitions: int = 0
110        self._use_global_cursor: bool = False
111        self._partition_serializer = PerPartitionKeySerializer()
112        # Track the last time a state message was emitted
113        self._last_emission_time: float = 0.0
114
115        self._set_initial_state(stream_state)
116
117    @property
118    def cursor_field(self) -> CursorField:
119        return self._cursor_field
120
121    @property
122    def state(self) -> MutableMapping[str, Any]:
123        state: dict[str, Any] = {"use_global_cursor": self._use_global_cursor}
124        if not self._use_global_cursor:
125            states = []
126            for partition_tuple, cursor in self._cursor_per_partition.items():
127                if cursor.state:
128                    states.append(
129                        {
130                            "partition": self._to_dict(partition_tuple),
131                            "cursor": copy.deepcopy(cursor.state),
132                        }
133                    )
134            state[self._PERPARTITION_STATE_KEY] = states
135
136        if self._global_cursor:
137            state[self._GLOBAL_STATE_KEY] = self._global_cursor
138        if self._lookback_window is not None:
139            state["lookback_window"] = self._lookback_window
140        if self._parent_state is not None:
141            state["parent_state"] = self._parent_state
142        return state
143
144    def close_partition(self, partition: Partition) -> None:
145        # Attempt to retrieve the stream slice
146        stream_slice: Optional[StreamSlice] = partition.to_slice()  # type: ignore[assignment]
147
148        # Ensure stream_slice is not None
149        if stream_slice is None:
150            raise ValueError("stream_slice cannot be None")
151
152        partition_key = self._to_partition_key(stream_slice.partition)
153        with self._lock:
154            self._semaphore_per_partition[partition_key].acquire()
155            if not self._use_global_cursor:
156                self._cursor_per_partition[partition_key].close_partition(partition=partition)
157                cursor = self._cursor_per_partition[partition_key]
158                if (
159                    partition_key in self._finished_partitions
160                    and self._semaphore_per_partition[partition_key]._value == 0
161                ):
162                    self._update_global_cursor(cursor.state[self.cursor_field.cursor_field_key])
163
164            self._check_and_update_parent_state()
165
166            self._emit_state_message()
167
168    def _check_and_update_parent_state(self) -> None:
169        """
170        Pop the leftmost partition state from _partition_parent_state_map only if
171        *all partitions* up to (and including) that partition key in _semaphore_per_partition
172        are fully finished (i.e. in _finished_partitions and semaphore._value == 0).
173        Additionally, delete finished semaphores with a value of 0 to free up memory,
174        as they are only needed to track errors and completion status.
175        """
176        last_closed_state = None
177
178        while self._partition_parent_state_map:
179            # Look at the earliest partition key in creation order
180            earliest_key = next(iter(self._partition_parent_state_map))
181
182            # Verify ALL partitions from the left up to earliest_key are finished
183            all_left_finished = True
184            for p_key, sem in list(
185                self._semaphore_per_partition.items()
186            ):  # Use list to allow modification during iteration
187                # If any earlier partition is still not finished, we must stop
188                if p_key not in self._finished_partitions or sem._value != 0:
189                    all_left_finished = False
190                    break
191                # Once we've reached earliest_key in the semaphore order, we can stop checking
192                if p_key == earliest_key:
193                    break
194
195            # If the partitions up to earliest_key are not all finished, break the while-loop
196            if not all_left_finished:
197                break
198
199            # Pop the leftmost entry from parent-state map
200            _, closed_parent_state = self._partition_parent_state_map.popitem(last=False)
201            last_closed_state = closed_parent_state
202
203            # Clean up finished semaphores with value 0 up to and including earliest_key
204            for p_key in list(self._semaphore_per_partition.keys()):
205                sem = self._semaphore_per_partition[p_key]
206                if p_key in self._finished_partitions and sem._value == 0:
207                    del self._semaphore_per_partition[p_key]
208                    logger.debug(f"Deleted finished semaphore for partition {p_key} with value 0")
209                if p_key == earliest_key:
210                    break
211
212        # Update _parent_state if we popped at least one partition
213        if last_closed_state is not None:
214            self._parent_state = last_closed_state
215
216    def ensure_at_least_one_state_emitted(self) -> None:
217        """
218        The platform expects at least one state message on successful syncs. Hence, whatever happens, we expect this method to be
219        called.
220        """
221        if not any(
222            semaphore_item[1]._value for semaphore_item in self._semaphore_per_partition.items()
223        ):
224            self._global_cursor = self._new_global_cursor
225            self._lookback_window = self._timer.finish()
226            self._parent_state = self._partition_router.get_stream_state()
227        self._emit_state_message(throttle=False)
228
229    def _throttle_state_message(self) -> Optional[float]:
230        """
231        Throttles the state message emission to once every 60 seconds.
232        """
233        current_time = time.time()
234        if current_time - self._last_emission_time <= 60:
235            return None
236        return current_time
237
238    def _emit_state_message(self, throttle: bool = True) -> None:
239        if throttle:
240            current_time = self._throttle_state_message()
241            if current_time is None:
242                return
243            self._last_emission_time = current_time
244        self._connector_state_manager.update_state_for_stream(
245            self._stream_name,
246            self._stream_namespace,
247            self.state,
248        )
249        state_message = self._connector_state_manager.create_state_message(
250            self._stream_name, self._stream_namespace
251        )
252        self._message_repository.emit_message(state_message)
253
254    def stream_slices(self) -> Iterable[StreamSlice]:
255        if self._timer.is_running():
256            raise RuntimeError("stream_slices has been executed more than once.")
257
258        slices = self._partition_router.stream_slices()
259        self._timer.start()
260        for partition, last, parent_state in iterate_with_last_flag_and_state(
261            slices, self._partition_router.get_stream_state
262        ):
263            yield from self._generate_slices_from_partition(partition, parent_state)
264
265    def _generate_slices_from_partition(
266        self, partition: StreamSlice, parent_state: Mapping[str, Any]
267    ) -> Iterable[StreamSlice]:
268        # Ensure the maximum number of partitions is not exceeded
269        self._ensure_partition_limit()
270
271        partition_key = self._to_partition_key(partition.partition)
272
273        cursor = self._cursor_per_partition.get(self._to_partition_key(partition.partition))
274        if not cursor:
275            cursor = self._create_cursor(
276                self._global_cursor,
277                self._lookback_window if self._global_cursor else 0,
278            )
279            with self._lock:
280                self._number_of_partitions += 1
281                self._cursor_per_partition[partition_key] = cursor
282        self._semaphore_per_partition[partition_key] = threading.Semaphore(0)
283
284        with self._lock:
285            if (
286                len(self._partition_parent_state_map) == 0
287                or self._partition_parent_state_map[
288                    next(reversed(self._partition_parent_state_map))
289                ]
290                != parent_state
291            ):
292                self._partition_parent_state_map[partition_key] = deepcopy(parent_state)
293
294        for cursor_slice, is_last_slice, _ in iterate_with_last_flag_and_state(
295            cursor.stream_slices(),
296            lambda: None,
297        ):
298            self._semaphore_per_partition[partition_key].release()
299            if is_last_slice:
300                self._finished_partitions.add(partition_key)
301            yield StreamSlice(
302                partition=partition, cursor_slice=cursor_slice, extra_fields=partition.extra_fields
303            )
304
305    def _ensure_partition_limit(self) -> None:
306        """
307        Ensure the maximum number of partitions does not exceed the predefined limit.
308
309        Steps:
310        1. Attempt to remove partitions that are marked as finished in `_finished_partitions`.
311           These partitions are considered processed and safe to delete.
312        2. If the limit is still exceeded and no finished partitions are available for removal,
313           remove the oldest partition unconditionally. We expect failed partitions to be removed.
314
315        Logging:
316        - Logs a warning each time a partition is removed, indicating whether it was finished
317          or removed due to being the oldest.
318        """
319        if not self._use_global_cursor and self.limit_reached():
320            logger.info(
321                f"Exceeded the 'SWITCH_TO_GLOBAL_LIMIT' of {self.SWITCH_TO_GLOBAL_LIMIT}. "
322                f"Switching to global cursor for {self._stream_name}."
323            )
324            self._use_global_cursor = True
325
326        with self._lock:
327            while len(self._cursor_per_partition) > self.DEFAULT_MAX_PARTITIONS_NUMBER - 1:
328                # Try removing finished partitions first
329                for partition_key in list(self._cursor_per_partition.keys()):
330                    if partition_key in self._finished_partitions and (
331                        partition_key not in self._semaphore_per_partition
332                        or self._semaphore_per_partition[partition_key]._value == 0
333                    ):
334                        oldest_partition = self._cursor_per_partition.pop(
335                            partition_key
336                        )  # Remove the oldest partition
337                        logger.warning(
338                            f"The maximum number of partitions has been reached. Dropping the oldest finished partition: {oldest_partition}. Over limit: {self._number_of_partitions - self.DEFAULT_MAX_PARTITIONS_NUMBER}."
339                        )
340                        break
341                else:
342                    # If no finished partitions can be removed, fall back to removing the oldest partition
343                    oldest_partition = self._cursor_per_partition.popitem(last=False)[
344                        1
345                    ]  # Remove the oldest partition
346                    logger.warning(
347                        f"The maximum number of partitions has been reached. Dropping the oldest partition: {oldest_partition}. Over limit: {self._number_of_partitions - self.DEFAULT_MAX_PARTITIONS_NUMBER}."
348                    )
349
350    def _set_initial_state(self, stream_state: StreamState) -> None:
351        """
352        Initialize the cursor's state using the provided `stream_state`.
353
354        This method supports global and per-partition state initialization.
355
356        - **Global State**: If `states` is missing, the `state` is treated as global and applied to all partitions.
357          The `global state` holds a single cursor position representing the latest processed record across all partitions.
358
359        - **Lookback Window**: Configured via `lookback_window`, it defines the period (in seconds) for reprocessing records.
360          This ensures robustness in case of upstream data delays or reordering. If not specified, it defaults to 0.
361
362        - **Per-Partition State**: If `states` is present, each partition's cursor state is initialized separately.
363
364        - **Parent State**: (if available) Used to initialize partition routers based on parent streams.
365
366        Args:
367            stream_state (StreamState): The state of the streams to be set. The format of the stream state should be:
368                {
369                    "states": [
370                        {
371                            "partition": {
372                                "partition_key": "value"
373                            },
374                            "cursor": {
375                                "last_updated": "2023-05-27T00:00:00Z"
376                            }
377                        }
378                    ],
379                    "state": {
380                        "last_updated": "2023-05-27T00:00:00Z"
381                    },
382                    lookback_window: 10,
383                    "parent_state": {
384                        "parent_stream_name": {
385                            "last_updated": "2023-05-27T00:00:00Z"
386                        }
387                    }
388                }
389        """
390        if not stream_state:
391            return
392
393        if (
394            self._PERPARTITION_STATE_KEY not in stream_state
395            and self._GLOBAL_STATE_KEY not in stream_state
396        ):
397            # We assume that `stream_state` is in a global format that can be applied to all partitions.
398            # Example: {"global_state_format_key": "global_state_format_value"}
399            self._set_global_state(stream_state)
400
401        else:
402            self._use_global_cursor = stream_state.get("use_global_cursor", False)
403
404            self._lookback_window = int(stream_state.get("lookback_window", 0))
405
406            for state in stream_state.get(self._PERPARTITION_STATE_KEY, []):
407                self._number_of_partitions += 1
408                self._cursor_per_partition[self._to_partition_key(state["partition"])] = (
409                    self._create_cursor(state["cursor"])
410                )
411
412            # set default state for missing partitions if it is per partition with fallback to global
413            if self._GLOBAL_STATE_KEY in stream_state:
414                self._set_global_state(stream_state[self._GLOBAL_STATE_KEY])
415
416        # Set initial parent state
417        if stream_state.get("parent_state"):
418            self._parent_state = stream_state["parent_state"]
419
420        # Set parent state for partition routers based on parent streams
421        self._partition_router.set_initial_state(stream_state)
422
423    def _set_global_state(self, stream_state: Mapping[str, Any]) -> None:
424        """
425        Initializes the global cursor state from the provided stream state.
426
427        If the cursor field key is present in the stream state, its value is parsed,
428        formatted, and stored as the global cursor. This ensures consistency in state
429        representation across partitions.
430        """
431        if self.cursor_field.cursor_field_key in stream_state:
432            global_state_value = stream_state[self.cursor_field.cursor_field_key]
433            final_format_global_state_value = self._connector_state_converter.output_format(
434                self._connector_state_converter.parse_value(global_state_value)
435            )
436
437            fixed_global_state = {
438                self.cursor_field.cursor_field_key: final_format_global_state_value
439            }
440
441            self._global_cursor = deepcopy(fixed_global_state)
442            self._new_global_cursor = deepcopy(fixed_global_state)
443
444    def observe(self, record: Record) -> None:
445        if not record.associated_slice:
446            raise ValueError(
447                "Invalid state as stream slices that are emitted should refer to an existing cursor"
448            )
449
450        record_cursor = self._connector_state_converter.output_format(
451            self._connector_state_converter.parse_value(self._cursor_field.extract_value(record))
452        )
453        self._update_global_cursor(record_cursor)
454        if not self._use_global_cursor:
455            self._cursor_per_partition[
456                self._to_partition_key(record.associated_slice.partition)
457            ].observe(record)
458
459    def _update_global_cursor(self, value: Any) -> None:
460        if (
461            self._new_global_cursor is None
462            or self._new_global_cursor[self.cursor_field.cursor_field_key] < value
463        ):
464            self._new_global_cursor = {self.cursor_field.cursor_field_key: copy.deepcopy(value)}
465
466    def _to_partition_key(self, partition: Mapping[str, Any]) -> str:
467        return self._partition_serializer.to_partition_key(partition)
468
469    def _to_dict(self, partition_key: str) -> Mapping[str, Any]:
470        return self._partition_serializer.to_partition(partition_key)
471
472    def _create_cursor(
473        self, cursor_state: Any, runtime_lookback_window: int = 0
474    ) -> ConcurrentCursor:
475        cursor = self._cursor_factory.create(
476            stream_state=deepcopy(cursor_state),
477            runtime_lookback_window=timedelta(seconds=runtime_lookback_window),
478        )
479        return cursor
480
481    def should_be_synced(self, record: Record) -> bool:
482        return self._get_cursor(record).should_be_synced(record)
483
484    def _get_cursor(self, record: Record) -> ConcurrentCursor:
485        if not record.associated_slice:
486            raise ValueError(
487                "Invalid state as stream slices that are emitted should refer to an existing cursor"
488            )
489        partition_key = self._to_partition_key(record.associated_slice.partition)
490        if partition_key not in self._cursor_per_partition:
491            raise ValueError(
492                "Invalid state as stream slices that are emitted should refer to an existing cursor"
493            )
494        cursor = self._cursor_per_partition[partition_key]
495        return cursor
496
497    def limit_reached(self) -> bool:
498        return self._number_of_partitions > self.SWITCH_TO_GLOBAL_LIMIT

 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 is_greater_than_or_equal(self, first: Record, second: Record) -> bool:
434        cursor_field = self.cursor_field.eval(self.config)  # type: ignore  # cursor_field is converted to an InterpolatedString in __post_init__
435        first_cursor_value = first.get(cursor_field)
436        second_cursor_value = second.get(cursor_field)
437        if first_cursor_value and second_cursor_value:
438            return self.parse_date(first_cursor_value) >= self.parse_date(second_cursor_value)
439        elif first_cursor_value:
440            return True
441        else:
442            return False
443
444    def set_runtime_lookback_window(self, lookback_window_in_seconds: int) -> None:
445        """
446        Updates the lookback window based on a given number of seconds if the new duration
447        is greater than the currently configured lookback window.
448
449        :param lookback_window_in_seconds: The lookback duration in seconds to potentially update to.
450        """
451        runtime_lookback_window = duration_isoformat(timedelta(seconds=lookback_window_in_seconds))
452        config_lookback = parse_duration(
453            self._lookback_window.eval(self.config) if self._lookback_window else "P0D"
454        )
455
456        # Check if the new runtime lookback window is greater than the current config lookback
457        if parse_duration(runtime_lookback_window) > config_lookback:
458            self._lookback_window = InterpolatedString.create(
459                runtime_lookback_window, parameters={}
460            )