Coverage for src/pydal2sql/cli.py: 96%
53 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-05 19:04 +0200
1"""
2Create the Typer cli.
3"""
5import sys
6import typing
7from typing import Optional
9import typer
10from configuraptor import Singleton
11from pydal2sql_core.cli_support import core_alter, core_create, core_stub
12from rich import print
13from typing_extensions import Never
15from .__about__ import __version__
16from .typer_support import (
17 DEFAULT_VERBOSITY,
18 IS_DEBUG,
19 ApplicationState,
20 Verbosity,
21 with_exit_code,
22)
23from .types import DBType_Option, OptionalArgument, OutputFormat_Option, Tables_Option
25app = typer.Typer(
26 no_args_is_help=True,
27)
28state = ApplicationState()
31def info(*args: str) -> None: # pragma: no cover
32 """
33 'print' but with blue text.
34 """
35 print(f"[blue]{' '.join(args)}[/blue]", file=sys.stderr)
38def warn(*args: str) -> None: # pragma: no cover
39 """
40 'print' but with yellow text.
41 """
42 print(f"[yellow]{' '.join(args)}[/yellow]", file=sys.stderr)
45def danger(*args: str) -> None: # pragma: no cover
46 """
47 'print' but with red text.
48 """
49 print(f"[red]{' '.join(args)}[/red]", file=sys.stderr)
52@app.command()
53@with_exit_code(hide_tb=not IS_DEBUG)
54def create(
55 filename: OptionalArgument[str] = None,
56 tables: Tables_Option = None,
57 db_type: DBType_Option = None,
58 dialect: DBType_Option = None,
59 magic: Optional[bool] = None,
60 noop: Optional[bool] = None,
61 function: Optional[str] = None,
62 output_format: OutputFormat_Option = None,
63 output_file: Optional[str] = None,
64) -> bool:
65 """
66 Build the CREATE statements for one or more pydal/typedal tables.
68 Todo:
69 more docs
71 Examples:
72 pydal2sql create models.py
73 cat models.py | pydal2sql
74 pydal2sql # output from stdin
75 """
76 dialect = db_type.value if db_type else dialect.value if dialect else None
78 config = state.update_config(
79 magic=magic,
80 noop=noop,
81 tables=tables,
82 function=function,
83 format=output_format,
84 input=filename,
85 output=output_file,
86 ).update(dialect=dialect, _allow_none=True)
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
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.
123 Todo:
124 docs
126 Examples:
127 > pydal2sql alter @b3f24091a9201d6 examples/magic.py
128 compare magic.py at commit b3f... to current (= as in workdir).
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
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)
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
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.
177 Returns:
178 bool: True if the stub migration is generated successfully, False otherwise.
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 )
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 )
197def show_config_callback() -> Never:
198 """
199 --show-config requested!
200 """
201 print(state)
202 raise typer.Exit(0)
205def version_callback() -> Never:
206 """
207 --version requested!
208 """
209 print(f"pydal2sql Version: {__version__}")
211 raise typer.Exit(0)
214@app.callback(invoke_without_command=True)
215def main(
216 _: typer.Context,
217 config: str = None,
218 verbosity: Verbosity = DEFAULT_VERBOSITY,
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)
231 state.load_config(config_file=config, verbosity=verbosity)
233 if show_config:
234 show_config_callback()
235 elif version:
236 version_callback()
237 # else: just continue