Coverage for src/pydal2sql/typer_support.py: 100%

1""" 

2Cli-specific support. 

3""" 

4 

5import contextlib 

6import functools 

7import inspect 

8import operator 

9import os 

10import sys 

11import typing 

12from dataclasses import dataclass 

13from enum import Enum, EnumMeta 

14from pathlib import Path 

15from typing import Any, Optional 

16 

17import configuraptor 

18import dotenv 

19import rich 

20import tomli 

21import typer 

22from black.files import find_project_root 

23from configuraptor import alias, postpone 

24from configuraptor.helpers import find_pyproject_toml 

25from pydal2sql_core.types import ( 

26 DEFAULT_OUTPUT_FORMAT, 

27 SUPPORTED_DATABASE_TYPES_WITH_ALIASES, 

28 SUPPORTED_OUTPUT_FORMATS, 

29) 

30from su6.core import ( 

31 EXIT_CODE_ERROR, 

32 EXIT_CODE_SUCCESS, 

33 T_Command, 

34 T_Inner_Wrapper, 

35 T_Outer_Wrapper, 

36) 

37from typing_extensions import Never 

38 

39 

40class ReprEnumMeta(EnumMeta): 

41 """ 

42 Give an Enum class a fancy repr. 

43 """ 

44 

45 def __repr__(cls) -> str: # sourcery skip 

46 """ 

47 Print all of the enum's members. 

48 """ 

49 options = typing.cast(typing.Iterable[Enum], cls.__members__.values()) # for mypy 

50 members_repr = ", ".join(f"{m.value!r}" for m in options) 

51 return f"{cls.__name__}({members_repr})" 

52 

53 

54class DynamicEnum(Enum, metaclass=ReprEnumMeta): 

55 """ 

56 Cmobine the enum class with the ReprEnumMeta metaclass. 

57 """ 

58 

59 

60def create_enum_from_literal(name: str, literal_type: typing.Any) -> typing.Type[DynamicEnum]: 

61 """ 

62 Transform a typing.Literal statement into an Enum. 

63 

64 literal_type can be a typing.Literal or a Union of Literals 

65 """ 

66 literals: list[str] = [] 

67 

68 if hasattr(literal_type, "__args__"): 

69 for arg in typing.get_args(literal_type): 

70 if hasattr(arg, "__args__"): 

71 # e.g. literal_type = typing.Union[typing.Literal['one', 'two']] 

72 literals.extend(typing.get_args(arg)) 

73 else: 

74 # e.g. literal_type = typing.Literal['one', 'two'] 

75 literals.append(arg) 

76 else: 

77 # e.g. literal_type = 'one' 

78 literals.append(str(literal_type)) 

79 

80 literals.sort() 

81 

82 enum_dict = {} 

83 

84 for literal in literals: 

85 enum_name = literal.replace(" ", "_").upper() 

86 enum_value = literal 

87 enum_dict[enum_name] = enum_value 

88 

89 return DynamicEnum(name, enum_dict) # type: ignore 

90 

91 

92class Verbosity(Enum): 

93 """ 

94 Verbosity is used with the --verbose argument of the cli commands. 

95 """ 

96 

97 # typer enum can only be string 

98 quiet = "1" 

99 normal = "2" 

100 verbose = "3" 

101 debug = "4" # only for internal use 

102 

103 @staticmethod 

104 def _compare( 

105 self: "Verbosity", 

106 other: "Verbosity_Comparable", 

107 _operator: typing.Callable[["Verbosity_Comparable", "Verbosity_Comparable"], bool], 

108 ) -> bool: 

109 """ 

110 Abstraction using 'operator' to have shared functionality between <, <=, ==, >=, >. 

111 

112 This enum can be compared with integers, strings and other Verbosity instances. 

113 

114 Args: 

115 self: the first Verbosity 

116 other: the second Verbosity (or other thing to compare) 

117 _operator: a callable operator (from 'operators') that takes two of the same types as input. 

118 """ 

119 match other: 

120 case Verbosity(): 

121 return _operator(self.value, other.value) 

122 case int(): 

123 return _operator(int(self.value), other) 

124 case str(): 

125 return _operator(int(self.value), int(other)) 

126 

127 def __gt__(self, other: "Verbosity_Comparable") -> bool: 

128 """ 

129 Magic method for self > other. 

130 """ 

131 return self._compare(self, other, operator.gt) 

132 

133 def __ge__(self, other: "Verbosity_Comparable") -> bool: 

134 """ 

135 Method magic for self >= other. 

136 """ 

137 return self._compare(self, other, operator.ge) 

138 

139 def __lt__(self, other: "Verbosity_Comparable") -> bool: 

140 """ 

141 Magic method for self < other. 

142 """ 

143 return self._compare(self, other, operator.lt) 

144 

145 def __le__(self, other: "Verbosity_Comparable") -> bool: 

146 """ 

147 Magic method for self <= other. 

148 """ 

149 return self._compare(self, other, operator.le) 

150 

151 def __eq__(self, other: typing.Union["Verbosity", str, int, object]) -> bool: 

152 """ 

153 Magic method for self == other. 

154 

155 'eq' is a special case because 'other' MUST be object according to mypy 

156 """ 

157 if other is Ellipsis or other is inspect._empty: 

158 # both instances of object; can't use Ellipsis or type(ELlipsis) = ellipsis as a type hint in mypy 

159 # special cases where Typer instanciates its cli arguments, 

160 # return False or it will crash 

161 return False 

162 if not isinstance(other, (str, int, Verbosity)): 

163 raise TypeError(f"Object of type {type(other)} can not be compared with Verbosity") 

164 return self._compare(self, other, operator.eq) 

165 

166 def __hash__(self) -> int: 

167 """ 

168 Magic method for `hash(self)`, also required for Typer to work. 

169 """ 

170 return hash(self.value) 

171 

172 

173Verbosity_Comparable = Verbosity | str | int 

174 

175DEFAULT_VERBOSITY = Verbosity.normal 

176 

177 

178class AbstractConfig(configuraptor.TypedConfig, configuraptor.Singleton): 

179 """ 

180 Used by state.config and plugin configs. 

181 """ 

182 

183 _strict = True 

184 

185 

186DB_Types: typing.Any = create_enum_from_literal("DBType", SUPPORTED_DATABASE_TYPES_WITH_ALIASES) 

187 

188 

189# @dataclass 

190class Config(AbstractConfig): 

191 """ 

192 Used as typed version of the [tool.pydal2sql] part of pyproject.toml. 

193 

194 Also accessible via state.config 

195 """ 

196 

197 # settings go here 

198 db_type: typing.Optional[SUPPORTED_DATABASE_TYPES_WITH_ALIASES] = postpone() 

199 tables: Optional[list[str]] = None 

200 magic: bool = False 

201 function: str = "define_tables" 

202 format: SUPPORTED_OUTPUT_FORMATS = DEFAULT_OUTPUT_FORMAT 

203 dialect: typing.Optional[SUPPORTED_DATABASE_TYPES_WITH_ALIASES] = alias("db_type") 

204 input: Optional[str] = None 

205 output: Optional[str] = None 

206 noop: bool = False 

207 

208 pyproject: typing.Optional[str] = None 

209 

210 

211MaybeConfig = Optional[Config] 

212 

213 

214def _get_pydal2sql_config(overwrites: dict[str, Any], toml_path: Optional[str | Path] = None) -> MaybeConfig: 

215 """ 

216 Parse the users pyproject.toml (found using black's logic) and extract the tool.pydal2sql part. 

217 

218 The types as entered in the toml are checked using _ensure_types, 

219 to make sure there isn't a string implicitly converted to a list of characters or something. 

220 

221 Args: 

222 overwrites: cli arguments can overwrite the config toml. 

223 toml_path: by default, black will search for a relevant pyproject.toml. 

224 If a toml_path is provided, that file will be used instead. 

225 """ 

226 if toml_path is None: 

227 toml_path = find_pyproject_toml() 

228 

229 if not toml_path: 

230 return None 

231 

232 with open(toml_path, "rb") as f: 

233 full_config = tomli.load(f) 

234 

235 tool_config = full_config["tool"] 

236 

237 config = configuraptor.load_into(Config, tool_config, key="pydal2sql") 

238 

239 config.update(pyproject=str(toml_path)) 

240 config.update(**overwrites) 

241 

242 return config 

243 

244 

245def get_pydal2sql_config(toml_path: str = None, verbosity: Verbosity = DEFAULT_VERBOSITY, **overwrites: Any) -> Config: 

246 """ 

247 Load the relevant pyproject.toml config settings. 

248 

249 Args: 

250 verbosity: if something goes wrong, level 3+ will show a warning and 4+ will raise the exception. 

251 toml_path: --config can be used to use a different file than ./pyproject.toml 

252 overwrites (dict[str, Any): cli arguments can overwrite the config toml. 

253 If a value is None, the key is not overwritten. 

254 """ 

255 # strip out any 'overwrites' with None as value 

256 overwrites = configuraptor.convert_config(overwrites) 

257 

258 try: 

259 if config := _get_pydal2sql_config(overwrites, toml_path=toml_path): 

260 return config 

261 raise ValueError("Falsey config?") 

262 except Exception as e: 

263 # something went wrong parsing config, use defaults 

264 if verbosity > 3: 

265 # verbosity = debug 

266 raise e 

267 elif verbosity > 2: 

268 # verbosity = verbose 

269 print("Error parsing pyproject.toml, falling back to defaults.", file=sys.stderr) 

270 return Config(**overwrites) 

271 

272 

273@dataclass() 

274class ApplicationState: 

275 """ 

276 Application State - global user defined variables. 

277 

278 State contains generic variables passed BEFORE the subcommand (so --verbosity, --config, ...), 

279 whereas Config contains settings from the config toml file, updated with arguments AFTER the subcommand 

280 (e.g. pydal2sql subcommand <directory> --flag), directory and flag will be updated in the config and not the state. 

281 

282 To summarize: 'state' is applicable to all commands and config only to specific ones. 

283 """ 

284 

285 verbosity: Verbosity = DEFAULT_VERBOSITY 

286 config_file: Optional[str] = None # will be filled with black's search logic 

287 config: MaybeConfig = None 

288 

289 def __post_init__(self) -> None: 

290 """ 

291 Runs after the dataclass init. 

292 """ 

293 

294 def load_config(self, **overwrites: Any) -> Config: 

295 """ 

296 Load the pydal2sql config from pyproject.toml (or other config_file) with optional overwriting settings. 

297 

298 Also updates attached plugin configs. 

299 """ 

300 if "verbosity" in overwrites: 

301 self.verbosity = overwrites["verbosity"] 

302 if "config_file" in overwrites: 

303 self.config_file = overwrites.pop("config_file") 

304 

305 self.config = get_pydal2sql_config(toml_path=self.config_file, **overwrites) 

306 return self.config 

307 

308 def get_config(self) -> Config: 

309 """ 

310 Get a filled config instance. 

311 """ 

312 return self.config or self.load_config() 

313 

314 def update_config(self, **values: Any) -> Config: 

315 """ 

316 Overwrite default/toml settings with cli values. 

317 

318 Example: 

319 `config = state.update_config(directory='src')` 

320 This will update the state's config and return the same object with the updated settings. 

321 """ 

322 existing_config = self.get_config() 

323 

324 values = configuraptor.convert_config(values) 

325 existing_config.update(**values) 

326 return existing_config 

327 

328 

329def with_exit_code(hide_tb: bool = True) -> T_Outer_Wrapper: 

330 """ 

331 Convert the return value of an app.command (bool or int) to an typer Exit with return code, \ 

332 Unless the return value is Falsey, in which case the default exit happens (with exit code 0 indicating success). 

333 

334 Usage: 

335 > @app.command() 

336 > @with_exit_code() 

337 def some_command(): ... 

338 

339 When calling a command from a different command, _suppress=True can be added to not raise an Exit exception. 

340 

341 See Also: 

342 github.com:trialandsuccess/su6-checker 

343 """ 

344 

345 def outer_wrapper(func: T_Command) -> T_Inner_Wrapper: 

346 @functools.wraps(func) 

347 def inner_wrapper(*args: Any, **kwargs: Any) -> Never: 

348 try: 

349 result = func(*args, **kwargs) 

350 except Exception as e: 

351 result = EXIT_CODE_ERROR 

352 if hide_tb: 

353 rich.print(f"[red]{e}[/red]", file=sys.stderr) 

354 else: # pragma: no cover 

355 raise e 

356 finally: 

357 sys.stdout.flush() 

358 sys.stderr.flush() 

359 

360 if isinstance(result, bool): 

361 if result in (None, True): 

362 # assume no issue then 

363 result = EXIT_CODE_SUCCESS 

364 elif result is False: 

365 result = EXIT_CODE_ERROR 

366 

367 raise typer.Exit(code=int(result or 0)) 

368 

369 return inner_wrapper 

370 

371 return outer_wrapper 

372 

373 

374def _is_debug() -> bool: # pragma: no cover 

375 folder, _ = find_project_root((os.getcwd(),)) 

376 if not folder: 

377 folder = Path(os.getcwd()) 

378 dotenv.load_dotenv(folder / ".env") 

379 

380 return os.getenv("IS_DEBUG") == "1" 

381 

382 

383def is_debug() -> bool: # pragma: no cover 

384 """ 

385 Returns whether IS_DEBUG = 1 in the .env. 

386 """ 

387 with contextlib.suppress(Exception): 

388 return _is_debug() 

389 return False 

390 

391 

392IS_DEBUG = is_debug()