Convert milliseconds since epoch UTC to datetime.
(
millis: int, opts: CodecOptions[Any]
)
| 146 | |
| 147 | |
| 148 | def _millis_to_datetime( |
| 149 | millis: int, opts: CodecOptions[Any] |
| 150 | ) -> Union[datetime.datetime, DatetimeMS]: |
| 151 | """Convert milliseconds since epoch UTC to datetime.""" |
| 152 | if ( |
| 153 | opts.datetime_conversion == DatetimeConversion.DATETIME |
| 154 | or opts.datetime_conversion == DatetimeConversion.DATETIME_CLAMP |
| 155 | or opts.datetime_conversion == DatetimeConversion.DATETIME_AUTO |
| 156 | ): |
| 157 | tz = opts.tzinfo or utc |
| 158 | if opts.datetime_conversion == DatetimeConversion.DATETIME_CLAMP: |
| 159 | millis = max(_min_datetime_ms(tz), min(millis, _max_datetime_ms(tz))) |
| 160 | elif opts.datetime_conversion == DatetimeConversion.DATETIME_AUTO: |
| 161 | if not (_min_datetime_ms(tz) <= millis <= _max_datetime_ms(tz)): |
| 162 | return DatetimeMS(millis) |
| 163 | |
| 164 | diff = ((millis % 1000) + 1000) % 1000 |
| 165 | seconds = (millis - diff) // 1000 |
| 166 | micros = diff * 1000 |
| 167 | |
| 168 | try: |
| 169 | if opts.tz_aware: |
| 170 | dt = EPOCH_AWARE + datetime.timedelta(seconds=seconds, microseconds=micros) |
| 171 | if opts.tzinfo: |
| 172 | dt = dt.astimezone(tz) |
| 173 | return dt |
| 174 | else: |
| 175 | return EPOCH_NAIVE + datetime.timedelta(seconds=seconds, microseconds=micros) |
| 176 | except ArithmeticError as err: |
| 177 | raise InvalidBSON(f"{err} {_DATETIME_ERROR_SUGGESTION}") from err |
| 178 | |
| 179 | elif opts.datetime_conversion == DatetimeConversion.DATETIME_MS: |
| 180 | return DatetimeMS(millis) |
| 181 | else: |
| 182 | raise ValueError("datetime_conversion must be an element of DatetimeConversion") |
no test coverage detected