Coverage for src / c41811 / config / processor / component.py: 100%

1# cython: language_level = 3 # noqa: ERA001 

2 

3 

4""" 

5组件配置处理器 

6 

7.. versionadded:: 0.2.0 

8""" 

9 

10import os 

11from collections.abc import Callable 

12from collections.abc import Mapping 

13from copy import deepcopy 

14from typing import Any 

15from typing import Literal 

16from typing import override 

17 

18from ..abc import ABCConfigFile 

19from ..abc import ABCConfigPool 

20from ..abc import ABCMetaParser 

21from ..abc import ABCSLProcessorPool 

22from ..basic.component import ComponentConfigData 

23from ..basic.component import ComponentMember 

24from ..basic.component import ComponentMeta 

25from ..basic.component import ComponentOrders 

26from ..basic.core import ConfigFile 

27from ..basic.mapping import MappingConfigData 

28from ..basic.object import NoneConfigData 

29from ..basic.sequence import SequenceConfigData 

30from ..errors import ComponentMetadataException 

31from ..main import BasicChainConfigSL 

32from ..main import RequiredPath 

33from ..utils import Ref 

34from ..validators import ValidatorOptions 

35 

36 

37class ComponentMetaParser[D: MappingConfigData[Any]](ABCMetaParser[D, ComponentMeta[D]]): 

38 """默认元信息解析器""" 

39 

40 _validator: RequiredPath[dict[str, Any], D] = RequiredPath( 

41 { 

42 "members": list[str | ComponentMember], 

43 "order": list[str], 

44 "orders": dict[Literal["create", "read", "update", "delete"], list[str]], 

45 }, 

46 static_config=ValidatorOptions(allow_modify=True, skip_missing=True), 

47 ) 

48 

49 @override 

50 def convert_config2meta(self, meta_config: D) -> ComponentMeta[D]: 

51 """ 

52 解析元配置 

53 

54 :param meta_config: 元配置 

55 :type meta_config: D 

56 

57 :return: 元数据 

58 :rtype: ComponentMeta[D] 

59 """ 

60 meta = self._validator.filter(Ref(meta_config)) 

61 

62 members = meta.get("members", SequenceConfigData()).data 

63 for i, member in enumerate(members): 

64 if isinstance(member, str): 

65 members[i] = ComponentMember(member) 

66 elif isinstance(member, dict): 

67 members[i] = ComponentMember(**member) 

68 

69 orders: ComponentOrders = ComponentOrders(**meta.get("orders", MappingConfigData()).data) 

70 order = meta.setdefault("order", [member.alias if member.alias else member.filename for member in members]) 

71 if not isinstance(order, list): 

72 order = order.data 

73 for name in order: 

74 # noinspection PyUnresolvedReferences 

75 for attr in orders.__dataclass_fields__: 

76 if name in getattr(orders, attr): 

77 continue 

78 getattr(orders, attr).append(name) 

79 

80 # noinspection PyUnresolvedReferences 

81 for attr in orders.__dataclass_fields__: 

82 o = getattr(orders, attr) 

83 if len(set(o)) != len(o): 

84 msg = f"name(s) repeated in {attr} order" 

85 raise ComponentMetadataException(msg) 

86 

87 return ComponentMeta(meta, orders, members, self) 

88 

89 @override 

90 def convert_meta2config(self, meta: ComponentMeta[D]) -> D: 

91 """ 

92 解析元数据 

93 

94 :param meta: 元数据 

95 :type meta: ComponentMeta[D] 

96 

97 :return: 元配置 

98 :rtype: D 

99 """ 

100 return meta.config 

101 

102 @override 

103 def validator(self, meta: ComponentMeta[D], *args: Any) -> ComponentMeta[D]: 

104 return self.convert_config2meta(meta.config) 

105 

106 @override 

107 def __eq__(self, other: Any) -> bool: 

108 if not isinstance(other, type(self)): 

109 return NotImplemented 

110 return self._validator == other._validator 

111 

112 __hash__ = None # type: ignore[assignment] 

113 

114 

115def _component_loader_kwargs_builder(kwargs: dict[str, Any]) -> Callable[[ComponentMember | None], dict[str, Any]]: 

116 # noinspection GrazieInspection 

117 """ 

118 构建组件加载参数 

119 

120 :param kwargs: 参数 

121 :type kwargs: dict[str, Any] 

122 :return: 构建器 

123 :rtype: Callable[[ComponentMember | None], dict[str, Any]] 

124 

125 .. versionadded:: 0.3.0 

126 """ 

127 format_mapping: Mapping[str | None, Any] | None = ( 

128 fmt if isinstance((fmt := kwargs.get("config_formats")), Mapping) else None 

129 ) 

130 

131 def builder(member: ComponentMember | None) -> dict[str, Any]: 

132 new_kwargs = deepcopy(kwargs) # 防止参数里有可变对象被load修改 

133 if format_mapping is not None: 

134 if member is None and None in format_mapping: 

135 new_kwargs["config_formats"] = format_mapping[None] 

136 elif isinstance(member, ComponentMember) and member.alias is not None and member.alias in format_mapping: 

137 new_kwargs["config_formats"] = format_mapping[member.alias] 

138 elif ( 

139 isinstance(member, ComponentMember) 

140 and member.filename is not None 

141 and member.filename in format_mapping 

142 ): 

143 new_kwargs["config_formats"] = format_mapping[member.filename] 

144 else: 

145 # 更符合语义 即此项不存在视为load未提供config_formats参数 

146 # 虽然这种细微的语义差异只有可能在极少数情况下影响自定义子类 

147 del new_kwargs["config_formats"] 

148 if isinstance(member, ComponentMember) and member.config_format is not None: 

149 # noinspection PyUnreachableCode 

150 match config_formats := new_kwargs.get("config_formats", None): 

151 case str(): 

152 config_formats = [config_formats, member.config_format] 

153 case None: 

154 config_formats = member.config_format 

155 case _: # 处理 Iterable[str] 顺手报错 

156 config_formats = list(config_formats) 

157 if member.config_format not in config_formats: 

158 config_formats.append(member.config_format) 

159 new_kwargs["config_formats"] = config_formats 

160 return new_kwargs 

161 

162 return builder 

163 

164 

165class ComponentSL(BasicChainConfigSL): 

166 """组件模式配置处理器""" 

167 

168 def __init__( 

169 self, 

170 *, 

171 reg_alias: str | None = None, 

172 create_dir: bool = True, 

173 meta_parser: ABCMetaParser[Any, ComponentMeta[Any]] | None = None, 

174 meta_file: str = "__meta__", 

175 ): 

176 """ 

177 :param reg_alias: 处理器别名 

178 :type reg_alias: str | None 

179 :param create_dir: 是否创建目录 

180 :type create_dir: bool 

181 :param meta_parser: 元数据解析器 

182 :type meta_parser: ABCMetaParser[Any, ComponentMeta[Any]] | None 

183 :param meta_file: 元信息文件名 

184 :type meta_file: str 

185 

186 .. versionchanged:: 0.3.0 

187 重构属性 ``initial_file`` 为参数 ``meta_file`` 并更改默认值 ``__init__`` 为 ``__meta__`` 

188 """ # noqa: D205 

189 super().__init__(reg_alias=reg_alias, create_dir=create_dir) 

190 

191 if meta_parser is None: 

192 meta_parser = ComponentMetaParser() 

193 

194 self.meta_parser: ABCMetaParser[Any, ComponentMeta[Any]] = meta_parser 

195 self.meta_file = meta_file 

196 

197 @property 

198 @override 

199 def processor_reg_name(self) -> str: 

200 return "component" 

201 

202 @property 

203 @override 

204 def supported_file_patterns(self) -> tuple[str, ...]: 

205 return ".component", ".comp" 

206 

207 @override 

208 def namespace_formatter(self, namespace: str, file_name: str) -> str: 

209 return os.path.normpath(os.path.join(namespace, self.filename_formatter(file_name))) 

210 

211 supported_file_classes = [ConfigFile] # noqa: RUF012 

212 

213 @override 

214 def save_file( 

215 self, 

216 config_pool: ABCConfigPool, 

217 config_file: ABCConfigFile[ComponentConfigData[Any, Any] | NoneConfigData], 

218 namespace: str, 

219 file_name: str, 

220 *args: Any, 

221 **kwargs: Any, 

222 ) -> None: 

223 config_data = config_file.config 

224 if isinstance(config_data, NoneConfigData): 

225 config_data = ComponentConfigData() 

226 elif not isinstance(config_data, ComponentConfigData): 

227 with self.raises(TypeError): 

228 msg = f"{namespace} is not a ComponentConfigData" 

229 raise TypeError(msg) 

230 

231 meta_config = self.meta_parser.convert_meta2config(config_data.meta) 

232 file_name, file_ext = os.path.splitext(file_name) 

233 super().save_file(config_pool, ConfigFile(meta_config), namespace, self.meta_file + file_ext, *args, **kwargs) 

234 

235 for member in config_data.meta.members: 

236 super().save_file( 

237 config_pool, 

238 ConfigFile(config_data[member.filename], config_format=member.config_format), 

239 namespace, 

240 member.filename, 

241 *args, 

242 **kwargs, 

243 ) 

244 

245 @override 

246 def load_file( 

247 self, config_pool: ABCConfigPool, namespace: str, file_name: str, *args: Any, **kwargs: Any 

248 ) -> ConfigFile[ComponentConfigData[Any, Any]]: 

249 # noinspection PyIncorrectDocstring 

250 """ 

251 加载指定命名空间的配置 

252 

253 :param config_pool: 配置池 

254 :type config_pool: ABCConfigPool 

255 :param namespace: 命名空间 

256 :type namespace: str 

257 :param file_name: 文件名 

258 :type file_name: str 

259 

260 可选参数 

261 ----------- 

262 :param config_formats: 指定成员配置格式 

263 :type config_formats: Mapping[str | None, Any] 

264 

265 :return: 配置文件 

266 :rtype: ConfigFile[ComponentConfigData[Any, Any]] 

267 

268 .. caution:: 

269 传递SL处理前没有清理已经缓存在配置池里的配置文件，返回的可能不是最新数据 

270 

271 .. versionchanged:: 0.3.0 

272 新增可选参数 ``config_formats`` 以支持指定成员的配置解析格式 

273 """ # noqa: RUF002 

274 file_name, file_ext = os.path.splitext(file_name) 

275 kwargs_builder = _component_loader_kwargs_builder(kwargs) 

276 

277 initial_file = super().load_file( 

278 config_pool, namespace, self.meta_file + file_ext, *args, **kwargs_builder(None) 

279 ) 

280 initial_data = initial_file.config 

281 

282 if not isinstance(initial_data, MappingConfigData): 

283 with self.raises(TypeError): 

284 msg = f"{namespace} is not a MappingConfigData" 

285 raise TypeError(msg) 

286 

287 meta = self.meta_parser.convert_config2meta(initial_data) 

288 members = {} 

289 for member in meta.members: 

290 members[member.filename] = ( 

291 super().load_file(config_pool, namespace, member.filename, *args, **kwargs_builder(member)).config 

292 ) 

293 

294 return ConfigFile(ComponentConfigData(meta, members), config_format=self.reg_name) 

295 

296 @override 

297 def initialize( 

298 self, 

299 processor_pool: ABCSLProcessorPool, 

300 root_path: str, 

301 namespace: str, 

302 file_name: str, 

303 *args: Any, 

304 **kwargs: Any, 

305 ) -> ConfigFile[ComponentConfigData[Any, Any]]: 

306 return ConfigFile(ComponentConfigData(ComponentMeta(parser=self.meta_parser)), config_format=self.reg_name) 

307 

308 

309__all__ = ( 

310 "ComponentMetaParser", 

311 "ComponentSL", 

312)