Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/extensions/loader.py: 0%

1import base64 

2import importlib 

3import logging 

4import sys 

5from hashlib import sha256 

6from importlib.util import module_from_spec, spec_from_loader 

7from pathlib import Path 

8from types import ModuleType 

9from urllib import request 

10from uuid import uuid1 

11 

12from coxbuild.exceptions import CoxbuildSchemaException 

13 

14from . import Extension 

15 

16logger = logging.getLogger("extension-loader") 

17 

18 

19class CoxbuildExtensionException(CoxbuildSchemaException): 

20 """Exception for coxbuild extension.""" 

21 pass 

22 

23 

24def hashed(src: str): 

25 return sha256(src.encode()).hexdigest() 

26 

27 

28def fromModule(module: ModuleType, version: str = "") -> Extension: 

29 """ 

30 Load extension from module. 

31 

32 mod@{hashcode}://{module name}@{version} 

33 

34 :param module: module to load 

35 :param hashcode: hashcode 

36 :param version: version 

37 """ 

38 logger.info("Load extension from module: %s", module.__name__) 

39 

40 uri = f"mod" 

41 

42 uri += f"://{module.__name__}" 

43 

44 rversion = getattr(module, "__version__", "") 

45 if version and rversion != version: 

46 raise CoxbuildExtensionException( 

47 f"Unmatched extension version, expect {version} but got {rversion}.") 

48 if rversion: 

49 uri += f"@{rversion}" 

50 return Extension(uri, module.__name__, module.__doc__ or "", version, "", module) 

51 

52 

53def fromSource(src: str, filename: str = "<string>", hashcode: str = "") -> Extension: 

54 """ 

55 Load extension from source. 

56 

57 src@{hashcode}://{source code in base64} 

58 

59 :param src: source code 

60 :param filename: filename 

61 :param hashcode: hashcode 

62 """ 

63 logger.info("Load extension from source: %s...", src[:20]) 

64 

65 if src == "stdin": 

66 src = sys.stdin.read() 

67 if src == "interactive": 

68 if hashcode: 

69 raise CoxbuildExtensionException( 

70 "Cannot specify hashcode for interactive source.") 

71 import code 

72 

73 id = str(uuid1()) 

74 spec = spec_from_loader(id, loader=None) 

75 mod = module_from_spec(spec) 

76 mod.__dict__["__module__"] = mod 

77 

78 ext = fromModule(mod) 

79 ext.uri = f"src://interactive@{id}" 

80 mod.__dict__["__extension__"] = ext 

81 

82 exec("from coxbuild.schema import *", mod.__dict__) 

83 exec("manager.register(__extension__)", mod.__dict__) 

84 code.interact(banner="Interactive Schema", local=mod.__dict__) 

85 exec("manager.unregister(__extension__)", mod.__dict__) 

86 else: 

87 rhashcode = hashed(src) 

88 if hashcode and rhashcode != hashcode: 

89 raise CoxbuildExtensionException( 

90 f"Unmatched hashcode, expect {hashcode} but got {rhashcode}.") 

91 

92 spec = spec_from_loader(rhashcode, loader=None) 

93 mod = module_from_spec(spec) 

94 

95 srccode = compile(src, filename, "exec") 

96 

97 exec("from coxbuild.schema import *", mod.__dict__) 

98 exec(srccode, mod.__dict__) 

99 

100 ext = fromModule(mod) 

101 ext.uri = f"src@{rhashcode}://{base64.b64encode(src.encode()).decode()}" 

102 ext.hashcode = rhashcode 

103 

104 return ext 

105 

106 

107def fromFile(file: Path, hashcode: str = "") -> Extension: 

108 """ 

109 Load extension from file. 

110 

111 file@{hashcode}://{file path} 

112 

113 :param file: file to load 

114 :param hashcode: hashcode 

115 """ 

116 logger.info("Load extension from file: %s", str(file)) 

117 

118 src = file.read_text() 

119 ext = fromSource(src, str(file), hashcode=hashcode) 

120 

121 ext.name = file.stem 

122 ext.uri = f"file@{ext.hashcode}://{str(file.resolve())}" 

123 return ext 

124 

125 

126def fromUrl(url: str, hashcode: str = "") -> Extension: 

127 """ 

128 Load extension from url. 

129 

130 url@{hashcode}://{url} 

131 

132 :param url: url to load 

133 :param hashcode: hashcode 

134 """ 

135 logger.info("Load extension from url: %s", url) 

136 

137 with request.urlopen(url) as f: 

138 src = f.read().decode("utf-8") 

139 

140 ext = fromSource(src, filename=url, hashcode=hashcode) 

141 ext.uri = f"url@{ext.hashcode}://{url}" 

142 return ext 

143 

144 

145def fromGallery(name: str, version: str = "", hashcode: str = "") -> Extension: 

146 """ 

147 Load extension from gallery. 

148 

149 ext@{hashcode}://{name}@{version} 

150 

151 :param name: name of extension 

152 :param version: version of extension 

153 :param hashcode: hashcode 

154 """ 

155 logger.info("Load extension from gallery: %s", name) 

156 

157 from .gallery import galleries 

158 

159 gals = galleries() 

160 

161 for gal in gals: 

162 ext = gal.load(name, version, hashcode) 

163 if ext: 

164 return ext 

165 

166 raise CoxbuildExtensionException( 

167 f"Failed to load extension from gallery: {name}@{version}") 

168 

169 

170def load(uri: str): 

171 splited = uri.split("://", 1) 

172 if len(splited) != 2: 

173 schema, path = "module", splited[0] 

174 else: 

175 schema, path = splited 

176 

177 hashcode = schema.split("@", 1)[1] if "@" in schema else "" 

178 schema = schema.split("@", 1)[0] if "@" in schema else schema 

179 

180 if schema == "mod": 

181 items = path.split("@", 1) 

182 if len(items) != 2: 

183 name, version = items[0], "" 

184 else: 

185 name, version = items 

186 ext = fromModule(importlib.import_module(name), version) 

187 return ext 

188 elif schema == "src": 

189 path = path.strip() 

190 if path == "stdin": 

191 src = path 

192 elif path == "interactive": 

193 src = path 

194 else: 

195 src = base64.b64decode(path.encode()).decode() 

196 ext = fromSource(src, filename="<string>", hashcode=hashcode) 

197 return ext 

198 elif schema == "file": 

199 file = Path(path) 

200 ext = fromFile(file, hashcode) 

201 return ext 

202 elif schema == "url": 

203 ext = fromUrl(path, hashcode) 

204 return ext 

205 elif schema == "ext": 

206 items = path.split("@", 1) 

207 if len(items) != 2: 

208 name, version = items[0], "" 

209 else: 

210 name, version = items 

211 ext = fromGallery(name, version, hashcode) 

212 return ext 

213 else: 

214 raise CoxbuildExtensionException(f"Unknown extension schema: {schema}")