Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/common/types.py: 60%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17from __future__ import annotations 

18 

19import re 

20from dataclasses import dataclass 

21from datetime import timedelta 

22from enum import Enum 

23from typing import Annotated, Literal 

24 

25from pydantic import ( 

26 AfterValidator, 

27 AliasGenerator, 

28 AwareDatetime, 

29 BaseModel, 

30 BeforeValidator, 

31 ConfigDict, 

32 field_validator, 

33 model_serializer, 

34 model_validator, 

35) 

36 

37from airflow._shared.timezones import timezone 

38 

39UtcDateTime = Annotated[AwareDatetime, AfterValidator(lambda d: d.astimezone(timezone.utc))] 

40"""UTCDateTime is a datetime with timezone information""" 

41 

42 

43def _validate_timedelta_field(td: timedelta | None) -> TimeDelta | None: 

44 """Validate the timedelta field and return it.""" 

45 if td is None: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return None 

47 return TimeDelta( 

48 days=td.days, 

49 seconds=td.seconds, 

50 microseconds=td.microseconds, 

51 ) 

52 

53 

54class TimeDelta(BaseModel): 

55 """TimeDelta can be used to interact with datetime.timedelta objects.""" 

56 

57 object_type: str = "TimeDelta" 

58 days: int 

59 seconds: int 

60 microseconds: int 

61 

62 model_config = ConfigDict( 

63 alias_generator=AliasGenerator( 

64 serialization_alias=lambda field_name: { 

65 "object_type": "__type", 

66 }.get(field_name, field_name), 

67 ) 

68 ) 

69 

70 

71TimeDeltaWithValidation = Annotated[TimeDelta, BeforeValidator(_validate_timedelta_field)] 

72 

73 

74# Common validator for theme icon fields (SVG-only, http(s) or app-relative path). 

75def _validate_theme_icon(value: str | None) -> str | None: 

76 if value is None: 

77 return value 

78 from urllib.parse import urlparse 

79 

80 parsed = urlparse(value) 

81 if parsed.scheme in ("http", "https"): 

82 path = parsed.path or "" 

83 elif parsed.scheme == "" and value.startswith("/"): 

84 path = value 

85 else: 

86 raise ValueError("theme.icon must be http(s) URL or app-relative path starting with '/'") 

87 if not path.lower().endswith(".svg"): 

88 raise ValueError("theme.icon must point to an SVG file (*.svg)") 

89 return value 

90 

91 

92# Alias type for theme icon fields with shared validation 

93ThemeIconType = Annotated[str | None, BeforeValidator(_validate_theme_icon)] 

94 

95 

96class Mimetype(str, Enum): 

97 """Mimetype for the `Content-Type` header.""" 

98 

99 TEXT = "text/plain" 

100 JSON = "application/json" 

101 FORM = "application/x-www-form-urlencoded" 

102 NDJSON = "application/x-ndjson" 

103 ANY = "*/*" 

104 

105 

106@dataclass 

107class ExtraMenuItem: 

108 """Define a menu item that can be added to the menu by auth managers or plugins.""" 

109 

110 text: str 

111 href: str 

112 

113 

114class MenuItem(Enum): 

115 """Define all menu items defined in the menu.""" 

116 

117 REQUIRED_ACTIONS = "Required Actions" 

118 ASSETS = "Assets" 

119 AUDIT_LOG = "Audit Log" 

120 CONFIG = "Config" 

121 CONNECTIONS = "Connections" 

122 DAGS = "Dags" 

123 DEADLINES = "Deadlines" 

124 DOCS = "Docs" 

125 JOBS = "Jobs" 

126 PLUGINS = "Plugins" 

127 POOLS = "Pools" 

128 PROVIDERS = "Providers" 

129 VARIABLES = "Variables" 

130 XCOMS = "XComs" 

131 

132 

133class UIAlert(BaseModel): 

134 """Optional alert to be shown at the top of the page.""" 

135 

136 text: str 

137 category: Literal["info", "warning", "error"] 

138 

139 

140class OklchColor(BaseModel): 

141 """Validates OKLCH color format from string oklch(l c h).""" 

142 

143 lightness: float 

144 chroma: float 

145 hue: float 

146 

147 @model_validator(mode="before") 

148 @classmethod 

149 def parse_oklch_string(cls, data): 

150 if isinstance(data, str): 

151 oklch_regex_pattern = r"^oklch\((-?\d+(?:\.\d+)?) (-?\d+(?:\.\d+)?) (-?\d+(?:\.\d+)?)\)$" 

152 match = re.match(oklch_regex_pattern, data) 

153 

154 if not match: 

155 raise ValueError(f"Invalid OKLCH format: {data} Expected format oklch(l c h)") 

156 

157 ligthness_str, chroma_str, hue_str = match.groups() 

158 

159 return { 

160 "lightness": float(ligthness_str), 

161 "chroma": float(chroma_str), 

162 "hue": float(hue_str), 

163 } 

164 return data 

165 

166 @field_validator("lightness") 

167 @classmethod 

168 def validate_lightness(cls, value: float) -> float: 

169 if value < 0 or value > 1: 

170 raise ValueError(f"Invalid lightness: {value} Must be between 0 and 1") 

171 return value 

172 

173 @field_validator("chroma") 

174 @classmethod 

175 def validate_chroma(cls, value: float) -> float: 

176 if value < 0 or value > 0.5: 

177 raise ValueError(f"Invalid chroma: {value} Must be between 0 and 0.5") 

178 return value 

179 

180 @field_validator("hue") 

181 @classmethod 

182 def validate_hue(cls, value: float) -> float: 

183 if value < 0 or value > 360: 

184 raise ValueError(f"Invalid hue: {value} Must be between 0 and 360") 

185 return value 

186 

187 @model_serializer(mode="plain") 

188 def serialize_model(self) -> str: 

189 return f"oklch({self.lightness} {self.chroma} {self.hue})" 

190 

191 

192# Private type aliases for theme token shapes 

193_ColorShade = dict[Literal["value"], OklchColor] 

194_SHADE_LITERAL = Literal["50", "100", "200", "300", "400", "500", "600", "700", "800", "900", "950"] 

195_ColorScale = dict[_SHADE_LITERAL, _ColorShade] 

196 

197 

198class ThemeColors(BaseModel): 

199 """Color tokens for the UI theme. All fields are optional; at least one must be provided.""" 

200 

201 brand: _ColorScale | None = None 

202 gray: _ColorScale | None = None 

203 black: _ColorShade | None = None 

204 white: _ColorShade | None = None 

205 

206 @model_validator(mode="after") 

207 def check_at_least_one_color(self) -> ThemeColors: 

208 if not any([self.brand, self.gray, self.black, self.white]): 

209 raise ValueError("At least one color token must be provided: brand, gray, black, or white") 

210 return self 

211 

212 

213class Theme(BaseModel): 

214 """JSON to modify Chakra's theme.""" 

215 

216 tokens: dict[Literal["colors"], ThemeColors] | None = None 

217 globalCss: dict[str, dict] | None = None 

218 icon: ThemeIconType = None 

219 icon_dark_mode: ThemeIconType = None