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
« prev ^ index » next coverage.py v7.11.0, created at 2026-04-22 14:51 +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 # noqa: A004
13from typing_extensions import Never
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
25app = typer.Typer(
26 no_args_is_help=True,
27)
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)
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)
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)
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.
67 Todo:
68 more docs
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
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 )
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 = 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)
231 if verbosity is None:
232 verbosity = DEFAULT_VERBOSITY
234 state.load_config(config_file=config, verbosity=verbosity)
236 if show_config:
237 show_config_callback()
238 elif version:
239 version_callback()
240 # else: just continue