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

1"""Validate values according to specs""" 

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 new Runtime object by generating or loading the specs. 

22  

23 Args: 

24 manifest_path: Optional path to the application manifest. If not provided, the generator will try to find it automatically. 

25 read_cache: If True, try to read the specs from a cache file.  

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

27 """ 

28 

29 sd = _get_specs(manifest_path, read_cache, write_cache) 

30 return Object(sd) 

31 

32 

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

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

35 

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

37 try: 

38 specs = generator_main.from_path(cache_path) 

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

40 return specs 

41 except gws.lib.jsonx.Error: 

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

43 

44 if manifest_path: 

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

46 

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

48 specs = generator_main.generate(manifest_path=manifest_path) 

49 gws.debug.time_end() 

50 

51 if write_cache: 

52 try: 

53 generator_main.to_path(cache_path, specs) 

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

55 except gws.lib.jsonx.Error: 

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

57 

58 return specs 

59 

60 

61## 

62 

63 

64class Object(gws.SpecRuntime): 

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

66 self.sd = sd 

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

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

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

70 

71 self.serverTypes = sd.serverTypes 

72 self.serverTypesDict = {} 

73 self.rawCommands = set() 

74 

75 for typ in self.serverTypes: 

76 self.serverTypesDict[typ.uid] = typ 

77 if typ.extName: 

78 self.serverTypesDict[typ.extName] = typ 

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

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

81 

82 self.strings = sd.strings 

83 self.chunks = sd.chunks 

84 

85 self.appBundlePaths = [] 

86 for chunk in self.chunks: 

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

88 if path not in self.appBundlePaths: 

89 self.appBundlePaths.append(path) 

90 

91 self._descCache = {} 

92 

93 def __getstate__(self): 

94 self._descCache = {} 

95 return vars(self) 

96 

97 def get_type(self, key): 

98 return self.serverTypesDict.get(key) 

99 

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

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

102 return r.read(value, type_name) 

103 

104 def object_descriptor(self, name): 

105 if name in self._descCache: 

106 return self._descCache[name] 

107 

108 typ = self.get_type(name) 

109 if not typ: 

110 return 

111 

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

113 extName=typ.extName, 

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

115 ident=typ.ident, 

116 modName=typ.modName, 

117 modPath=typ.modPath, 

118 classPtr=None, 

119 ) 

120 

121 return self._descCache[name] 

122 

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

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

125 if not ext_name: 

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

127 ext_name += '.' + ext_type 

128 setattr(cls, 'extName', ext_name) 

129 setattr(cls, 'extType', ext_type) 

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

131 extName=ext_name, 

132 extType=ext_type, 

133 ident=cls.__name__, 

134 modName='', 

135 modPath='', 

136 classPtr=cls, 

137 ) 

138 

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

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

141 if cls: 

142 return cls 

143 

144 desc = None 

145 if real_name: 

146 desc = self.object_descriptor(real_name) 

147 elif ext_name: 

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

149 if not desc: 

150 return 

151 

152 if not desc.classPtr: 

153 try: 

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

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

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

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

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

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

160 

161 return desc.classPtr 

162 

163 def command_descriptor(self, command_category, command_name): 

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

165 command_category = gws.CommandCategory.raw 

166 

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

168 

169 if name in self._descCache: 

170 return self._descCache[name] 

171 

172 typ = self.get_type(name) 

173 

174 if not typ: 

175 return 

176 

177 return gws.ExtCommandDescriptor( 

178 extName=typ.extName, 

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

180 extCommandCategory=command_category, 

181 tArg=typ.tArg, 

182 tOwner=typ.tOwner, 

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

184 methodName=typ.ident, 

185 ) 

186 

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

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

189 cmds = [] 

190 

191 for typ in self.serverTypes: 

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

193 continue 

194 

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

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

197 if not m: 

198 continue 

199 cmd1 = m.group(1) 

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

201 

202 args = [] 

203 arg_typ = self.get_type(typ.tArg) 

204 if arg_typ: 

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

206 prop_typ = self.get_type(prop_type_uid) 

207 if not prop_typ: 

208 continue 

209 args.append( 

210 gws.Data( 

211 name=name, 

212 type=prop_typ.tValue, 

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

214 defaultValue=prop_typ.defaultValue, 

215 hasDefault=prop_typ.hasDefault, 

216 ) 

217 ) 

218 

219 entry = gws.Data( 

220 cmd1=cmd1, 

221 cmd2=cmd2, 

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

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

224 ) 

225 cmds.append(entry) 

226 

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

228 

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

230 ext_name = gws.ext.name_for(classref) 

231 if ext_name: 

232 return None, '', ext_name 

233 

234 if isinstance(classref, str): 

235 return None, classref, '' 

236 

237 if isinstance(classref, type): 

238 return classref, '', '' 

239 

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

241 

242 def get_config_types(self, lang): 

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

244 ts = [] 

245 

246 for typ in self.serverTypes: 

247 if not typ.isConfig: 

248 continue 

249 d = dict(vars(typ)) 

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

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

252 if 'enumDocs' in d: 

253 for k in d['enumDocs']: 

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

255 if key in strs: 

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

257 ts.append(d) 

258 

259 return ts