Coverage for src/mafw/scripts/mafw_exe.py: 94%
321 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5The execution framework.
7This module provides the run functionality to the whole library.
9It is heavily relying on ``click`` for the generation of commands, options, and arguments.
11.. click:: mafw.scripts.mafw_exe:cli
12 :prog: mafw
13 :nested: full
15"""
17import datetime
18import logging
19import pathlib
20import shutil
21import sys
22import warnings
23from enum import IntEnum
24from typing import Any
26import click
27from click.exceptions import ClickException
28from pwiz import DATABASE_MAP, make_introspector # type: ignore[import-untyped]
29from rich import print as rprint
30from rich import traceback
31from rich.align import Align
32from rich.logging import RichHandler
33from rich.prompt import Prompt
34from rich.rule import Rule
35from rich.table import Table
36from rich_pyfiglet import RichFiglet
38from mafw.__about__ import __version__
39from mafw.db.db_configurations import db_scheme, default_conf
40from mafw.db.db_wizard import dump_models
41from mafw.enumerators import ProcessorExitStatus
42from mafw.lazy_import import LazyImportProcessor
43from mafw.mafw_errors import AbortProcessorException
44from mafw.plugin_manager import get_plugin_manager
45from mafw.runner import MAFwApplication
46from mafw.tools.click_extensions import (
47 AbbreviateGroup,
48 check_ci_completion_guard,
49 completion_script_path,
50 completion_source_script,
51 install_completion,
52 is_script_already_installed,
53 resolve_completion_shell,
54 uninstall_completion_files,
55)
56from mafw.tools.parallel import is_free_threading
57from mafw.tools.shell_tools import CONSOLE
58from mafw.tools.toml_tools import generate_steering_file
60suppress = [click]
61traceback.install(show_locals=True, suppress=suppress)
64class MAFwGroup(AbbreviateGroup):
65 """Click group with abbreviation and MAFw-specific exit handling."""
67 group_class: type[click.Group] = AbbreviateGroup
69 def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
70 """Parse arguments and record the selected root command before invoke."""
71 rv = super().parse_args(ctx, args)
72 if ( 72 ↛ 76line 72 didn't jump to line 76 because the condition on line 72 was always true
73 ctx.parent is None
74 ): # this is assuring that we store only the first command corresponding to the root context
75 ctx.meta['_mafw_invoked_subcommand'] = self._detect_command_name(args)
76 return rv
78 def _detect_command_name(self, args: list[str]) -> str | None:
79 """Return the first subcommand token while skipping root option values."""
80 # This is probably an overkill. After the default parsing the first element in args should be the first
81 # (abbreviated or fully typed) command. All global options will be removed from args.
82 # so potentially we could return self.get_command(ctx, args[0]),
83 # I will leave it like this because it might be useful in the future.
84 option_params = {opt for param in self.params for opt in getattr(param, 'opts', [])}
85 expects_value = {
86 opt for param in self.params if not getattr(param, 'is_flag', False) for opt in getattr(param, 'opts', [])
87 }
88 skip_next = False
89 for token in args:
90 if skip_next: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 skip_next = False
92 continue
93 if token in option_params: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 if token in expects_value:
95 skip_next = True
96 continue
97 if token.startswith('-'): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 continue
99 selected = self.get_command(click.Context(self), token)
100 return selected.name if selected is not None else token
101 return None
103 def invoke(self, ctx: click.Context) -> Any:
104 """Invoke the command and normalize Click exceptions to MAFw exits."""
105 print_banner(ctx)
107 try:
108 return super().invoke(ctx)
109 except ClickException as exc:
110 exc.show()
111 sys.exit(exc.exit_code)
112 except (SystemExit, click.exceptions.Exit):
113 raise
114 except Exception:
115 sys.exit(ReturnValue.Error)
117 def main(self, *args: Any, **kwargs: Any) -> None: # type: ignore[override]
118 """Run the CLI and convert returned statuses to process exit codes."""
119 try:
120 rv = super().main(*args, standalone_mode=False, **kwargs) # type: ignore[call-overload]
121 if isinstance(rv, (ReturnValue, int)):
122 sys.exit(rv)
123 sys.exit(ReturnValue.OK)
124 except ClickException as exc:
125 exc.show()
126 sys.exit(exc.exit_code)
127 except SystemExit:
128 raise
129 except Exception:
130 sys.exit(ReturnValue.Error)
133LEVELS = {'debug': 10, 'info': 20, 'warning': 30, 'error': 40, 'critical': 50}
134CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
136_warnings_captured = False
138welcome_message = RichFiglet(
139 'MAFw',
140 colors=['#ff0000', 'magenta1', 'blue3'],
141 horizontal=True,
142 # font='banner4',
143 remove_blank_lines=True,
144 border='ROUNDED',
145 border_color='#ff0000',
146)
149def print_banner(ctx: click.Context) -> None:
150 """
151 Print the welcome banner only once, only for rich UI,
152 and only if not disabled.
154 .. note::
156 ctx.obj is not yet populated when Group.invoke() is called,
157 but ctx.params contains parsed options for the current group. We
158 therefore inspect the root context params.
159 """
160 # avoid printing while click is still doing resilient parsing (e.g., --help)
161 if getattr(ctx, 'resilient_parsing', False):
162 return
164 root = ctx.find_root()
166 # Parse-time command resolution records the selected subcommand before
167 # invoke() emits any output, so completion can remain quiet.
168 # Possible improvements: allow selection of commands for which there must not be any banner printing.
169 if root.meta.get('_mafw_invoked_subcommand') == 'completion':
170 return
172 params = getattr(root, 'params', {}) or {}
173 ui = params.get('ui')
174 no_banner = params.get('no_banner')
176 # Normalize and check
177 if no_banner:
178 return
179 if not ui: # pragma: no cover
180 # if nothing available, be conservative and don't print
181 return
182 if str(ui).lower() != 'rich':
183 return
185 # Print only once per process / invocation
186 if root.meta.get('_banner_printed', False):
187 return
189 console = CONSOLE
190 console.print(welcome_message)
191 root.meta['_banner_printed'] = True
194def custom_formatwarning(
195 message: Warning | str, category: type[Warning], filename: str, lineno: int, line: str | None = None
196) -> str:
197 """Return the pure message of the warning."""
198 return str(message)
201def is_bugged_warning_capture_version() -> bool:
202 """Check if the Python version has the warning capture bug in free-threading mode.
204 Python 3.14.0 through 3.14.3 (free-threading builds only) have a bug where
205 ``logging.captureWarnings(True)`` can crash or misbehave. This function returns
206 True when the current interpreter is affected.
208 .. note::
210 See https://github.com/python/cpython/pull/146374 for details.
212 .. todo::
214 Remove this workaround when Python 3.14 is no longer a supported version
215 (expected late 2030).
217 :return: True if the interpreter is affected by the warning capture bug, False otherwise.
218 :rtype: bool
219 """
220 version = sys.version_info
221 return version.major == 3 and version.minor == 14 and version.micro <= 3 and is_free_threading()
224warnings.formatwarning = custom_formatwarning
225# Disable warning capture on bugged free-threading builds (3.14.0–3.14.3).
226if is_bugged_warning_capture_version(): 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 logging.captureWarnings(False)
228else:
229 logging.captureWarnings(True)
231# get the root logger
232log = logging.getLogger()
235class ReturnValue(IntEnum):
236 """Enumerator to handle the script return value."""
238 OK = 0
239 """No error"""
241 Error = 1
242 """Generic error"""
245def logger_setup(level: str, ui: str, tracebacks: bool) -> None:
246 """Set up the logger.
248 This function is actually configuring the root logger level from the command line options and it attaches either
249 a RichHandler or a StreamHandler depending on the user interface type.
251 The `tracebacks` flag is used only by the RichHandler. Printing the tracebacks is rather useful when debugging
252 the code, but it could be detrimental for final users. In normal circumstances, tracebacks is set to False,
253 and is turned on when the debug flag is activated.
255 :param level: Logging level as a string.
256 :type level: str
257 :param ui: User interface as a string ('rich' or 'console').
258 :type ui: str
259 :param tracebacks: Enable/disable the logging of exception tracebacks.
260 """
261 level = level.lower()
262 ui = ui.lower()
264 log.setLevel(LEVELS[level])
265 handler: logging.Handler
267 if ui == 'rich':
268 fs = '%(message)s'
269 handler = RichHandler(
270 rich_tracebacks=tracebacks, markup=True, show_path=False, log_time_format='%Y%m%d-%H:%M:%S'
271 )
272 else:
273 fs = '%(asctime)s - %(levelname)s - %(message)s'
274 handler = logging.StreamHandler()
276 formatter = logging.Formatter(fs)
277 handler.setFormatter(formatter)
278 log.addHandler(handler)
279 # _ensure_warning_capture()
282def display_exception(exception: Exception, show_traceback: bool = False) -> None:
283 """
284 Display exception information with optional debug details.
286 This function logs exception information at the critical level. When show_traceback is enabled,
287 it logs the full exception including traceback information. Otherwise, it logs a simplified
288 message directing users to enable debug mode for more details.
290 :param exception: The exception to be displayed and logged.
291 :type exception: Exception
292 :param show_traceback: Flag indicating whether to show detailed traceback information. Defaults to False
293 :type show_traceback: bool
294 """
296 if show_traceback:
297 log.critical('A critical error occurred')
298 log.exception(exception)
299 else:
300 log.critical('A critical error occurred. Set option -D to get traceback output')
301 log.exception(f'{exception.__class__.__name__}: {exception}', exc_info=False, stack_info=False, stacklevel=1)
304@click.group(invoke_without_command=True, context_settings=CONTEXT_SETTINGS, name='mafw', cls=MAFwGroup)
305@click.pass_context
306@click.option(
307 '--log-level',
308 type=click.Choice(['debug', 'info', 'warning', 'error', 'critical'], case_sensitive=False),
309 show_default=True,
310 default='info',
311 help='Log level',
312)
313@click.option(
314 '--ui',
315 type=click.Choice(['console', 'rich'], case_sensitive=False),
316 default='rich',
317 help='The user interface',
318 show_default=True,
319)
320@click.option('-D', '--debug', is_flag=True, default=False, help='Show debug information about errors')
321@click.option('--no-banner', is_flag=True, default=False, help='Disable the welcome banner')
322@click.version_option(__version__, '-v', '--version')
323def cli(ctx: click.core.Context, log_level: str, ui: str, debug: bool, no_banner: bool) -> None:
324 """
325 The Modular Analysis Framework execution.
327 This is the command line interface where you can configure and launch your analysis tasks.
329 More information on our documentation page.
330 \f
332 :param ctx: The click context.
333 :type ctx: click.core.Context
334 :param log_level: The logging level as a string. Choice from debug, info, warning, error and critical.
335 :type log_level: str
336 :param ui: The user interface as a string. Choice from console and rich.
337 :type ui: str
338 :param debug: Flag to show debug information about exception.
339 :type debug: bool
340 :param no_banner: Flag to disable the welcome banner.
341 :type no_banner: bool
342 """
343 ctx.ensure_object(dict)
344 ctx.obj = {'log_level': log_level, 'ui': ui, 'debug': debug, 'no_banner': no_banner}
345 logger_setup(log_level, ui, debug)
347 if ctx.invoked_subcommand is None:
348 rprint('Use --help to get a quick help on the mafw command.')
351@cli.group(name='completion')
352@click.pass_context
353def completion(ctx: click.Context) -> None:
354 """
355 Manage shell completion for the ``mafw`` command.
357 The completion workflow installs the Click-generated shell code into the
358 active virtual environment, updates the activation script so completion is
359 loaded automatically, and exposes a ``show`` helper for direct evaluation.
361 \f
363 .. versionadded:: v2.2
365 :param ctx: The click context.
366 :type ctx: click.core.Context
367 """
368 ctx.ensure_object(dict)
369 ctx.obj['tool_name'] = 'mafw'
372@completion.command(name='install')
373@click.option(
374 '-s',
375 '--shell',
376 type=click.Choice(['auto', 'bash', 'zsh', 'fish'], case_sensitive=False),
377 default='auto',
378 show_default=True,
379 help='Target shell for completion installation',
380)
381@click.option('-F', '--force', is_flag=True, default=False, help='Reinstall completion even if already loaded')
382@click.pass_context
383def completion_install(ctx: click.Context, shell: str, force: bool) -> None:
384 """
385 Install the ``mafw`` shell completion script.
387 When ``--shell`` is omitted, the command guesses the shell from ``$SHELL``.
388 The generated Click completion script is stored in the active virtual
389 environment and the activation script is updated so that future shell
390 sessions load completion automatically.
392 \f
394 .. versionadded:: v2.2
396 :param ctx: The click context.
397 :type ctx: click.core.Context
398 :param shell: Target shell selector.
399 :type shell: str
400 :param force: Reinstall completion even if already loaded.
401 :type force: bool
402 """
403 check_ci_completion_guard()
404 tool_name = ctx.obj['tool_name']
405 resolved_shell = resolve_completion_shell(shell)
406 script_path = completion_script_path(tool_name, resolved_shell)
408 if is_script_already_installed(tool_name, resolved_shell) and not force: 408 ↛ 409line 408 didn't jump to line 409 because the condition on line 408 was never true
409 raise click.ClickException(
410 f'MAFw completion is already installed in {script_path}. Use --force to reinstall it.'
411 )
413 install_completion(tool_name, resolved_shell, force, script_path)
414 rprint(f'Completion script installed in [blue underline]{script_path}[/blue underline].')
415 rprint('Exit and re-enter the virtual environment to activate shell completion.')
418@completion.command(name='uninstall')
419@click.pass_context
420def completion_uninstall(ctx: click.Context) -> None:
421 """
422 Remove the installed ``mafw`` shell completion files.
424 This command removes generated completion files from ``share/mafw`` and
425 strips the marker block from the activation scripts.
427 .. versionadded:: v2.2
429 :param ctx: The click context.
430 :type ctx: click.core.Context
431 """
432 tool_name = ctx.obj['tool_name']
433 uninstall_completion_files(tool_name, None)
434 rprint(f'Shell completion for {tool_name} has been removed from the active virtual environment.')
437@completion.command(name='show')
438@click.option(
439 '-s',
440 '--shell',
441 type=click.Choice(['auto', 'bash', 'zsh', 'fish'], case_sensitive=False),
442 default='auto',
443 show_default=True,
444 help='Target shell for completion output',
445)
446@click.pass_context
447def completion_show(ctx: click.Context, shell: str) -> None:
448 """
449 Display the Click completion script on standard output.
451 The output is intentionally clean so it can be used with ``eval``:
453 .. code-block:: console
455 eval "$(mafw completion show)"
457 .. versionadded:: v2.2
459 :param ctx: The click context.
460 :type ctx: click.core.Context
461 :param shell: Target shell selector.
462 :type shell: str
463 """
464 check_ci_completion_guard()
465 tool_name = ctx.obj['tool_name']
466 resolved_shell = resolve_completion_shell(shell)
467 click.echo(completion_source_script(tool_name, resolved_shell), nl=True)
470@cli.command(name='list')
471@click.pass_obj
472def list_processors(obj: dict[str, Any]) -> ReturnValue:
473 """Display the list of available processors.
475 This command will retrieve all available processors via the plugin manager. Both internal and external processors
476 will be listed if the ext-plugins option is passed.
477 \f
479 """
480 try:
481 plugin_manager = get_plugin_manager()
482 plugins = plugin_manager.load_plugins({'processors'})
483 available_processors = plugins.processor_list
484 print('\n')
485 table = Table(
486 title='Available processors',
487 header_style='orange3',
488 expand=True,
489 title_style='italic red',
490 )
491 table.add_column('Processor Name', justify='left', style='cyan')
492 table.add_column('Package Name', justify='left', style='cyan')
493 table.add_column('Module', justify='left', style='cyan')
494 mafw_processors = 0
495 other_processors = 0
496 for processor in available_processors:
497 if isinstance(processor, LazyImportProcessor):
498 package, module = processor.plugin_qualname.split('.', 1)
499 name = processor.plugin_name
500 else:
501 package, module = processor.__module__.split('.', 1)
502 name = processor.__name__
503 table.add_row(name, package, module)
504 if package == 'mafw':
505 mafw_processors += 1
506 else:
507 other_processors += 1
508 table.caption = f'Total processors = {len(available_processors)}, internal = {mafw_processors}, external = {other_processors}'
509 table.caption_style = 'italic green'
510 console = CONSOLE
511 console.print(Align.center(table))
512 return ReturnValue.OK
513 except Exception as e:
514 display_exception(e, show_traceback=obj['debug'])
515 return ReturnValue.Error
518@cli.command(name='steering')
519@click.pass_obj
520@click.option('--show/--no-show', default=False, help='Display the generated steering file on console')
521@click.option('--ext-plugins/--no-ext-plugin', default=True, help='Load external plugins')
522@click.option('--open-editor/--no-open-editor', default=False, help='Open the file in your editor.')
523@click.option(
524 '--db-engine',
525 type=click.Choice(['sqlite', 'mysql', 'postgresql'], case_sensitive=False),
526 help='Select a DB engine',
527 default='sqlite',
528)
529@click.option('--db-url', type=str, default=':memory:', help='URL to the DB')
530@click.argument('steering-file', type=click.Path())
531def generate_steering(
532 obj: dict[str, Any],
533 show: bool,
534 ext_plugins: bool,
535 open_editor: bool,
536 steering_file: pathlib.Path,
537 db_engine: str,
538 db_url: str,
539) -> ReturnValue:
540 """Generates a steering file with the default parameters of all available processors.
542 STEERING_FILE A path to the steering file to execute.
544 The user must modify the generated steering file to ensure it can be executed using the run command.
545 \f
547 :param obj: The context object being passed from the main command.
548 :type obj: dict
549 :param show: Display the steering file in the console after the generation. Defaults to False.
550 :type show: bool
551 :param ext_plugins: Extend the search for processor to external libraries.
552 :type ext_plugins: bool
553 :param open_editor: Open a text editor after the generation to allow direct editing.
554 :type open_editor: bool
555 :param steering_file: The steering file path.
556 :type steering_file: Path
557 :param db_engine: The name of the db engine.
558 :type db_engine: str
559 :param db_url: The URL of the database.
560 :type db_url: str
561 """
562 try:
563 plugin_manager = get_plugin_manager()
564 plugins = plugin_manager.load_plugins({'processors'})
565 available_processors = plugins.processor_list
566 # db_engine is already sure to be in the default conf because the Choice is assuring it.
567 database_conf = default_conf[db_engine]
568 database_conf['URL'] = db_scheme[db_engine] + db_url
569 generate_steering_file(steering_file, available_processors, database_conf)
571 if show:
572 console = CONSOLE
573 with open(steering_file) as fp:
574 text = fp.read()
575 with console.pager():
576 console.print(text, highlight=True)
577 console.print(Rule())
579 if open_editor:
580 click.edit(filename=str(steering_file))
581 else:
582 rprint(f'A generic steering file has been saved in [blue underline]{steering_file}[/blue underline].')
583 rprint('Open it in your favourite text editor, change the processors_to_run list and save it.')
584 rprint('')
585 rprint(f'To execute it launch: [blue]mafw run {steering_file}[/blue].')
587 return ReturnValue.OK
589 except Exception as e:
590 display_exception(e, show_traceback=obj['debug'])
591 return ReturnValue.Error
594@cli.command()
595@click.pass_obj
596@click.argument('steering-file', type=click.Path())
597def run(obj: dict[str, Any], steering_file: click.Path) -> ReturnValue:
598 """Runs a steering file.
600 STEERING_FILE A path to the steering file to execute.
602 \f
604 :param obj: The context object being passed from the main command.
605 :type obj: dict
606 :param steering_file: The path to the output steering file.
607 :type steering_file: Path
608 """
609 try:
610 app = MAFwApplication(steering_file) # type: ignore
611 pes = app.run()
612 if pes == ProcessorExitStatus.Successful:
613 rv = ReturnValue.OK
614 else:
615 rv = ReturnValue.Error
616 return rv
618 except AbortProcessorException:
619 return ReturnValue.Error
620 except Exception as e:
621 display_exception(e, show_traceback=obj['debug'])
622 return ReturnValue.Error
625@cli.group
626@click.pass_context
627def db(ctx: click.core.Context) -> None:
628 """
629 Advanced database commands.
631 The db group of commands offers a set of useful database operations. Invoke the help option of each command for
632 more details.
633 \f
635 :param ctx: The click context.
636 :type ctx: click.core.Context
637 """
640@db.command(name='wizard')
641@click.pass_context
642@click.option(
643 '-o',
644 '--output-file',
645 type=click.Path(),
646 default=pathlib.Path.cwd() / pathlib.Path('my_model.py'),
647 help='The name of the output file with the reflected model.',
648)
649@click.option('-s', '--schema', type=str, help='The name of the DB schema')
650@click.option(
651 '-t', '--tables', type=str, multiple=True, help='Generate model for selected tables. Multiple option possible.'
652)
653@click.option('--overwrite/--no-overwrite', default=True, help='Overwrite output file if already exists.')
654@click.option('--preserve-order/--no-preserve-order', default=True, help='Preserve column order.')
655@click.option('--with-views/--without-views', default=False, help='Include also database views.')
656@click.option('--ignore-unknown/--no-ignore-unknown', default=False, help='Ignore unknown fields.')
657@click.option('--snake-case/--no-snake-case', default=True, help='Use snake case for table and field names.')
658@click.option('--host', type=str, help='Hostname for the DB server.')
659@click.option('-p', '--port', type=int, help='Port number for the DB server.')
660@click.option('-u', '--user', '--username', type=str, help='Username for the connection to the DB server.')
661@click.option('--password', prompt=True, prompt_required=False, hide_input=True, help='Insert password when prompted')
662@click.option('-e', '--engine', type=click.Choice(sorted(DATABASE_MAP)), help='The DB engine')
663@click.argument('database', type=str)
664def wizard(
665 ctx: click.core.Context,
666 overwrite: bool,
667 tables: tuple[str, ...] | None,
668 preserve_order: bool,
669 with_views: bool,
670 ignore_unknown: bool,
671 snake_case: bool,
672 output_file: click.Path | pathlib.Path | str,
673 host: str,
674 port: int,
675 user: str,
676 password: str,
677 engine: str,
678 schema: str,
679 database: str,
680) -> ReturnValue:
681 """
682 Reflect an existing DB into a python module.
684 mafw db wizard [Options] Database
686 Database Name of the Database to be reflected.
688 About connection options (user / host / port):
690 That information will be used only in case you are trying to access a network database (MySQL or PostgreSQL). In
691 case of Sqlite, the parameters will be discarded.
693 About passwords:
695 If you need to specify a password to connect to the DB server, just add --password in the command line without
696 typing your password as clear text. You will be prompted to insert the password with hidden characters at the start
697 of the processor.
699 About engines:
701 The full list of supported engines is provided in the option below. If you do not specify any
702 engine and the database is actually an existing filename, then engine is set to Sqlite, otherwise to postgresql.
704 \f
706 :param database: The name of the database.
707 :type database: str
708 :param schema: The database schema to be reflected.
709 :type schema: str
710 :param engine: The database engine. A selection of possible values is provided in the script help.
711 :type engine: str
712 :param password: The password for the DB connection. Not used in case of Sqlite.
713 :type password: str
714 :param user: The username for the DB connection. Not used in case of Sqlite.
715 :type user: str
716 :param port: The port number of the database server. Not used in case of Sqlite.
717 :type port: int
718 :param host: The database hostname. Not used in case of Sqlite.
719 :type host: str
720 :param output_file: The filename for the output python module.
721 :type output_file: click.Path | pathlib.Path | str
722 :param snake_case: Flag to select snake_case convention for table and field names, or all small letter formatting.
723 :type snake_case: bool
724 :param ignore_unknown: Flag to ignore unknown fields. If False, an unknown field will be labelled with UnknownField.
725 :type ignore_unknown: bool
726 :param with_views: Flag to include views in the reflected elements.
727 :type with_views: bool
728 :param preserve_order: Flag to select if table fields should be reflected in the original order (True) or in
729 alphabetical order (False)
730 :type preserve_order: bool
731 :param tables: A tuple containing a selection of table names to be reflected.
732 :type tables: tuple[str, ...]
733 :param overwrite: Flag to overwrite the output file if exists. If False and the output file already exists, the
734 user can decide what to do.
735 :type overwrite: bool
736 :param ctx: The click context, that includes the original object with global options.
737 :type ctx: click.core.Context
738 :return: The script return value
739 """
740 obj = ctx.obj
742 if isinstance(output_file, (str, click.Path)): 742 ↛ 746line 742 didn't jump to line 746 because the condition on line 742 was always true
743 output_file = pathlib.Path(str(output_file))
745 # if not overwrite, check if the file exists
746 if not overwrite and output_file.exists():
747 answer = Prompt.ask(
748 f'A module ({output_file.name}) already exists. Do you want to overwrite, cancel or backup?',
749 case_sensitive=False,
750 choices=['o', 'c', 'b'],
751 show_choices=True,
752 show_default=True,
753 default='b',
754 )
755 if answer == 'c':
756 return ReturnValue.OK
757 elif answer == 'b': 757 ↛ 764line 757 didn't jump to line 764 because the condition on line 757 was always true
758 bck_filename = output_file.parent / pathlib.Path(
759 output_file.stem + f'_{datetime.datetime.now():%Y%m%dT%H%M%S}' + output_file.suffix
760 )
762 shutil.copy(output_file, bck_filename)
764 if tables == ():
765 tables = None
767 if engine is None:
768 engine = 'sqlite' if pathlib.Path(database).exists() else 'postgresql'
770 # prepare the connection options
771 if engine in ['sqlite', 'sqlite3']:
772 # for sqlite the connection
773 keys: list[str] = ['schema']
774 values: list[str | int] = [schema]
775 else:
776 keys = ['host', 'port', 'user', 'schema', 'password']
777 values = [host, port, user, schema, password]
779 connection_options: dict[str, Any] = {}
780 for k, v in zip(keys, values):
781 if v:
782 connection_options[k] = v
784 try:
785 introspector = make_introspector(engine, database, **connection_options)
786 except Exception as e:
787 msg = f'[red]Problem generating an introspector instance of {database}.'
788 display_exception(e, show_traceback=obj['debug'])
789 return ReturnValue.Error
791 try:
792 with open(output_file, 'w') as out_file:
793 dump_models(
794 out_file,
795 introspector,
796 tables,
797 preserve_order=preserve_order,
798 include_views=with_views,
799 ignore_unknown=ignore_unknown,
800 snake_case=snake_case,
801 )
802 except Exception as e:
803 display_exception(e, obj['debug'])
804 return ReturnValue.Error
806 msg = f'[green]Database {database} successfully reflected in {output_file.name}'
807 log.info(msg)
808 return ReturnValue.OK
811if __name__ == '__main__':
812 # Use the custom main method that handles exit codes
813 cli.main()