Coverage for src / sql_tool / cli / commands / query.py: 100%

1from __future__ import annotations 

2 

3from typing import Annotated 

4 

5import structlog 

6import typer 

7 

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

9from sql_tool.core.query_source import resolve_query_source 

10 

11log = structlog.get_logger() 

12 

13 

14def query_command( 

15 ctx: typer.Context, 

16 file: Annotated[ 

17 str | None, 

18 typer.Argument(help="SQL file to execute"), 

19 ] = None, 

20 execute: Annotated[ 

21 str | None, 

22 typer.Option("--execute", "-e", help="Execute inline SQL query"), 

23 ] = None, 

24 timeout: Annotated[ 

25 float | None, 

26 typer.Option("--timeout", "-t", help="Query timeout in seconds"), 

27 ] = None, 

28) -> None: 

29 """Execute a SQL query from file, inline (-e), or stdin.""" 

30 sql = resolve_query_source(inline=execute, file_path=file) 

31 

32 with get_client(ctx, timeout=timeout) as client: 

33 result = client.execute_query(sql) 

34 

35 output_result(ctx, result)