Coverage for obj2cli/_cli_click.py: 52%

1import difflib 

2import functools 

3import importlib 

4import importlib.resources 

5import io 

6import os 

7import pickle 

8import pkgutil 

9import pprint 

10import subprocess 

11import sys 

12from logging import DEBUG, INFO, basicConfig, getLogger 

13 

14import click 

15import yaml 

16from jinja2 import Environment 

17 

18from .parser import Parser 

19from .version import VERSION 

20 

21log = getLogger(__name__) 

22 

23out_formats = { 

24 "pickle": lambda d, f: pickle.dump(d, f), 

25 "yaml": lambda d, f: yaml.dump(d, stream=f), 

26 "pprint": lambda d, f: f.write(pprint.pformat(d).encode("utf-8")), 

27} 

28 

29in_formats = { 

30 "pickle": lambda f: pickle.load(f), 

31 "yaml": lambda f: yaml.load(f, Loader=yaml.FullLoader), 

32} 

33 

34 

35@click.version_option(version=VERSION, prog_name="obj2cli") 

36@click.group(invoke_without_command=True) 

37@click.pass_context 

38def cli(ctx): 

39 if ctx.invoked_subcommand is None: 

40 print(ctx.get_help()) 

41 

42 

43def set_verbose(flag): 

44 fmt = "%(asctime)s %(levelname)s %(message)s" 

45 if flag: 

46 basicConfig(level=DEBUG, format=fmt) 

47 else: 

48 basicConfig(level=INFO, format=fmt) 

49 

50 

51_cli_option = [ 

52 click.option("--verbose/--no-verbose"), 

53] 

54 

55_cls_option = _cli_option + [ 

56 click.option("--module", type=str), 

57 click.option("--package", type=str), 

58 click.option("--class", "klass", type=str), 

59] 

60 

61out_option = [ 

62 click.option( 

63 "--output", 

64 type=click.File("wb"), 

65 default=sys.stdout.buffer, 

66 show_default="STDOUT", 

67 ), 

68 click.option( 

69 "--format", 

70 type=click.Choice(out_formats.keys()), 

71 default="yaml", 

72 show_default=True, 

73 ), 

74] 

75 

76in_option = [ 

77 click.option( 

78 "--input", type=click.File("rb"), default=sys.stdin.buffer, show_default="STDIN" 

79 ), 

80 click.option( 

81 "--format", 

82 type=click.Choice(in_formats.keys()), 

83 default="yaml", 

84 show_default=True, 

85 ), 

86] 

87 

88 

89def multi_options(decs): 

90 def deco(f): 

91 for dec in reversed(decs): 

92 f = dec(f) 

93 return f 

94 

95 return deco 

96 

97 

98def cli_option(func): 

99 @functools.wraps(func) 

100 def wrap(verbose, *args, **kwargs): 

101 set_verbose(verbose) 

102 return func(*args, **kwargs) 

103 

104 return multi_options(_cli_option)(wrap) 

105 

106 

107def cls_option(func): 

108 @functools.wraps(func) 

109 def wrap(verbose, klass, module, package, *args, **kwargs): 

110 set_verbose(verbose) 

111 log.debug("loading module %s, package=%s", module, package) 

112 mod = importlib.import_module(name=module, package=package) 

113 if klass is not None: 

114 cls = getattr(mod, klass) 

115 log.debug("class: %s", cls) 

116 else: 

117 cls = None 

118 return func(cls, *args, **kwargs) 

119 

120 return multi_options(_cls_option)(wrap) 

121 

122 

123def resource_option(dest, dirname=None, ext=""): 

124 if dirname is None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 dirname = dest 

126 

127 def _resource_option(func): 

128 @functools.wraps(func) 

129 def wrap(*args, **kwargs): 

130 val = kwargs.pop(dest) 

131 exval = kwargs.pop(f"{dest}_example") 

132 if val is not None: 

133 # file handle is handed to `func` and must stay open past this 

134 # scope; the caller is responsible for closing it. 

135 kwargs[dest] = open(val, "rb") # noqa: SIM115 

136 else: 

137 kwargs[dest] = io.BytesIO( 

138 pkgutil.get_data(__package__, os.path.join(dirname, exval + ext)) 

139 ) 

140 return func(*args, **kwargs) 

141 

142 try: 

143 names = [ 

144 p.name 

145 for p in importlib.resources.files(__package__) 

146 .joinpath(dirname) 

147 .iterdir() 

148 ] 

149 except FileNotFoundError: 

150 names = [] 

151 log.debug("resources in pkg=%s, dir=%s: %s", __package__, dirname, names) 

152 names = [x[: -len(ext)] for x in filter(lambda f: f.endswith(ext), names)] 

153 opts = [ 

154 click.option(f"--{dest}", type=click.Path()), 

155 click.option(f"--{dest}-example", type=click.Choice(names)), 

156 ] 

157 return multi_options(opts)(wrap) 

158 

159 return _resource_option 

160 

161 

162@cli.command() 

163@cls_option 

164@multi_options(out_option) 

165def parse(cls, format, output): 

166 ps = Parser() 

167 log.debug("start parse: %s", cls.__name__) 

168 data = ps.parse_cls(cls) 

169 out_formats.get(format)(data, output) 

170 # pickle.dump(data, sys.stdout.buffer) 

171 

172 

173def template_args(data, cls=None): 

174 ps = Parser() 

175 funcs = {} 

176 for k, v in data.items(): 

177 if k in ("__classmeta__", "__init__", "doc"): 

178 continue 

179 funcs[k] = ps.fn_args(v) 

180 if cls is None: 

181 cls = data.get("__classmeta__", {}).get("class", None) 

182 return { 

183 "klass": cls, 

184 "parsed": data, 

185 "constructor": ps.fn_args(data.get("__init__", {})), 

186 "funcs": funcs, 

187 "classmeta": data.get("__classmeta__", {}), 

188 } 

189 

190 

191@cli.command("generate") 

192@cls_option 

193@multi_options(in_option) 

194@resource_option(dest="template", dirname="template", ext="_cli.j2") 

195@click.option("--autopep8/--no-autopep8", default=False) 

196def gen(cls, input, format, template, autopep8): 

197 data = in_formats.get(format)(input) 

198 env = Environment() 

199 tmpl = env.from_string(template.read().decode("utf-8")) 

200 res = tmpl.render(**template_args(data, cls)) 

201 if autopep8: 

202 p = subprocess.Popen(["autopep8", "-"], stdin=subprocess.PIPE) 

203 p.communicate(res.encode("utf8")) 

204 else: 

205 print(res) 

206 

207 

208@cli.command() 

209@cls_option 

210@multi_options(in_option) 

211@resource_option(dest="template", dirname="template", ext="_cli.j2") 

212@click.argument("other", type=click.File("r"), required=True) 

213def diff(cls, input, format, template, other): 

214 data = in_formats.get(format)(input) 

215 env = Environment() 

216 tmpl = env.from_string(template.read()) 

217 res = tmpl.render(**template_args(data, cls)) 

218 before = other.read().split("\n") 

219 current = res.split("\n") 

220 differ = difflib.unified_diff( 

221 before, current, fromfile="before.py", tofile="current.py", lineterm="" 

222 ) 

223 print("\n".join(differ)) 

224 

225 

226@cli.command("print") 

227@cli_option 

228@multi_options(in_option) 

229def show(input, format): 

230 data = in_formats.get(format)(input) 

231 pprint.pprint(data) 

232 

233 

234@cli.command("print-tmpl-args") 

235@cls_option 

236@multi_options(in_option) 

237def show_tmplarg(cls, input, format): 

238 data = in_formats.get(format)(input) 

239 pprint.pprint(template_args(data, cls)) 

240 

241 

242if __name__ == "__main__": 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true

243 cli()