Merge pull request #4 from gitmobkab/cli_refactor

merge of the Cli refactor

Author: gitmobkab

CHANGELOG.md

@@ -9,11 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- security improvements against sql injections
 - official postgres adapter
 - unique keyword and generated values
+- a `copia run` command
 
 ### Changed
 
+- the tui can now be launch with `copia tui` rather than `copia`
 - the dsl syntax now explicitly require parantheses after the name of a generator
 - the `ref` generator is now called `fetch` as the name is more intuitive
 - the cli logs messages appereance

src/copia/adapters/mysql_adapter.py

@@ -69,11 +69,16 @@ def insert(self, table: str, rows: Sequence[dict[str, Any]], batch_size: int = 2
         query = f"INSERT INTO {table} ({columns_query}) VALUES ({placeholders})"
 
         iterator = iter(rows)
-        with self._connection.cursor() as cursor:
-            while batch := list(islice(iterator, batch_size)):
-                coerced_batch = list(map(self._coerce_row, batch))
-                cursor.executemany(query, coerced_batch)
-        self._connection.commit()
+        
+        try:
+            with self._connection.cursor() as cursor:
+                while batch := list(islice(iterator, batch_size)):
+                    coerced_batch = list(map(self._coerce_row, batch))
+                    cursor.executemany(query, coerced_batch)
+            self._connection.commit()
+        except Exception as err:
+            self._connection.rollback()
+            raise err
 
     def _coerce_row(self, row: dict[str, Any]) -> dict[str, Any]:
         coerced_row: dict[str, Any] = {}

src/copia/adapters/postgres_adapter.py

@@ -70,13 +70,16 @@ def insert(self, table: str, rows: Sequence[dict[str, Any]], batch_size: int = 2
             self.escape_columns(columns),
             SQL(", ").join(placeholders)
         )
-    
-        iterator = iter(rows)
-        with self._connection.cursor() as cursor:
-            while batch := list(islice(iterator, batch_size)):
-                cursor.executemany(composed_query, batch)
-        self._connection.commit()
-        
+
+        try:
+            iterator = iter(rows)
+            with self._connection.cursor() as cursor:
+                while batch := list(islice(iterator, batch_size)):
+                    cursor.executemany(composed_query, batch)
+            self._connection.commit()
+        except Exception as err:
+            self._connection.rollback()
+            raise err       
 
     def close(self) -> None:
         self._connection.close()

src/copia/cli/app.py

@@ -1,119 +1,51 @@
 import typer
-
+from typing import get_args
 from copia._version import __version__
-from .commands import list_app, init_command
-from .exit_codes import ExitCodes
-from copia.adapters import get_adapter
-from ..tui import CopiaApp
-from copia.cli.console_utils import print_error, info, success, error, echo
-from .config import (
-    get_profile,
-    LOCAL_COPIA_FILE,
-    GLOBAL_COPIA_FILE
+from .commands import (
+    list_command,
+    init_command,
+    tui_command,
+    run_command
+)
+from .config import Adapter
+from .utils import echo
+
+
+AVAILABLES_ADAPTERS: tuple = get_args(Adapter)
+
+def get_help_msg():
+    help_msg = "An Interactive TUI app for Databases seeding"
+    help_msg += "\n\nSupported databases:"
+    for value in AVAILABLES_ADAPTERS:
+        help_msg += f'\n- {value}'
+    return help_msg
+
+app = typer.Typer(
+    name='copia',
+    help=get_help_msg(),
+    no_args_is_help=True,
+    invoke_without_command=True,
+    context_settings={
+        'help_option_names': ['--help', '-h']
+    }
 )
-
-app = typer.Typer(invoke_without_command=True)
 
 app.add_typer(init_command)
-app.add_typer(list_app)
+app.add_typer(list_command)
+app.add_typer(tui_command)
+app.add_typer(run_command)
 
 @app.callback()
 def main(
     ctx: typer.Context,
-    profile_name: str = typer.Option("default",
-                                "-p", "--profile",
-                                help="The profile to use for the session"),
+
     version_flag: bool = typer.Option(False, 
                                  "-v", "--version",
                                  help="Display the current version and exit"),
-    
-    help_flag: bool = typer.Option(False, 
-                              "-h", "--help",
-                              help="Display this message and exit"),
-    
-    search_globals_only: bool = typer.Option(False,
-                                            "-g", "--global",
-                                            help=f"Search only in [green]'{GLOBAL_COPIA_FILE}'"),
-    search_locals_only: bool = typer.Option(False,
-                                            "-l", "--local",
-                                            help=f"Search only in [green]'{LOCAL_COPIA_FILE}'"),
-         ):
-    
-    """An Interactive TUI app for MySQL Database seeding"""
-    
+    ):    
     if ctx.invoked_subcommand is not None:
         return
-    
-    if help_flag:
-        echo(ctx.get_help())
-        raise typer.Exit()
-        
+            
     if version_flag:
         echo(f"[blue]copia {__version__}")
-        raise typer.Exit()
-    
-    profile = None
-    try:
-        if search_globals_only and search_locals_only:
-            error("Cannot use --global | -g and --local | -l at the same time")
-            raise typer.Exit()
-
-        if search_globals_only:
-            profile = get_profile(profile_name, "global")
-        elif search_locals_only:
-            profile = get_profile(profile_name, "local")
-        else:
-            profile = get_any_profile(profile_name)
-    except Exception as err:
-        print_error(err)
-        raise typer.Exit()
-
-
-    info(f"Found profile: {profile}")
-    info("Connecting to db...")
-    
-    try:
-        adapter = get_adapter(profile)
-        success("Connection to db successfull")
-    except ImportError:
-        print_error(f"Missing dependencies for adapter {profile.adapter!r}",
-                    f'Try "pip install copia-seed\\[{profile.adapter}]"')
-        raise typer.Exit(ExitCodes.RESOURCE_ERROR)
-    except Exception as connection_err:
-        print_error(connection_err)
-        raise typer.Exit(ExitCodes.CONNEXION_TO_DB_FAILED)
-
-    CopiaApp(adapter).run()
-    success("Bye.")
-        
-            
-def get_any_profile(profile_name: str):
-    """
-    Try to fetch the profile from the local config file,
-    and fall back to global on any fail.
-    Any exception from local fetching get silently swallowed
-    
-    All the exceptions raised by this function are always in the global fetch state
-    
-    Args:
-        profile_name (str): the name by which the profile is identified
-
-    Raises:
-        FileNotFoundError: when the file couldn't be found.
-        PermissionError: when permission to read the file had been denied by the os.
-        InvalidConfigError: when the config isn't valid TOML or the value of the "profiles" key is not a dict/TOML Table
-        ProfileNotFoundError: when the "profiles.profile_name" key don't exist, this also includes missing "profiles" key
-        FoundProfileIsNotATableError: when the "profiles.profile_name" key exist but is not a dict
-        InvalidProfileError: when the profile isn't valid
-        
-    Returns:
-        Profile: a valid Profile object
-    """
-    try:
-        info(f"Fetching profiles.\"{profile_name}\" from local config...")
-        return get_profile(profile_name, "local")
-    except Exception as err:
-        error(f"{err}")
-        info("Falling back to global config...")
-        info(f"Fetching profiles.\"{profile_name}\" from global config...")
-        return get_profile(profile_name, "global")
+        raise typer.Exit()

src/copia/cli/commands/__init__.py

@@ -1,2 +1,4 @@
-from .list import list_profile_app as list_app
-from .init import init_command
+from .list import list_command
+from .init import init_command
+from .tui import tui_command
+from .run import run_command

src/copia/cli/commands/init.py

@@ -4,8 +4,8 @@
 from rich.prompt import Confirm
 import typer
 
-from copia.cli.exit_codes import ExitCodes
-from copia.cli.console_utils import echo, print_error, info, warning, success
+from copia.cli.utils.exit_codes import ExitCodes
+from copia.cli.utils.console_utils import echo, print_error, info, warning, success
 from ..config.loaders import (
     resolve_config_path
 )
@@ -19,9 +19,9 @@ def main(ctx: typer.Context,
         is_global_config: bool = typer.Option(
     False, "-g", "--global", help="create the config file as a global config"
 )):
-    """[blue]generate a template config file for copia[/]
+    """Generate a template config file for copia.
     
-    by default it create a local scoped config file
+    By default it create a local scoped config file
     """
 
     if help_flag:

src/copia/cli/commands/list.py

@@ -1,27 +1,26 @@
 import typer
-from copia.cli.console_utils import echo
+from copia.cli.utils.console_utils import echo
 
 from ..config.loaders import load_config, get_profile_from_config
 from ..config.exceptions import InvalidProfileError
 
-list_profile_app = typer.Typer()
+list_command = typer.Typer()
 
-@list_profile_app.command("list")
-def list_profiles(
-    ctx: typer.Context,
-    help_flag: bool = 
-                typer.Option(False, "-h", "--help", help="Display this message and exit")):
+# now that i look at it, it's horrible
+# better refactor that shit soon
+
+# TODO: FUCKING REFACTOR THAT, IT'S UGLY JUST BY SCROLLING
+# I DON'T WANNA THINK ABOUT WHY I WROTE IT LIKE THAT 
+
+@list_command.command("list")
+def main():
 
-    """[blue]list all profiles defined in both global and local config files"""
+    """List all profiles defined in both global and local config files."""
     local_config = None
     global_config = None
     local_failure_reason = None
     global_failure_reason = None
 
-    if help_flag:
-        echo(ctx.get_help())
-        return
-    
     try:
         local_config = load_config("local")
     except Exception as err: