Coverage for src / mafw / steering / models.py: 97%

1# Copyright 2026 European Union 

2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu) 

3# SPDX-License-Identifier: EUPL-1.2 

4"""Domain models for steering metadata construction. 

5 

6:Author: Bulgheroni Antonio 

7:Description: Simple dataclasses that hold editable steering metadata without execution logic. 

8""" 

9 

10from __future__ import annotations 

11 

12from abc import ABC 

13from dataclasses import dataclass, field 

14from enum import StrEnum 

15from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast 

16 

17from mafw.db.db_filter import ExprParser 

18 

19if TYPE_CHECKING: 

20 from mafw.db.db_filter import ExprNode 

21 

22from mafw.enumerators import LogicalOp 

23 

24 

25class ParameterSource(StrEnum): 

26 """Origin of a processor parameter value in the GUI pipeline.""" 

27 

28 CONFIG = 'config' 

29 INHERITED = 'inherited' 

30 DEFAULT = 'default' 

31 

32 

33class ParameterSchemaStatus(StrEnum): 

34 """Schema reconciliation status for a processor parameter.""" 

35 

36 OK = 'OK' 

37 DEPRECATED = 'Deprecated' 

38 NEW = 'New' 

39 UNKNOWN = 'Unknown' 

40 

41 

42class ProcessorSchemaStatus(StrEnum): 

43 """Schema reconciliation status for a processor.""" 

44 

45 OK = 'OK' 

46 UNKNOWN = 'Unknown' 

47 

48 

49class FilterKind(StrEnum): 

50 """The kind of filter configuration.""" 

51 

52 MODEL = 'model' 

53 FIELD = 'field' 

54 CONDITIONAL = 'conditional' 

55 

56 

57@dataclass 

58class Condition: 

59 """A single filter condition.""" 

60 

61 operator: str | LogicalOp 

62 value: Any 

63 is_implicit: bool = True 

64 is_valid: bool = True 

65 

66 

67@dataclass 

68class FieldInfo: 

69 """Metadata about a model field.""" 

70 

71 name: str 

72 type: type 

73 help_text: Optional[str] = None 

74 

75 

76@dataclass 

77class ModelInfo: 

78 """Metadata about a database model.""" 

79 

80 field_info: Dict[str, FieldInfo] = field(default_factory=dict) 

81 

82 

83@dataclass 

84class FilterConfig(ABC): 

85 """Base configuration for a filter.""" 

86 

87 name: str 

88 kind: FilterKind = field(init=False) 

89 enabled: bool = True 

90 model: str = '' # Common field for all filters 

91 

92 

93@dataclass 

94class ModelFilterConfig(FilterConfig): 

95 """ 

96 Filter applied to a specific database model. 

97 

98 Each field condition is AND'ed by default unless a custom 

99 logical expression is provided via `logic`. 

100 """ 

101 

102 conditions: Dict[str, Condition] = field(default_factory=dict) 

103 logic_str_original: Optional[str] = None 

104 logic_buffer: Optional[str] = None 

105 logic_ast: Optional[ExprNode] = None 

106 logic_dirty: bool = False 

107 logic_is_valid: bool = True 

108 model_info: Optional[ModelInfo] = field(default=None, init=False) 

109 

110 @property 

111 def logic(self) -> Optional[str]: 

112 """Backward compatibility for logic string.""" 

113 return self.logic_str_original 

114 

115 @logic.setter 

116 def logic(self, value: Optional[str]) -> None: 

117 self.logic_str_original = value 

118 if value is not None: 

119 try: 

120 self.logic_ast = ExprParser(value).parse() 

121 except Exception: 

122 self.logic_ast = None 

123 else: 

124 self.logic_ast = None 

125 self.logic_dirty = True 

126 

127 def __post_init__(self) -> None: 

128 self.kind = FilterKind.MODEL 

129 self.fill_model_info() 

130 

131 def fill_model_info(self) -> None: 

132 """Populate model_info by inspecting the database model.""" 

133 from mafw.db.db_model import mafw_model_register 

134 

135 if not self.model: 

136 return 

137 

138 try: 

139 model_class = mafw_model_register.get_model(self.model) 

140 except KeyError: 

141 return 

142 

143 self.model_info = ModelInfo() 

144 for field_obj in model_class._meta.sorted_fields: # type: ignore[attr-defined] 

145 f_type = _field_type_from_peewee(field_obj) 

146 f_help = getattr(field_obj, 'help_text', None) 

147 self.model_info.field_info[field_obj.name] = FieldInfo(name=field_obj.name, type=f_type, help_text=f_help) 

148 

149 

150@dataclass 

151class FieldFilterConfig(FilterConfig): 

152 """ 

153 Filter applied to a single field of a specific model. 

154 

155 Conditions are AND'ed by default unless a custom logical 

156 expression is defined via `logic`. 

157 """ 

158 

159 field_name: str = '' 

160 conditions: Dict[str, Condition] = field(default_factory=dict) 

161 logic_str_original: Optional[str] = None 

162 logic_buffer: Optional[str] = None 

163 logic_ast: Optional[ExprNode] = None 

164 logic_dirty: bool = False 

165 logic_is_valid: bool = True 

166 

167 @property 

168 def logic(self) -> Optional[str]: 

169 """Backward compatibility for logic string.""" 

170 return self.logic_str_original 

171 

172 @logic.setter 

173 def logic(self, value: Optional[str]) -> None: 

174 self.logic_str_original = value 

175 if value is not None: 

176 try: 

177 self.logic_ast = ExprParser(value).parse() 

178 except Exception: 

179 self.logic_ast = None 

180 else: 

181 self.logic_ast = None 

182 self.logic_dirty = True 

183 

184 def __post_init__(self) -> None: 

185 self.kind = FilterKind.FIELD 

186 

187 

188@dataclass 

189class ConditionalFilterConfig(FilterConfig): 

190 """ 

191 Conditional filter applied to a model. 

192 

193 Semantics: 

194 IF condition THEN then_clause ELSE else_clause 

195 """ 

196 

197 condition: Optional[Condition] = None 

198 then_clause: Optional[Condition] = None 

199 else_clause: Optional[Condition] = None 

200 condition_field: str = '' 

201 then_field: str = '' 

202 else_field: Optional[str] = None 

203 

204 auto_named: bool = False 

205 model_info: Optional[ModelInfo] = field(default=None, init=False) 

206 

207 def __post_init__(self) -> None: 

208 self.kind = FilterKind.CONDITIONAL 

209 self.fill_model_info() 

210 

211 def is_complete(self) -> bool: 

212 """Return whether every active clause of the conditional filter is valid.""" 

213 

214 if not self.condition or not self.condition.is_valid: 

215 return False 

216 if not self.then_clause or not self.then_clause.is_valid: 

217 return False 

218 if self.else_clause and not self.else_clause.is_valid: 

219 return False 

220 return True 

221 

222 def fill_model_info(self) -> None: 

223 """Populate model_info by inspecting the database model.""" 

224 from mafw.db.db_model import mafw_model_register 

225 

226 if not self.model: 

227 return 

228 

229 try: 

230 model_class = mafw_model_register.get_model(self.model) 

231 except KeyError: 

232 return 

233 

234 self.model_info = ModelInfo() 

235 for field_obj in model_class._meta.sorted_fields: # type: ignore[attr-defined] 

236 f_type = _field_type_from_peewee(field_obj) 

237 f_help = getattr(field_obj, 'help_text', None) 

238 self.model_info.field_info[field_obj.name] = FieldInfo(name=field_obj.name, type=f_type, help_text=f_help) 

239 

240 

241@dataclass 

242class ParameterConfig: 

243 """Configuration and status for a single processor parameter. 

244 

245 :param name: The name of the parameter. 

246 :param value: The current value of the parameter. 

247 :param default: The default value of the parameter. 

248 :param source: The source of the parameter value. 

249 :param status: The schema reconciliation status of the parameter. 

250 :param help: Documentation text for the parameter. 

251 :param type: The type hint for the parameter. 

252 """ 

253 

254 name: str 

255 value: Any = None 

256 default: Any = None 

257 source: ParameterSource = ParameterSource.CONFIG 

258 status: ParameterSchemaStatus = ParameterSchemaStatus.OK 

259 help: str | None = None 

260 active_override: bool = False 

261 type: Any = None 

262 

263 

264@dataclass 

265class ProcessorRef: 

266 """Reference to a processor entry used in processors_to_run lists. 

267 

268 :param base_name: The canonical name of the processor without replica suffix. 

269 :param replica: Optional replica identifier appended to the base name with ``#``. 

270 """ 

271 

272 base_name: str 

273 replica: Optional[str] = None 

274 

275 @property 

276 def full_name(self) -> str: 

277 """Return the replica-aware full identifier for this reference.""" 

278 return f'{self.base_name}#{self.replica}' if self.replica else self.base_name 

279 

280 

281@dataclass 

282class ProcessorConfig: 

283 """Configuration store for an individual processor or replica section. 

284 

285 :param name: Replica-aware section name (``Processor`` or ``Processor#Replica``). 

286 :param parameters: Parameter configurations and status. 

287 :param processor_status: Schema reconciliation status for the processor. 

288 :param filters: Mapping from model name to list of filter configuration objects. 

289 :param logic_str_original: Optional ``__logic__`` string stored in the per-filter table. 

290 :param new_only: Optional ``__new_only__`` override. 

291 :param inheritance: Optional ``__inheritance__`` flag for replicas. 

292 :param has_filter_root: Whether the filter table includes a root entry. 

293 """ 

294 

295 name: str 

296 parameters: Dict[str, ParameterConfig] = field(default_factory=dict) 

297 processor_status: ProcessorSchemaStatus = ProcessorSchemaStatus.OK 

298 filters: Dict[str, List[FilterConfig]] = field(default_factory=dict) 

299 logic_str_original: Optional[str] = None 

300 logic_buffer: Optional[str] = None 

301 logic_ast: Optional[ExprNode] = None 

302 logic_dirty: bool = False 

303 logic_is_valid: bool = True 

304 new_only: Optional[bool] = None 

305 inheritance: Optional[bool] = None 

306 has_filter_root: bool = True 

307 schema_filter_models: List[str] = field(default_factory=list) 

308 

309 @property 

310 def filter_logic(self) -> Optional[str]: 

311 """Backward compatibility for filter_logic string.""" 

312 return self.logic_str_original 

313 

314 @filter_logic.setter 

315 def filter_logic(self, value: Optional[str]) -> None: 

316 self.logic_str_original = value 

317 if value is not None: 

318 try: 

319 self.logic_ast = ExprParser(value).parse() 

320 except Exception: 

321 self.logic_ast = None 

322 else: 

323 self.logic_ast = None 

324 self.logic_dirty = True 

325 

326 

327@dataclass 

328class GroupConfig: 

329 """Configuration for a logical processor group section. 

330 

331 :param name: Section name used to reference the group in ``processors_to_run``. 

332 :param processors: Ordered list of processor names and/or nested group names. 

333 :param description: Optional documentation string for the group. 

334 :param attributes: Additional arbitrary attributes for the group. 

335 """ 

336 

337 name: str 

338 processors: List[str] 

339 description: Optional[str] = None 

340 attributes: Dict[str, Any] = field(default_factory=dict) 

341 

342 

343@dataclass 

344class GlobalSettings: 

345 """Top-level globals that appear outside any processor or group table. 

346 

347 :param processors_to_run: Ordered list of processor or group names to execute. 

348 :param analysis_name: Optional name for the analysis. 

349 :param analysis_description: Optional description for the analysis. 

350 :param new_only: Optional flag to process only new data. 

351 :param create_standard_tables: Optional flag to create standard output tables. 

352 """ 

353 

354 processors_to_run: List[str] = field(default_factory=list) 

355 analysis_name: Optional[str] = None 

356 analysis_description: Optional[str] = None 

357 new_only: Optional[bool] = None 

358 create_standard_tables: Optional[bool] = None 

359 

360 

361@dataclass 

362class DBConfiguration: 

363 """Database section content mirrored from the steering file. 

364 

365 :param url: Optional database URL connection string. 

366 :param pragmas: SQLite pragma settings to apply on connection. 

367 :param attributes: Additional arbitrary attributes for the database configuration. 

368 :param enabled: Whether ``DBConfiguration`` should be serialized. 

369 """ 

370 

371 url: Optional[str] = None 

372 pragmas: Dict[str, Any] = field(default_factory=dict) 

373 attributes: Dict[str, Any] = field(default_factory=dict) 

374 enabled: bool = True 

375 

376 

377@dataclass 

378class UIConfiguration: 

379 """User-interface section content mirrored from the steering file. 

380 

381 :param interface: The interface type to use (e.g., ``rich``). 

382 """ 

383 

384 interface: str = 'rich' 

385 

386 

387_FIELD_TYPE_STRINGS: dict[str, type] = { 

388 'INT': int, 

389 'INTEGER': int, 

390 'BIGINT': int, 

391 'SMALLINT': int, 

392 'TINYINT': int, 

393 'FLOAT': float, 

394 'REAL': float, 

395 'DOUBLE': float, 

396 'DECIMAL': float, 

397 'NUMERIC': float, 

398 'CHAR': str, 

399 'VARCHAR': str, 

400 'TEXT': str, 

401 'STRING': str, 

402 'DATE': str, 

403 'DATETIME': str, 

404 'TIME': str, 

405 'BOOLEAN': bool, 

406 'BOOL': bool, 

407 'AUTO': int, 

408} 

409"""Mapping from Peewee field_type strings to Python primitives.""" 

410 

411 

412def _normalize_field_type(field_type: Any) -> type: 

413 """Normalize a field descriptor to a supported Python primitive.""" 

414 

415 if field_type in (int, float, str, bool): 

416 return cast(type, field_type) 

417 if isinstance(field_type, str): 

418 return _FIELD_TYPE_STRINGS.get(field_type.upper(), str) 

419 return str 

420 

421 

422def _field_type_from_peewee(field_obj: Any) -> type: 

423 """Extract a primitive type from the given Peewee field definition.""" 

424 

425 for attr in ('field_type', 'data_type'): 

426 raw_type = getattr(field_obj, attr, None) 

427 if raw_type is not None: 

428 return _normalize_field_type(raw_type) 

429 return str