Coverage for gws-app/gws/spec/runtime.py: 64%

1"""Spec runtime: loads the specs and implements ``gws.SpecRuntime``.""" 

2 

3from typing import Optional 

4 

5import re 

6 

7import gws 

8import gws.lib.jsonx 

9import gws.lib.dynimport 

10 

11from . import core, reader 

12from .generator import main as generator_main 

13 

14Error = core.Error 

15ReadError = core.ReadError 

16GeneratorError = core.GeneratorError 

17LoadError = core.LoadError 

18 

19 

20def create(manifest_path: str = '', read_cache=False, write_cache=False) -> 'Object': 

21 """Create a spec runtime object by generating or loading the specs. 

22 

23 The cache file is stored in ``gws.c.SPEC_DIR``, one file per manifest path. 

24 

25 Args: 

26 manifest_path: Path to the application manifest. Without a manifest, only the built-in plugins are used. 

27 read_cache: If True, load the specs from the cache file if it exists. 

28 write_cache: If True, write the generated specs to the cache file. 

29 

30 Returns: 

31 The spec runtime object. 

32 

33 Raises: 

34 ``GeneratorError``: If the specs cannot be generated. 

35 """ 

36 

37 sd = _get_specs(manifest_path, read_cache, write_cache) 

38 return Object(sd) 

39 

40 

41def _get_specs(manifest_path: str = '', read_cache=False, write_cache=False) -> core.SpecData: 

42 cache_path = gws.u.ensure_dir(gws.c.SPEC_DIR) + '/spec_' + gws.u.to_uid(manifest_path or '') + '.json' 

43 

44 if read_cache and gws.u.is_file(cache_path): 

45 try: 

46 specs = generator_main.from_path(cache_path) 

47 gws.log.debug(f'spec.create: loaded from {cache_path!r}') 

48 return specs 

49 except gws.lib.jsonx.Error: 

50 gws.log.exception(f'spec.create: load failed') 

51 

52 if manifest_path: 

53 gws.log.debug(f'spec.create: using manifest {manifest_path!r}...') 

54 

55 gws.debug.time_start('SPEC GENERATOR') 

56 specs = generator_main.generate(manifest_path=manifest_path) 

57 gws.debug.time_end() 

58 

59 if write_cache: 

60 try: 

61 generator_main.to_path(cache_path, specs) 

62 gws.log.debug(f'spec.create: stored to {cache_path!r}') 

63 except gws.lib.jsonx.Error: 

64 gws.log.exception(f'spec.create: store failed') 

65 

66 return specs 

67 

68 

69## 

70 

71 

72class Object(gws.SpecRuntime): 

73 """Spec runtime object.""" 

74 

75 def __init__(self, sd: core.SpecData): 

76 """Create the runtime from spec data. 

77 

78 Args: 

79 sd: Spec data produced by the generator or loaded from a cache file. 

80 """ 

81 

82 self.sd = sd 

83 self.manifest = gws.ApplicationManifest(sd.meta['manifest']) 

84 self.manifestPath = sd.meta['manifestPath'] 

85 self.version = sd.meta['version'] 

86 

87 self.serverTypes = sd.serverTypes 

88 self.serverTypesDict = {} 

89 self.rawCommands = set() 

90 

91 for typ in self.serverTypes: 

92 self.serverTypesDict[typ.uid] = typ 

93 if typ.extName: 

94 self.serverTypesDict[typ.extName] = typ 

95 if typ.extName.startswith(core.v.EXT_COMMAND_PREFIX + str(gws.CommandCategory.raw) + '.'): 

96 self.rawCommands.add(typ.extName.split('.')[-1]) 

97 

98 self.strings = sd.strings 

99 self.chunks = sd.chunks 

100 

101 self.appBundlePaths = [] 

102 for chunk in self.chunks: 

103 path = chunk.bundleDir + '/' + gws.c.JS_BUNDLE 

104 if path not in self.appBundlePaths: 

105 self.appBundlePaths.append(path) 

106 

107 self._descCache = {} 

108 

109 def __getstate__(self): 

110 """Clear the descriptor cache before pickling.""" 

111 

112 self._descCache = {} 

113 return vars(self) 

114 

115 def get_type(self, key): 

116 """Get a type by uid or extension name. 

117 

118 Args: 

119 key: Type uid or extension name. 

120 

121 Returns: 

122 The type or ``None`` if not found. 

123 """ 

124 

125 return self.serverTypesDict.get(key) 

126 

127 def read(self, value, type_name, path='', options=None): 

128 r = reader.Reader(self, path, options) 

129 return r.read(value, type_name) 

130 

131 def object_descriptor(self, name): 

132 if name in self._descCache: 

133 return self._descCache[name] 

134 

135 typ = self.get_type(name) 

136 if not typ: 

137 return 

138 

139 self._descCache[name] = gws.ExtObjectDescriptor( 

140 extName=typ.extName, 

141 extType=typ.extName.split('.').pop(), 

142 ident=typ.ident, 

143 modName=typ.modName, 

144 modPath=typ.modPath, 

145 classPtr=None, 

146 ) 

147 

148 return self._descCache[name] 

149 

150 def register_object(self, classref, ext_type, cls): 

151 _, _, ext_name = self.parse_classref(classref) 

152 if not ext_name: 

153 raise Error(f'invalid class reference {classref!r}') 

154 ext_name += '.' + ext_type 

155 setattr(cls, 'extName', ext_name) 

156 setattr(cls, 'extType', ext_type) 

157 self._descCache[ext_name] = gws.ExtObjectDescriptor( 

158 extName=ext_name, 

159 extType=ext_type, 

160 ident=cls.__name__, 

161 modName='', 

162 modPath='', 

163 classPtr=cls, 

164 ) 

165 

166 def get_class(self, classref, ext_type=None): 

167 cls, real_name, ext_name = self.parse_classref(classref) 

168 if cls: 

169 return cls 

170 

171 desc = None 

172 if real_name: 

173 desc = self.object_descriptor(real_name) 

174 elif ext_name: 

175 desc = self.object_descriptor(ext_name + '.' + (ext_type or core.v.DEFAULT_VARIANT_TAG)) 

176 if not desc: 

177 return 

178 

179 if not desc.classPtr: 

180 try: 

181 mod = gws.lib.dynimport.import_from_path(desc.modPath, gws.c.APP_DIR) 

182 except gws.lib.dynimport.Error as exc: 

183 raise LoadError(f'cannot load class {classref!r} from {desc.modPath!r}') from exc 

184 desc.classPtr = getattr(mod, desc.ident) 

185 setattr(desc.classPtr, 'extName', desc.extName) 

186 setattr(desc.classPtr, 'extType', desc.extType) 

187 

188 return desc.classPtr 

189 

190 def command_descriptor(self, command_category, command_name): 

191 if command_name in self.rawCommands and command_category != gws.CommandCategory.raw: 

192 command_category = gws.CommandCategory.raw 

193 

194 name = core.v.EXT_COMMAND_PREFIX + str(command_category) + '.' + command_name 

195 

196 if name in self._descCache: 

197 return self._descCache[name] 

198 

199 typ = self.get_type(name) 

200 

201 if not typ: 

202 return 

203 

204 return gws.ExtCommandDescriptor( 

205 extName=typ.extName, 

206 extType=typ.extName.split('.').pop(), 

207 extCommandCategory=command_category, 

208 tArg=typ.tArg, 

209 tOwner=typ.tOwner, 

210 owner=self.object_descriptor(typ.tOwner), 

211 methodName=typ.ident, 

212 ) 

213 

214 def cli_commands(self, lang='en'): 

215 """List the CLI commands with their arguments and docs. 

216 

217 A command name like ``gws.ext.command.cli.serverStart`` is split into 

218 ``cmd1='server'`` and ``cmd2='start'``. 

219 

220 Args: 

221 lang: Language code for the docs, falls back to English. 

222 

223 Returns: 

224 A list of ``gws.Data`` objects with ``cmd1``, ``cmd2``, ``doc`` and ``args``, 

225 sorted by command name. Each argument has ``name``, ``type``, ``doc``, 

226 ``defaultValue`` and ``hasDefault``. 

227 """ 

228 

229 strings = self.strings.get(lang) or self.strings['en'] 

230 cmds = [] 

231 

232 for typ in self.serverTypes: 

233 if not typ.extName.startswith(core.v.EXT_COMMAND_CLI_PREFIX): 

234 continue 

235 

236 # e.g "gws.ext.command.cli.serverStart" -> [server, start] 

237 m = re.search(r'\.([a-z]+)(\w+)$', typ.extName) 

238 if not m: 

239 continue 

240 cmd1 = m.group(1) 

241 cmd2 = m.group(2).lower() 

242 

243 args = [] 

244 arg_typ = self.get_type(typ.tArg) 

245 if arg_typ: 

246 for name, prop_type_uid in arg_typ.tProperties.items(): 

247 prop_typ = self.get_type(prop_type_uid) 

248 if not prop_typ: 

249 continue 

250 args.append( 

251 gws.Data( 

252 name=name, 

253 type=prop_typ.tValue, 

254 doc=strings.get(prop_type_uid) or self.strings['en'].get(prop_type_uid) or '', 

255 defaultValue=prop_typ.defaultValue, 

256 hasDefault=prop_typ.hasDefault, 

257 ) 

258 ) 

259 

260 entry = gws.Data( 

261 cmd1=cmd1, 

262 cmd2=cmd2, 

263 doc=strings.get(typ.uid) or self.strings['en'].get(typ.uid) or '', 

264 args=sorted(args, key=lambda a: a.name), 

265 ) 

266 cmds.append(entry) 

267 

268 return sorted(cmds, key=lambda c: (c.cmd1, c.cmd2)) 

269 

270 def parse_classref(self, classref: gws.ClassRef) -> tuple[Optional[type], str, str]: 

271 ext_name = gws.ext.name_for(classref) 

272 if ext_name: 

273 return None, '', ext_name 

274 

275 if isinstance(classref, str): 

276 return None, classref, '' 

277 

278 if isinstance(classref, type): 

279 return classref, '', '' 

280 

281 raise Error(f'invalid class reference {classref!r}') 

282 

283 def get_config_types(self, lang): 

284 strs = self.strings.get(lang) or self.strings['en'] 

285 ts = [] 

286 

287 for typ in self.serverTypes: 

288 if not typ.isConfig: 

289 continue 

290 d = dict(vars(typ)) 

291 d['doc'] = strs.get(typ.uid) or '' 

292 d['title'] = strs.get(typ.uid + '._title') or '' 

293 if 'enumDocs' in d: 

294 for k in d['enumDocs']: 

295 key = typ.uid + '.' + k 

296 if key in strs: 

297 d['enumDocs'][k] = strs[key] 

298 ts.append(d) 

299 

300 return ts