diff --git a/docs/cli.rst b/docs/cli.rst index 8baa194..013d323 100644 --- a/docs/cli.rst +++ b/docs/cli.rst @@ -21,6 +21,11 @@ This is the default command for ``sqlite-utils``, so you can instead use this:: $ sqlite-utils dogs.db "select * from dogs" +You can pass named parameters to the query using ``-p``:: + + $ sqlite-utils query dogs.db "select :num * :num2" -p num 5 -p num2 6 + [{":num * :num2": 30}] + Use ``--nl`` to get back newline-delimited JSON objects:: $ sqlite-utils dogs.db "select * from dogs" --nl diff --git a/sqlite_utils/cli.py b/sqlite_utils/cli.py index f387602..155ea54 100644 --- a/sqlite_utils/cli.py +++ b/sqlite_utils/cli.py @@ -677,11 +677,18 @@ def drop_view(path, view): @click.argument("sql") @output_options @click.option("-r", "--raw", is_flag=True, help="Raw output, first column of first row") -def query(path, sql, nl, arrays, csv, no_headers, table, fmt, json_cols, raw): +@click.option( + "-p", + "--param", + multiple=True, + type=(str, str), + help="Named :parameters for SQL query", +) +def query(path, sql, nl, arrays, csv, no_headers, table, fmt, json_cols, raw, param): "Execute SQL query and return the results as JSON" db = sqlite_utils.Database(path) with db.conn: - cursor = db.conn.execute(sql) + cursor = db.conn.execute(sql, dict(param)) if cursor.description is None: # This was an update/insert headers = ["rows_affected"] diff --git a/tests/test_cli.py b/tests/test_cli.py index 3584840..4012c97 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -789,6 +789,27 @@ def test_query_json_binary(db_path): ] +@pytest.mark.parametrize( + "sql,params,expected", + [ + ("select 1 + 1 as out", {"p": "2"}, 2), + ("select 1 + :p as out", {"p": "2"}, 3), + ( + "select :hello as out", + {"hello": """This"has'many'quote"s"""}, + """This"has'many'quote"s""", + ), + ], +) +def test_query_params(db_path, sql, params, expected): + extra_args = [] + for key, value in params.items(): + extra_args.extend(["-p", key, value]) + result = CliRunner().invoke(cli.cli, [db_path, sql] + extra_args) + assert result.exit_code == 0, str(result) + assert json.loads(result.output.strip()) == [{"out": expected}] + + def test_query_json_with_json_cols(db_path): db = Database(db_path) with db.conn: