Coverage for src / sql_tool / cli / commands / service.py: 62%

1"""Service and maintenance CLI commands. 

2 

3Thin CLI layer: typer decorators, argument parsing, output formatting. 

4Business logic delegated to core.postgres module. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import Annotated 

10 

11import typer 

12 

13from sql_tool.cli.commands._shared import get_client, output_result 

14from sql_tool.core.postgres import ( 

15 check_server, 

16 kill_backend, 

17 list_user_tables, 

18 vacuum_tables, 

19) 

20 

21service_app = typer.Typer(help="Service and maintenance commands") 

22 

23 

24@service_app.callback(invoke_without_command=True) 

25def service_callback( 

26 ctx: typer.Context, 

27) -> None: 

28 if not ctx.invoked_subcommand: 

29 typer.echo(ctx.get_help()) 

30 raise typer.Exit() 

31 

32 

33@service_app.command("vacuum") 

34def vacuum_command( 

35 ctx: typer.Context, 

36 table_arg: Annotated[ 

37 str | None, 

38 typer.Argument(help="Table name to vacuum"), 

39 ] = None, 

40 full: Annotated[ 

41 bool, 

42 typer.Option("--full", help="Run VACUUM FULL (locks table, rewrites)"), 

43 ] = False, 

44 all_tables: Annotated[ 

45 bool, 

46 typer.Option("--all", help="Vacuum all user tables"), 

47 ] = False, 

48) -> None: 

49 """VACUUM marks dead rows as reusable. VACUUM FULL rewrites tables but requires exclusive lock.""" 

50 if not table_arg and not all_tables: 

51 typer.echo("Error: Must specify table name or --all", err=True) 

52 raise typer.Exit(2) 

53 

54 with get_client(ctx) as client: 

55 table_names = list_user_tables(client) if all_tables else [table_arg] # type: ignore[list-item] 

56 

57 count = vacuum_tables(client, table_names, full=full) 

58 

59 typer.echo(f"Vacuumed {count} table(s)") 

60 

61 

62@service_app.command("kill") 

63def kill_command( 

64 ctx: typer.Context, 

65 pid: Annotated[int, typer.Argument(help="Backend process ID to terminate")], 

66 cancel: Annotated[ 

67 bool, 

68 typer.Option("--cancel", help="Cancel query only (don't kill connection)"), 

69 ] = False, 

70) -> None: 

71 """pg_terminate_backend() kills the connection. pg_cancel_backend() cancels query but keeps connection alive.""" 

72 with get_client(ctx) as client: 

73 success = kill_backend(client, pid, cancel=cancel) 

74 

75 action = "Cancelled" if cancel else "Terminated" 

76 if success: 

77 typer.echo(f"{action} backend {pid}") 

78 else: 

79 typer.echo(f"Error: Backend {pid} not found or already terminated", err=True) 

80 raise typer.Exit(1) 

81 

82 

83@service_app.command("check") 

84def check_command(ctx: typer.Context) -> None: 

85 """ 

86 Check PostgreSQL server connectivity and version. 

87 

88 Reports server version, current database, connected user, and uptime. 

89 Useful for verifying connection parameters and server availability. 

90 """ 

91 with get_client(ctx) as client: 

92 result = check_server(client) 

93 

94 output_result(ctx, result)