feat(alerts): per-connection and per-database alert overrides

sqlit/cli.py

@@ -58,7 +58,7 @@ def _extract_project_dir(argv: list[str]) -> tuple[Path | None, list[str]]:
     Returns (project_dir, remaining_argv). Exits with an error if the
     user passed a path that doesn't resolve to an existing directory.
     """
-    subcommands = {"connections", "connection", "connect", "query", "docker"}
+    subcommands = {"connections", "connection", "connect", "query", "docker", "alerts"}
     result_argv: list[str] = []
     project_dir: Path | None = None
 
@@ -587,6 +587,11 @@ def main() -> int:
         add_schema_arguments(provider_parser, schema, include_name=True, name_required=True)
         provider_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password")
         provider_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password")
+        provider_parser.add_argument(
+            "--alert",
+            metavar="MODE",
+            help="Per-connection query alert mode: off|delete|write",
+        )
 
     edit_parser = conn_subparsers.add_parser("edit", help="Edit an existing connection")
     edit_parser.add_argument("connection_name", help="Name of connection to edit")
@@ -606,6 +611,11 @@ def main() -> int:
     edit_parser.add_argument("--file-path", help="Database file path (SQLite only)")
     edit_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password")
     edit_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password")
+    edit_parser.add_argument(
+        "--alert",
+        metavar="MODE",
+        help="Per-connection query alert mode: off|delete|write|unset (clears the override)",
+    )
 
     delete_parser = conn_subparsers.add_parser("delete", help="Delete a connection")
     delete_parser.add_argument("connection_name", help="Name of connection to delete")
@@ -622,6 +632,11 @@ def main() -> int:
         add_schema_arguments(provider_parser, schema, include_name=True, name_required=False)
         provider_parser.add_argument("--password-command", dest="password_command", help="Shell command to retrieve the database password")
         provider_parser.add_argument("--ssh-password-command", dest="ssh_password_command", help="Shell command to retrieve the SSH password")
+        provider_parser.add_argument(
+            "--alert",
+            metavar="MODE",
+            help="Per-connection query alert mode: off|delete|write",
+        )
 
     query_parser = subparsers.add_parser("query", help="Execute a SQL query")
     query_parser.add_argument("--connection", "-c", required=True, help="Connection name to use")
@@ -648,6 +663,44 @@ def main() -> int:
     docker_subparsers = docker_parser.add_subparsers(dest="docker_command", help="Docker commands")
     docker_subparsers.add_parser("list", help="List detected database containers")
 
+    # Query alert management
+    alerts_parser = subparsers.add_parser(
+        "alerts",
+        help="Manage query confirmation alert overrides",
+        description=(
+            "Manage query alert overrides. Hierarchy: database > connection > global."
+        ),
+    )
+    alerts_subparsers = alerts_parser.add_subparsers(dest="alerts_command", help="Alert commands")
+    alerts_subparsers.add_parser("list", help="Show global and all configured overrides")
+
+    alerts_set = alerts_subparsers.add_parser("set", help="Set an alert mode at a scope")
+    alerts_set.add_argument("mode", help="off | delete | write")
+    alerts_set.add_argument(
+        "--connection",
+        "-c",
+        help="Target a specific saved connection (omit for global)",
+    )
+    alerts_set.add_argument(
+        "--database",
+        "-d",
+        help="Target a database on the connection (requires --connection)",
+    )
+
+    alerts_unset = alerts_subparsers.add_parser(
+        "unset", help="Clear an alert override (falls back to the next scope)"
+    )
+    alerts_unset.add_argument(
+        "--connection",
+        "-c",
+        help="Connection whose override to clear",
+    )
+    alerts_unset.add_argument(
+        "--database",
+        "-d",
+        help="Database whose override to clear (requires --connection)",
+    )
+
     log_startup_step("cli_parser_end")
 
     with startup_span("cli_parse_args"):
@@ -758,6 +811,13 @@ def main() -> int:
             print(f"Error: {exc}")
             return 1
 
+        from sqlit.domains.connections.cli.commands import _apply_alert_option
+
+        alert_error = _apply_alert_option(temp_config, getattr(args, "alert", None))
+        if alert_error:
+            print(f"Error: {alert_error}")
+            return 1
+
         app = SSMSTUI(services=services, startup_connection=temp_config)
         return _run_app(app)
 
@@ -786,6 +846,22 @@ def main() -> int:
             docker_parser.print_help()
             return 1
 
+    if args.command == "alerts":
+        from sqlit.domains.query.cli.alerts_commands import (
+            cmd_alerts_list,
+            cmd_alerts_set,
+            cmd_alerts_unset,
+        )
+
+        if args.alerts_command == "list":
+            return cmd_alerts_list(args, services=services)
+        if args.alerts_command == "set":
+            return cmd_alerts_set(args, services=services)
+        if args.alerts_command == "unset":
+            return cmd_alerts_unset(args, services=services)
+        alerts_parser.print_help()
+        return 1
+
     parser.print_help()
     return 1
 

sqlit/domains/connections/cli/commands.py

@@ -19,12 +19,37 @@
     get_database_type_labels,
 )
 from sqlit.domains.connections.providers.catalog import get_provider_schema
+from sqlit.domains.query.app.alerts import (
+    CONNECTION_ALERT_OPTION,
+    format_alert_mode,
+    parse_alert_mode,
+)
 from sqlit.shared.app.runtime import RuntimeConfig
 from sqlit.shared.app.services import AppServices, build_app_services
 
 from .helpers import build_connection_config_from_args
 
 
+def _apply_alert_option(config: ConnectionConfig, raw: str | None) -> str | None:
+    """Apply an --alert value (mode token or "unset") to a connection.
+
+    Returns an error message on invalid input, otherwise None.
+    """
+    if raw is None:
+        return None
+    token = raw.strip().lower()
+    if not token:
+        return None
+    if token in {"unset", "clear", "default"}:
+        config.options.pop(CONNECTION_ALERT_OPTION, None)
+        return None
+    mode = parse_alert_mode(token)
+    if mode is None:
+        return f"Invalid --alert value '{raw}'. Use off|delete|write|unset."
+    config.set_option(CONNECTION_ALERT_OPTION, format_alert_mode(mode))
+    return None
+
+
 def _find_connection_index(connections: list[ConnectionConfig], name: str) -> int | None:
     for idx, conn in enumerate(connections):
         if conn.name == name:
@@ -204,6 +229,11 @@ def cmd_connection_create(args: Any, *, services: AppServices | None = None) ->
         print(f"Error: {exc}")
         return 1
 
+    alert_error = _apply_alert_option(config, getattr(args, "alert", None))
+    if alert_error:
+        print(f"Error: {alert_error}")
+        return 1
+
     connections.append(config)
     _ensure_password_storage(services, config)
     _save_connections(services, connections)
@@ -270,6 +300,11 @@ def cmd_connection_edit(args: Any, *, services: AppServices | None = None) -> in
 
             conn.endpoint = FileEndpoint(path=file_path)
 
+    alert_error = _apply_alert_option(conn, getattr(args, "alert", None))
+    if alert_error:
+        print(f"Error: {alert_error}")
+        return 1
+
     _ensure_password_storage(services, conn)
 
     _save_connections(services, connections)

sqlit/domains/query/app/alerts.py

@@ -4,10 +4,17 @@
 
 import re
 from enum import IntEnum
+from typing import Any
 
 from sqlit.domains.query.app.multi_statement import split_statements
 
 
+# Settings / option keys for the alert mode hierarchy.
+GLOBAL_ALERT_SETTING = "query_alert_mode"
+DATABASE_ALERT_SETTING = "query_alert_modes_by_database"
+CONNECTION_ALERT_OPTION = "query_alert_mode"
+
+
 class AlertMode(IntEnum):
     """Alert mode thresholds (user-configured)."""
 
@@ -129,3 +136,43 @@ def _coerce_alert_mode(value: int) -> AlertMode | None:
     if mode in {AlertMode.OFF, AlertMode.DELETE, AlertMode.WRITE}:
         return mode
     return None
+
+
+def make_db_alert_key(connection_name: str, database: str) -> str:
+    """Compose the key used to store per-database alert overrides."""
+    return f"{connection_name}::{database}"
+
+
+def resolve_alert_mode(
+    *,
+    global_mode: Any = None,
+    connection_option: Any = None,
+    database_override: Any = None,
+) -> tuple[AlertMode, str]:
+    """Resolve the effective alert mode given the three scopes.
+
+    Returns a (mode, source) tuple where source is one of
+    ``"database"``, ``"connection"``, or ``"global"`` — describing which
+    scope supplied the resolved value.
+    """
+    mode = parse_alert_mode(database_override)
+    if mode is not None:
+        return mode, "database"
+    mode = parse_alert_mode(connection_option)
+    if mode is not None:
+        return mode, "connection"
+    mode = parse_alert_mode(global_mode)
+    if mode is not None:
+        return mode, "global"
+    return AlertMode.OFF, "global"
+
+
+def lookup_database_override(
+    overrides: Any,
+    connection_name: str | None,
+    database: str | None,
+) -> Any:
+    """Look up a per-database alert override from a stored mapping."""
+    if not connection_name or not database or not isinstance(overrides, dict):
+        return None
+    return overrides.get(make_db_alert_key(connection_name, database))