Coverage for src\sql_organizer\main.py: 86%

1from pathlib import Path 

2from typing import Annotated 

3 

4import rich 

5import typer 

6from pydantic import ValidationError 

7 

8from sql_organizer.combiner import combine_files 

9from sql_organizer.file import SqlFile 

10from sql_organizer.file_formatter import get_standard_formatter 

11from sql_organizer.search import FileExtension, get_all_sql_files 

12from sql_organizer.sorter import SORTERS, sort_paths 

13 

14app = typer.Typer() 

15 

16 

17@app.command() 

18def combine( 

19 path: Annotated[ 

20 Path, typer.Argument(help="path, where the SQL files are located") 

21 ] = Path("."), 

22 extension: Annotated[list[str], typer.Option("--extension", "-e")] = ["sql"], 

23 sorters: Annotated[ 

24 list[str], 

25 typer.Option( 

26 "--sorter", "-so", help=f"Available sorters: {', '.join(SORTERS)}" 

27 ), 

28 ] = [ 

29 "folder", 

30 "first_number", 

31 ], 

32 skip_errors: bool = False, 

33 uncomment_use: Annotated[bool, typer.Option("--uncomment_use", "-uu")] = False, 

34 target: Annotated[Path, typer.Option("--target", "-t")] = Path("./target.sql"), 

35 overwrite: bool = False, 

36) -> None: 

37 for sorter in sorters: 

38 if sorter not in SORTERS: 

39 rich.print( 

40 f"[bold red]Value Error![/bold red] sorter [bold blue]{sorter}\ 

41[/bold blue] is invalid" 

42 ) 

43 return 

44 if not overwrite and target.exists(): 

45 rich.print("[bold red]OS Error![/bold red] Target already exists!") 

46 return 

47 try: 

48 extensions = [FileExtension(extension=e) for e in extension] 

49 except ValidationError: 

50 rich.print( 

51 "[bold red]Value Error![/bold red] Extension should not be an empty string" 

52 ) 

53 return 

54 files = get_all_sql_files(path, extensions) 

55 if len(files) == 0: 

56 rich.print( 

57 f"[yellow]No files with [bold blue]\ 

58{', '.join(e.extension for e in extensions)}\ 

59[/bold blue] extension{'s' if len(extensions) > 1 else ''} found![/yellow]" 

60 ) 

61 return 

62 sorted_files = sort_paths(files, [SORTERS[s] for s in sorters]) 

63 sql_files = [] 

64 for file in sorted_files: 

65 sql_file = SqlFile.from_path(file) 

66 if isinstance(sql_file, OSError): 

67 color = "yellow" if skip_errors else "red" 

68 rich.print( 

69 f"[bold {color}]OS Error[/bold {color}] Could not open file {file.absolute()}" 

70 ) 

71 if not skip_errors: 

72 return 

73 sql_files.append(sql_file) 

74 

75 formatter = get_standard_formatter(uncomment_use) 

76 combine_files(target, sql_files, formatter) 

77 rich.print("[bold green]Success![/bold green]")