airbyte_cdk.sources.declarative.datetime.min_max_datetime

  1#
  2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
  3#
  4
  5import datetime as dt
  6from dataclasses import InitVar, dataclass, field
  7from typing import Any, Mapping, Optional, Union
  8
  9from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
 10from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 11
 12
 13@dataclass
 14class MinMaxDatetime:
 15    """
 16    Compares the provided date against optional minimum or maximum times. If date is earlier than
 17    min_date, then min_date is returned. If date is greater than max_date, then max_date is returned.
 18    If neither, the input date is returned.
 19
 20    The timestamp format accepts the same format codes as datetime.strfptime, which are
 21    all the format codes required by the 1989 C standard.
 22    Full list of accepted format codes: https://man7.org/linux/man-pages/man3/strftime.3.html
 23
 24    Attributes:
 25        datetime (Union[InterpolatedString, str]): InterpolatedString or string representing the datetime in the format specified by `datetime_format`
 26        datetime_format (str): Format of the datetime passed as argument
 27        min_datetime (Union[InterpolatedString, str]): Represents the minimum allowed datetime value.
 28        max_datetime (Union[InterpolatedString, str]): Represents the maximum allowed datetime value.
 29    """
 30
 31    datetime: Union[InterpolatedString, str]
 32    parameters: InitVar[Mapping[str, Any]]
 33    # datetime_format is a unique case where we inherit it from the parent if it is not specified before using the default value
 34    # which is why we need dedicated getter/setter methods and private dataclass field
 35    datetime_format: str
 36    _datetime_format: str = field(init=False, repr=False, default="")
 37    min_datetime: Union[InterpolatedString, str] = ""
 38    max_datetime: Union[InterpolatedString, str] = ""
 39
 40    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
 41        self.datetime = InterpolatedString.create(self.datetime, parameters=parameters or {})
 42        self._parser = DatetimeParser()
 43        self.min_datetime = (
 44            InterpolatedString.create(self.min_datetime, parameters=parameters)  # type: ignore [assignment]  #  expression has type "InterpolatedString | None", variable has type "InterpolatedString | str"
 45            if self.min_datetime
 46            else None
 47        )  # type: ignore
 48        self.max_datetime = (
 49            InterpolatedString.create(self.max_datetime, parameters=parameters)  # type: ignore [assignment]  #  expression has type "InterpolatedString | None", variable has type "InterpolatedString | str"
 50            if self.max_datetime
 51            else None
 52        )  # type: ignore
 53
 54    def get_datetime(
 55        self, config: Mapping[str, Any], **additional_parameters: Mapping[str, Any]
 56    ) -> dt.datetime:
 57        """
 58        Evaluates and returns the datetime
 59        :param config: The user-provided configuration as specified by the source's spec
 60        :param additional_parameters: Additional arguments to be passed to the strings for interpolation
 61        :return: The evaluated datetime
 62        """
 63        # We apply a default datetime format here instead of at instantiation, so it can be set by the parent first
 64        datetime_format = self._datetime_format
 65        if not datetime_format:
 66            datetime_format = "%Y-%m-%dT%H:%M:%S.%f%z"
 67
 68        time = self._parser.parse(
 69            str(
 70                self.datetime.eval(  # type: ignore[union-attr] # str has no attribute "eval"
 71                    config,
 72                    **additional_parameters,
 73                )
 74            ),
 75            datetime_format,
 76        )  # type: ignore # datetime is always cast to an interpolated string
 77
 78        if self.min_datetime:
 79            min_time = str(self.min_datetime.eval(config, **additional_parameters))  # type: ignore # min_datetime is always cast to an interpolated string
 80            if min_time:
 81                min_datetime = self._parser.parse(min_time, datetime_format)  # type: ignore # min_datetime is always cast to an interpolated string
 82                time = max(time, min_datetime)
 83        if self.max_datetime:
 84            max_time = str(self.max_datetime.eval(config, **additional_parameters))  # type: ignore # max_datetime is always cast to an interpolated string
 85            if max_time:
 86                max_datetime = self._parser.parse(max_time, datetime_format)
 87                time = min(time, max_datetime)
 88        return time
 89
 90    @property  # type: ignore # properties don't play well with dataclasses...
 91    def datetime_format(self) -> str:
 92        """The format of the string representing the datetime"""
 93        return self._datetime_format
 94
 95    @datetime_format.setter
 96    def datetime_format(self, value: str) -> None:
 97        """Setter for the datetime format"""
 98        # Covers the case where datetime_format is not provided in the constructor, which causes the property object
 99        # to be set which we need to avoid doing
100        if not isinstance(value, property):
101            self._datetime_format = value
102
103    @classmethod
104    def create(
105        cls,
106        interpolated_string_or_min_max_datetime: Union[InterpolatedString, str, "MinMaxDatetime"],
107        parameters: Optional[Mapping[str, Any]] = None,
108    ) -> "MinMaxDatetime":
109        if parameters is None:
110            parameters = {}
111        if isinstance(interpolated_string_or_min_max_datetime, InterpolatedString) or isinstance(
112            interpolated_string_or_min_max_datetime, str
113        ):
114            return MinMaxDatetime(  # type: ignore [call-arg]
115                datetime=interpolated_string_or_min_max_datetime, parameters=parameters
116            )
117        else:
118            return interpolated_string_or_min_max_datetime