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

54 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2026-04-22 14:51 +0200

1""" 

2Create the Typer cli. 

3""" 

4 

5import sys 

6import typing 

7from typing import Optional 

8 

9import typer 

10from configuraptor import Singleton 

11from pydal2sql_core.cli_support import core_alter, core_create, core_stub 

12from rich import print # noqa: A004 

13from typing_extensions import Never 

14 

15from .__about__ import __version__ 

16from .typer_support import ( 

17 DEFAULT_VERBOSITY, 

18 IS_DEBUG, 

19 Verbosity, 

20 state, 

21 with_exit_code, 

22) 

23from .types import DBType_Option, OptionalArgument, OutputFormat_Option, Tables_Option 

24 

25app = typer.Typer( 

26 no_args_is_help=True, 

27) 

28 

29 

30def info(*args: str) -> None: # pragma: no cover 

31 """ 

32 'print' but with blue text. 

33 """ 

34 print(f"[blue]{' '.join(args)}[/blue]", file=sys.stderr) 

35 

36 

37def warn(*args: str) -> None: # pragma: no cover 

38 """ 

39 'print' but with yellow text. 

40 """ 

41 print(f"[yellow]{' '.join(args)}[/yellow]", file=sys.stderr) 

42 

43 

44def danger(*args: str) -> None: # pragma: no cover 

45 """ 

46 'print' but with red text. 

47 """ 

48 print(f"[red]{' '.join(args)}[/red]", file=sys.stderr) 

49 

50 

51@app.command() 

52@with_exit_code(hide_tb=not IS_DEBUG) 

53def create( 

54 filename: OptionalArgument[str] = None, 

55 tables: Tables_Option = None, 

56 db_type: DBType_Option = None, 

57 dialect: DBType_Option = None, 

58 magic: Optional[bool] = None, 

59 noop: Optional[bool] = None, 

60 function: Optional[str] = None, 

61 output_format: OutputFormat_Option = None, 

62 output_file: Optional[str] = None, 

63) -> bool: 

64 """ 

65 Build the CREATE statements for one or more pydal/typedal tables. 

66 

67 Todo: 

68 more docs 

69 

70 Examples: 

71 pydal2sql create models.py 

72 cat models.py | pydal2sql 

73 pydal2sql # output from stdin 

74 """ 

75 dialect = db_type.value if db_type else dialect.value if dialect else None 

76 

77 config = state.update_config( 

78 magic=magic, 

79 noop=noop, 

80 db_type=dialect, 

81 tables=tables, 

82 function=function, 

83 format=output_format, 

84 input=filename, 

85 output=output_file, 

86 ) 

87 

88 if core_create( 

89 filename=config.input, 

90 db_type=config.db_type, 

91 tables=config.tables, 

92 verbose=state.verbosity > Verbosity.normal, 

93 noop=config.noop, 

94 magic=config.magic, 

95 function=config.function, 

96 output_format=config.format, 

97 output_file=config.output, 

98 ): 

99 print("[green] success! [/green]", file=sys.stderr) 

100 return True 

101 else: 

102 print("[red] create failed! [/red]", file=sys.stderr) 

103 return False 

104 

105 

106@app.command() 

107@with_exit_code(hide_tb=not IS_DEBUG) 

108def alter( 

109 filename_before: OptionalArgument[str] = None, 

110 filename_after: OptionalArgument[str] = None, 

111 db_type: DBType_Option = None, 

112 dialect: DBType_Option = None, 

113 tables: Tables_Option = None, 

114 magic: Optional[bool] = None, 

115 noop: Optional[bool] = None, 

116 function: Optional[str] = None, 

117 output_format: OutputFormat_Option = None, 

118 output_file: Optional[str] = None, 

119) -> bool: 

120 """ 

121 Create the migration statements from one state to the other, by writing CREATE, ALTER and DROP statements. 

122 

123 Todo: 

124 docs 

125 

126 Examples: 

127 > pydal2sql alter @b3f24091a9201d6 examples/magic.py 

128 compare magic.py at commit b3f... to current (= as in workdir). 

129 

130 > pydal2sql alter examples/magic.py@@b3f24091a9201d6 examples/magic_after_rename.py@latest 

131 compare magic.py (which was renamed to magic_after_rename.py), 

132 at a specific commit to the latest version in git (ignore workdir version). 

133 """ 

134 dialect = db_type.value if db_type else dialect.value if dialect else None 

135 

136 config = state.update_config( 

137 magic=magic, 

138 noop=noop, 

139 tables=tables, 

140 function=function, 

141 format=output_format, 

142 input=filename_before, 

143 output=output_file, 

144 ).update(dialect=dialect, _allow_none=True) 

145 

146 if core_alter( 

147 config.input, 

148 filename_after or config.input, 

149 db_type=config.db_type, 

150 tables=config.tables, 

151 verbose=state.verbosity > Verbosity.normal, 

152 noop=config.noop, 

153 magic=config.magic, 

154 function=config.function, 

155 output_format=config.format, 

156 output_file=config.output, 

157 ): 

158 print("[green] success! [/green]", file=sys.stderr) 

159 return True 

160 else: 

161 print("[red] alter failed! [/red]", file=sys.stderr) 

162 return False 

163 

164 

165@app.command() 

166@with_exit_code(hide_tb=not IS_DEBUG) 

167def stub( 

168 migration_name: typing.Annotated[str, typer.Argument()] = "stub_migration", 

169 output_format: OutputFormat_Option = None, 

170 output_file: Optional[str] = None, 

171 dry_run: typing.Annotated[bool, typer.Option("--dry", "--dry-run")] = False, 

172 is_typedal: typing.Annotated[bool, typer.Option("--typedal", "-t")] = False, 

173) -> bool: 

174 """ 

175 Command to generate a stub migration. 

176 

177 Returns: 

178 bool: True if the stub migration is generated successfully, False otherwise. 

179 

180 This command updates the configuration with the provided options and calls the core_stub function to generate the 

181 migration. 

182 """ 

183 config = state.update_config( 

184 format=output_format, 

185 output=output_file, 

186 ) 

187 

188 return core_stub( 

189 migration_name, # raw, without date or number 

190 output_format=config.format, 

191 output_file=config.output, 

192 dry_run=dry_run, 

193 is_typedal=is_typedal, 

194 ) 

195 

196 

197def show_config_callback() -> Never: 

198 """ 

199 --show-config requested! 

200 """ 

201 print(state) 

202 raise typer.Exit(0) 

203 

204 

205def version_callback() -> Never: 

206 """ 

207 --version requested! 

208 """ 

209 print(f"pydal2sql Version: {__version__}") 

210 

211 raise typer.Exit(0) 

212 

213 

214@app.callback(invoke_without_command=True) 

215def main( 

216 _: typer.Context, 

217 config: str = None, 

218 verbosity: Verbosity = None, 

219 # stops the program: 

220 show_config: bool = False, 

221 version: bool = False, 

222) -> None: 

223 """ 

224 This script can be used to generate the create or alter sql from pydal or typedal. 

225 """ 

226 if state.config: 

227 # if a config already exists, it's outdated, so we clear it. 

228 # only really applicable in Pytest scenarios where multiple commands are executed after eachother 

229 Singleton.clear(state.config) 

230 

231 if verbosity is None: 

232 verbosity = DEFAULT_VERBOSITY 

233 

234 state.load_config(config_file=config, verbosity=verbosity) 

235 

236 if show_config: 

237 show_config_callback() 

238 elif version: 

239 version_callback() 

240 # else: just continue