Coverage for /usr/local/lib/python3.12/site-packages/prefect/types/_datetime.py: 20%

1from __future__ import annotations 1 ctx1a

2 

3import datetime 1 ctx1a

4import sys 1 ctx1a

5from contextlib import contextmanager 1 ctx1a

6from typing import Any, cast 1 ctx1a

7from unittest import mock 1 ctx1a

8from zoneinfo import ZoneInfo, ZoneInfoNotFoundError, available_timezones 1 ctx1a

9 

10import humanize 1 ctx1a

11from dateutil.parser import parse 1 ctx1a

12from typing_extensions import TypeAlias 1 ctx1a

13 

14if sys.version_info >= (3, 13): 14 ↛ 15line 14 didn't jump to line 15 because the condition on line 14 was never true1 ctx1a

15 DateTime: TypeAlias = datetime.datetime 

16 Date: TypeAlias = datetime.date 

17 Duration: TypeAlias = datetime.timedelta 

18else: 

19 import pendulum 1 ctx1a

20 import pendulum.tz 1 ctx1a

21 from pydantic_extra_types.pendulum_dt import ( 1 ctx1a

22 Date as PydanticDate, 

23 ) 

24 from pydantic_extra_types.pendulum_dt import ( 1 ctx1a

25 DateTime as PydanticDateTime, 

26 ) 

27 from pydantic_extra_types.pendulum_dt import ( 1 ctx1a

28 Duration as PydanticDuration, 

29 ) 

30 

31 DateTime: TypeAlias = PydanticDateTime 1 ctx1a

32 Date: TypeAlias = PydanticDate 1 ctx1a

33 Duration: TypeAlias = PydanticDuration 1 ctx1a

34 

35 

36def parse_datetime(dt: str) -> datetime.datetime: 1 ctx1a

37 if sys.version_info >= (3, 13): 

38 parsed_dt = parse(dt) 

39 if parsed_dt.tzinfo is None: 

40 # Assume UTC if no timezone is provided 

41 return parsed_dt.replace(tzinfo=ZoneInfo("UTC")) 

42 else: 

43 return parsed_dt 

44 else: 

45 return cast(datetime.datetime, pendulum.parse(dt)) 

46 

47 

48def get_timezones() -> tuple[str, ...]: 1 ctx1a

49 return tuple(available_timezones()) 

50 

51 

52def create_datetime_instance(v: datetime.datetime) -> datetime.datetime: 1 ctx1a

53 if sys.version_info >= (3, 13): 

54 if v.tzinfo is None: 

55 # Assume UTC if no timezone is provided 

56 return v.replace(tzinfo=ZoneInfo("UTC")) 

57 else: 

58 return v 

59 

60 return DateTime.instance(v) 

61 

62 

63def from_timestamp(ts: float, tz: str | Any = "UTC") -> datetime.datetime: 1 ctx1a

64 if sys.version_info >= (3, 13): 

65 if not isinstance(tz, str): 

66 # Handle pendulum edge case 

67 tz = tz.name 

68 return datetime.datetime.fromtimestamp(ts, ZoneInfo(tz)) 

69 

70 return pendulum.from_timestamp(ts, tz) 

71 

72 

73def human_friendly_diff( 1 ctx1a

74 dt: datetime.datetime | None, other: datetime.datetime | None = None 

75) -> str: 

76 if dt is None: 

77 return "" 

78 

79 def _normalize(ts: datetime.datetime) -> datetime.datetime: 

80 """Return *ts* with a valid ZoneInfo; fall back to UTC if needed.""" 

81 if ts.tzinfo is None: 

82 local_tz = datetime.datetime.now().astimezone().tzinfo 

83 return ts.replace(tzinfo=local_tz).astimezone(ZoneInfo("UTC")) 

84 

85 if isinstance(ts.tzinfo, ZoneInfo): 

86 return ts # already valid 

87 

88 if tz_name := getattr(ts.tzinfo, "name", None): 

89 try: 

90 return ts.replace(tzinfo=ZoneInfo(tz_name)) 

91 except ZoneInfoNotFoundError: 

92 pass 

93 

94 return ts.astimezone(ZoneInfo("UTC")) 

95 

96 dt = _normalize(dt) 

97 

98 if other is not None: 

99 other = _normalize(other) 

100 

101 if sys.version_info >= (3, 13): 

102 # humanize expects ZoneInfo or None 

103 return humanize.naturaltime(dt, when=other) 

104 

105 # Ensure consistency for pendulum path by using UTC 

106 pendulum_dt = DateTime.instance(dt.astimezone(ZoneInfo("UTC"))) 

107 pendulum_other = ( 

108 DateTime.instance(other.astimezone(ZoneInfo("UTC"))) if other else None 

109 ) 

110 return pendulum_dt.diff_for_humans(other=pendulum_other) 

111 

112 

113def now( 1 ctx1a

114 tz: str | Any = "UTC", 

115) -> datetime.datetime: 

116 if sys.version_info >= (3, 13): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true3 ctx1bcd

117 from whenever import ZonedDateTime 

118 

119 if isinstance(getattr(tz, "name", None), str): 

120 tz = tz.name 

121 

122 return ZonedDateTime.now(tz).py_datetime() 

123 else: 

124 return pendulum.now(tz) 3 ctx1bcd

125 

126 

127def end_of_period(dt: datetime.datetime, period: str) -> datetime.datetime: 1 ctx1a

128 """ 

129 Returns the end of the specified unit of time. 

130 

131 Args: 

132 dt: The datetime to get the end of. 

133 period: The period to get the end of. 

134 Valid values: 'second', 'minute', 'hour', 'day', 

135 'week' 

136 

137 Returns: 

138 DateTime: A new DateTime representing the end of the specified unit. 

139 

140 Raises: 

141 ValueError: If an invalid unit is specified. 

142 """ 

143 if sys.version_info >= (3, 13): 

144 from whenever import Weekday, ZonedDateTime, days 

145 

146 if not isinstance(dt.tzinfo, ZoneInfo): 

147 zdt = ZonedDateTime.from_py_datetime( 

148 dt.replace(tzinfo=ZoneInfo(dt.tzname() or "UTC")) 

149 ) 

150 else: 

151 zdt = ZonedDateTime.from_py_datetime(dt) 

152 if period == "second": 

153 zdt = zdt.replace(nanosecond=999999999) 

154 elif period == "minute": 

155 zdt = zdt.replace(second=59, nanosecond=999999999) 

156 elif period == "hour": 

157 zdt = zdt.replace(minute=59, second=59, nanosecond=999999999) 

158 elif period == "day": 

159 zdt = zdt.replace(hour=23, minute=59, second=59, nanosecond=999999999) 

160 elif period == "week": 

161 days_till_end_of_week: int = ( 

162 Weekday.SUNDAY.value - zdt.date().day_of_week().value 

163 ) 

164 zdt = zdt + days(days_till_end_of_week) 

165 zdt = zdt.replace( 

166 hour=23, 

167 minute=59, 

168 second=59, 

169 nanosecond=999999999, 

170 ) 

171 else: 

172 raise ValueError(f"Invalid period: {period}") 

173 

174 return zdt.py_datetime() 

175 else: 

176 return DateTime.instance(dt).end_of(period) 

177 

178 

179def start_of_day(dt: datetime.datetime | DateTime) -> datetime.datetime: 1 ctx1a

180 """ 

181 Returns the start of the specified unit of time. 

182 

183 Args: 

184 dt: The datetime to get the start of. 

185 

186 Returns: 

187 datetime.datetime: A new datetime.datetime representing the start of the specified unit. 

188 

189 Raises: 

190 ValueError: If an invalid unit is specified. 

191 """ 

192 if sys.version_info >= (3, 13): 

193 from whenever import ZonedDateTime 

194 

195 if hasattr(dt, "tz"): 

196 zdt = ZonedDateTime.from_timestamp( 

197 dt.timestamp(), tz=dt.tz.name if dt.tz else "UTC" 

198 ) 

199 else: 

200 zdt = ZonedDateTime.from_py_datetime(dt) 

201 

202 return zdt.start_of_day().py_datetime() 

203 else: 

204 return DateTime.instance(dt).start_of("day") 

205 

206 

207def earliest_possible_datetime() -> datetime.datetime: 1 ctx1a

208 return datetime.datetime.min.replace(tzinfo=ZoneInfo("UTC")) 1 ctx1a

209 

210 

211@contextmanager 1 ctx1a

212def travel_to(dt: Any): 1 ctx1a

213 if sys.version_info >= (3, 13): 

214 with mock.patch("prefect.types._datetime.now", return_value=dt): 

215 yield 

216 

217 else: 

218 from pendulum import travel_to 

219 

220 with travel_to(dt, freeze=True): 

221 yield 

222 

223 

224def in_local_tz(dt: datetime.datetime) -> datetime.datetime: 1 ctx1a

225 if sys.version_info >= (3, 13): 

226 from whenever import PlainDateTime, ZonedDateTime 

227 

228 if dt.tzinfo is None: 

229 wdt = PlainDateTime.from_py_datetime(dt) 

230 else: 

231 if not isinstance(dt.tzinfo, ZoneInfo): 

232 if key := getattr(dt.tzinfo, "key", None): 

233 dt = dt.replace(tzinfo=ZoneInfo(key)) 

234 else: 

235 utc_dt = dt.astimezone(datetime.timezone.utc) 

236 dt = utc_dt.replace(tzinfo=ZoneInfo("UTC")) 

237 

238 wdt = ZonedDateTime.from_py_datetime(dt).to_system_tz() 

239 

240 return wdt.py_datetime() 

241 

242 return DateTime.instance(dt).in_tz(pendulum.tz.local_timezone()) 

243 

244 

245def to_datetime_string(dt: datetime.datetime, include_tz: bool = True) -> str: 1 ctx1a

246 if include_tz: 

247 return dt.strftime("%Y-%m-%d %H:%M:%S %Z") 

248 else: 

249 return dt.strftime("%Y-%m-%d %H:%M:%S")