Coverage for formkit_ninja / parser / database_node_path.py: 7.62%

1""" 

2DatabaseNodePath: NodePath that reads configuration from database. 

3 

4This module provides DatabaseNodePath, a NodePath subclass that queries 

5the CodeGenerationConfig model for type mappings and field arguments, 

6enabling database-driven code generation configuration. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING 

12 

13from django.conf import settings 

14 

15from formkit_ninja.parser.type_convert import NodePath 

16 

17if TYPE_CHECKING: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 from formkit_ninja.code_generation_config import CodeGenerationConfig 

19 

20 

21class DatabaseNodePath(NodePath): 

22 """ 

23 NodePath that reads type mappings from database. 

24 

25 Priority cascade: 

26 1. CodeGenerationConfig matching node_name (if exists) 

27 2. CodeGenerationConfig matching options_pattern 

28 3. CodeGenerationConfig matching formkit_type 

29 4. Django settings FORMKIT_NINJA['TYPE_MAPPINGS'] 

30 5. Default TypeConverterRegistry 

31 

32 Example usage: 

33 config = GeneratorConfig( 

34 app_name="myapp", 

35 output_dir=Path("./generated"), 

36 node_path_class=DatabaseNodePath, 

37 ) 

38 """ 

39 

40 def __init__(self, *nodes, **kwargs): 

41 super().__init__(*nodes, **kwargs) 

42 self._config_cache: dict[str, CodeGenerationConfig | None] = {} 

43 

44 def _get_config(self) -> "CodeGenerationConfig | None": 

45 """ 

46 Get matching CodeGenerationConfig for current node. 

47 

48 Returns the highest priority config that matches: 

49 1. node_name (exact match) 

50 2. options_pattern (startswith match) 

51 3. formkit_type (exact match, no node_name/options_pattern set) 

52 

53 Returns: 

54 Matching CodeGenerationConfig or None 

55 """ 

56 # Build cache key (include options to handle pattern matching) 

57 formkit_val = getattr(self.node, "formkit", "") 

58 name_val = getattr(self.node, "name", "") 

59 options_val = str(getattr(self.node, "options", "")) 

60 cache_key = f"{formkit_val}-{name_val}-{options_val}" 

61 

62 if cache_key in self._config_cache: 

63 return self._config_cache[cache_key] 

64 

65 from formkit_ninja.code_generation_config import CodeGenerationConfig 

66 

67 # Priority 1: Try node_name match (highest priority) 

68 if hasattr(self.node, "name") and self.node.name: 

69 config = ( 

70 CodeGenerationConfig.objects.filter( # type: ignore[attr-defined] 

71 is_active=True, 

72 node_name=self.node.name, 

73 ) 

74 .order_by("-priority") 

75 .first() 

76 ) 

77 if config: 

78 self._config_cache[cache_key] = config 

79 return config 

80 

81 # Priority 2: Try options pattern match 

82 if hasattr(self.node, "options") and self.node.options: 

83 options_str = str(self.node.options) 

84 

85 # Build filter - include formkit_type if available 

86 filter_kwargs = { 

87 "is_active": True, 

88 "options_pattern__isnull": False, 

89 } 

90 if hasattr(self.node, "formkit"): 

91 filter_kwargs["formkit_type"] = self.node.formkit 

92 

93 configs = CodeGenerationConfig.objects.filter(**filter_kwargs).order_by("-priority") # type: ignore[attr-defined] 

94 

95 for cfg in configs: 

96 if cfg.options_pattern and options_str.startswith(cfg.options_pattern): 

97 self._config_cache[cache_key] = cfg 

98 return cfg 

99 

100 # Priority 3: Try formkit_type match (no node_name/options_pattern) 

101 if hasattr(self.node, "formkit"): 

102 config = ( 

103 CodeGenerationConfig.objects.filter( # type: ignore[attr-defined] 

104 is_active=True, 

105 formkit_type=self.node.formkit, 

106 node_name__isnull=True, 

107 options_pattern__isnull=True, 

108 ) 

109 .order_by("-priority") 

110 .first() 

111 ) 

112 if config: 

113 self._config_cache[cache_key] = config 

114 return config 

115 

116 self._config_cache[cache_key] = None 

117 return None 

118 

119 def _get_from_settings(self, field: str) -> str | dict | None: 

120 """ 

121 Get configuration from Django settings. 

122 

123 Checks FORMKIT_NINJA settings in order: 

124 1. NAME_MAPPINGS[node.name][field] 

125 2. OPTIONS_MAPPINGS (pattern match) 

126 3. TYPE_MAPPINGS[node.formkit][field] 

127 

128 Args: 

129 field: Field name ('pydantic_type', 'django_type', 'django_args') 

130 

131 Returns: 

132 Configuration value or None 

133 """ 

134 formkit_settings = getattr(settings, "FORMKIT_NINJA", {}) 

135 

136 # Check NAME_MAPPINGS first (highest priority in settings) 

137 if hasattr(self.node, "name") and self.node.name: 

138 name_mappings = formkit_settings.get("NAME_MAPPINGS", {}) 

139 if self.node.name in name_mappings: 

140 mapping = name_mappings[self.node.name] 

141 if field in mapping: 

142 return mapping[field] 

143 

144 # Check OPTIONS_MAPPINGS 

145 if hasattr(self.node, "options") and self.node.options: 

146 options_str = str(self.node.options) 

147 options_mappings = formkit_settings.get("OPTIONS_MAPPINGS", {}) 

148 for pattern, mapping in options_mappings.items(): 

149 if options_str.startswith(pattern): 

150 if field in mapping: 

151 return mapping[field] 

152 

153 # Check TYPE_MAPPINGS 

154 if hasattr(self.node, "formkit"): 

155 type_mappings = formkit_settings.get("TYPE_MAPPINGS", {}) 

156 if self.node.formkit in type_mappings: 

157 mapping = type_mappings[self.node.formkit] 

158 if field in mapping: 

159 return mapping[field] 

160 

161 return None 

162 

163 def to_pydantic_type(self) -> str: 

164 """ 

165 Get Pydantic type for this node. 

166 Prioritizes fields already on the node (via super()). 

167 """ 

168 # 1. Check if it's already on the node 

169 res = super().to_pydantic_type() 

170 # super().to_pydantic_type() returns converter.pydantic_type if not found on node. 

171 # We need to know if it was found on the node or not. 

172 if hasattr(self.node, "pydantic_field_type") and self.node.pydantic_field_type: 

173 return res 

174 

175 # 2. Check database config 

176 config = self._get_config() 

177 if config and config.pydantic_type: 

178 return config.pydantic_type 

179 

180 # Check Django settings 

181 settings_type = self._get_from_settings("pydantic_type") 

182 if settings_type: 

183 return str(settings_type) 

184 

185 # Fall back to parent implementation (default converters) 

186 return super().to_pydantic_type() 

187 

188 def to_django_type(self) -> str: 

189 """ 

190 Get Django field type for this node. 

191 Prioritizes fields already on the node (via super()). 

192 """ 

193 # 1. Check if it's already on the node 

194 if hasattr(self.node, "django_field_type") and self.node.django_field_type: 

195 return self.node.django_field_type 

196 

197 # 2. Check database config 

198 config = self._get_config() 

199 if config and config.django_type: 

200 return config.django_type 

201 

202 # Check Django settings 

203 settings_type = self._get_from_settings("django_type") 

204 if settings_type: 

205 return str(settings_type) 

206 

207 # Fall back to parent implementation 

208 return super().to_django_type() 

209 

210 def to_django_args(self) -> str: 

211 """ 

212 Get Django field arguments for this node. 

213 Prioritizes fields already on the node (via super()). 

214 """ 

215 # 1. Check if it's already on the node 

216 args = getattr(self.node, "django_field_args", {}) 

217 pos_args = getattr(self.node, "django_field_positional_args", []) 

218 if args or pos_args: 

219 return super().to_django_args() 

220 

221 # 2. Check database config 

222 config = self._get_config() 

223 if config: 

224 return config.get_django_args_str() 

225 

226 # Check Django settings 

227 settings_args = self._get_from_settings("django_args") 

228 settings_pos_args = self._get_from_settings("django_positional_args") 

229 

230 if settings_args or settings_pos_args: 

231 return self._django_args_dict_to_str( 

232 settings_args if isinstance(settings_args, dict) else {}, 

233 settings_pos_args if isinstance(settings_pos_args, list) else [], 

234 ) 

235 

236 # Fall back to parent implementation 

237 return super().to_django_args() 

238 

239 def get_validators(self) -> list[str]: 

240 """ 

241 Get validators for this node. 

242 Prioritizes fields already on the node (via super()). 

243 """ 

244 # 1. Check if it's already on the node (handled by super()) 

245 res = super().get_validators() 

246 # super().get_validators() returns an empty list if not found on node. 

247 # If `res` is not empty, it means it came from the node. 

248 if res: 

249 return res 

250 

251 # 2. Check database config 

252 config = self._get_config() 

253 if config and config.validators: 

254 return config.validators 

255 

256 # Check settings 

257 settings_validators = self._get_from_settings("validators") 

258 if settings_validators and isinstance(settings_validators, list): 

259 return settings_validators 

260 

261 return [] 

262 

263 def get_extra_imports(self) -> list[str]: 

264 """ 

265 Get extra imports for this node. 

266 Prioritizes fields already on the node (via super()). 

267 """ 

268 # 1. Check if it's already on the node (handled by super()) 

269 res = super().get_extra_imports() 

270 # super().get_extra_imports() returns an empty list if not found on node. 

271 # If `res` is not empty, it means it came from the node. 

272 if res: 

273 return res 

274 

275 # 2. Check database config 

276 config = self._get_config() 

277 if config and config.extra_imports: 

278 return config.extra_imports 

279 

280 # Check settings 

281 settings_imports = self._get_from_settings("extra_imports") 

282 if settings_imports and isinstance(settings_imports, list): 

283 return settings_imports 

284 

285 return super().get_extra_imports() 

286 

287 @staticmethod 

288 def _django_args_dict_to_str(args_dict: dict, positional_args: list | None = None) -> str: 

289 """ 

290 Convert django_args dict and positional_args list to string format. 

291 

292 Args: 

293 args_dict: Dict of keyword field arguments 

294 positional_args: List of positional field arguments 

295 

296 Returns: 

297 Comma-separated string of arguments 

298 """ 

299 parts = [] 

300 

301 # Handle positional arguments first 

302 if positional_args: 

303 for value in positional_args: 

304 parts.append(str(value)) 

305 

306 # Handle keyword arguments 

307 for key, value in args_dict.items(): 

308 if isinstance(value, bool): 

309 parts.append(f"{key}={str(value)}") 

310 elif isinstance(value, (int, float)): 

311 parts.append(f"{key}={value}") 

312 elif isinstance(value, str): 

313 # Handle model references (e.g., "app.Model" or models.CASCADE) 

314 if value in {"True", "False", "None"}: 

315 parts.append(f"{key}={value}") 

316 elif value.startswith("models.") or ("." in value and not value.startswith('"')): 

317 parts.append(f"{key}={value}") 

318 else: 

319 parts.append(f'{key}="{value}"') 

320 else: 

321 parts.append(f"{key}={value}") 

322 

323 return ", ".join(parts)