Refactor adapters - make it easier to make new ones

Author: Maxteabag

sqlit/cli.py

@@ -387,11 +387,10 @@ def _build_temp_connection(args: argparse.Namespace) -> ConnectionConfig | None:
     """Build a temporary connection config from CLI args, if provided."""
     db_type = getattr(args, "db_type", None)
     file_path = getattr(args, "file_path", None)
-    if not db_type and file_path:
-        db_type = "sqlite"
-        setattr(args, "db_type", db_type)
     if not db_type:
-        if any(getattr(args, name, None) for name in ("file_path", "server", "host", "database")):
+        if file_path:
+            raise ValueError("--db-type is required when using --file-path")
+        if any(getattr(args, name, None) for name in ("server", "host", "database")):
             raise ValueError("--db-type is required for temporary connections")
         return None
 

sqlit/cli_helpers.py

@@ -3,33 +3,24 @@
 from __future__ import annotations
 
 import argparse
+from dataclasses import fields
+from functools import lru_cache
 from typing import Any, Iterable
 
 from .config import ConnectionConfig
 from .db.schema import ConnectionSchema, FieldType
 
-CONNECTION_ARG_NAMES = {
-    "name",
-    "server",
-    "host",
-    "port",
-    "database",
-    "username",
-    "password",
-    "file_path",
-    "auth_type",
-    "supabase_region",
-    "supabase_project_id",
-    "ssh_enabled",
-    "ssh_host",
-    "ssh_port",
-    "ssh_username",
-    "ssh_auth_type",
-    "ssh_key_path",
-    "ssh_password",
-    "driver",
-    "oracle_role",
-}
+@lru_cache(maxsize=1)
+def _get_connection_arg_names() -> set[str]:
+    from .db.schema import get_all_schemas
+
+    names = {
+        field.name
+        for schema in get_all_schemas().values()
+        for field in schema.fields
+    }
+    names.add("host")
+    return names
 
 
 def add_schema_arguments(
@@ -85,6 +76,8 @@ def build_connection_config_from_args(
     strict: bool = True,
 ) -> ConnectionConfig:
     """Build a ConnectionConfig from CLI args based on a provider schema."""
+    from .db.providers import normalize_connection_config
+
     raw_values = _extract_raw_values(schema, args)
 
     missing = _find_missing_required_fields(schema, raw_values)
@@ -99,10 +92,12 @@ def build_connection_config_from_args(
             raise ValueError(f"Unexpected fields for {schema.display_name}: {extras_args}")
 
     config_name = name or default_name or f"Temp {schema.display_name}"
-    config_values = {
+    config_values: dict[str, Any] = {
         "name": config_name,
         "db_type": schema.db_type,
     }
+    options: dict[str, Any] = {}
+    base_fields = {f.name for f in fields(ConnectionConfig)}
 
     # Fields where None means "not set" vs "" means "explicitly empty"
     nullable_fields = {"password", "ssh_password"}
@@ -113,21 +108,27 @@ def build_connection_config_from_args(
             value = ""
         if field.name == "ssh_enabled":
             if isinstance(value, bool):
-                config_values[field.name] = value
+                normalized = value
             else:
-                config_values[field.name] = str(value).lower() == "enabled"
-        else:
+                normalized = str(value).lower() == "enabled"
+            config_values[field.name] = normalized
+            continue
+
+        if field.name in base_fields:
             config_values[field.name] = value
+        else:
+            options[field.name] = value
 
     if "port" in config_values and not config_values["port"]:
         config_values["port"] = schema.default_port or ""
 
     if schema.has_advanced_auth:
-        auth_type = config_values.get("auth_type") or "sql"
-        config_values["auth_type"] = auth_type
-        config_values["trusted_connection"] = auth_type == "windows"
+        auth_type = options.get("auth_type") or "sql"
+        options["auth_type"] = auth_type
+        options["trusted_connection"] = auth_type == "windows"
 
-    return ConnectionConfig(**config_values)
+    config_values["options"] = options
+    return normalize_connection_config(ConnectionConfig(**config_values))
 
 
 def _extract_raw_values(schema: ConnectionSchema, args: Any) -> dict[str, Any]:
@@ -158,7 +159,7 @@ def _find_missing_required_fields(schema: ConnectionSchema, raw_values: dict[str
 def _find_unexpected_fields(schema: ConnectionSchema, args: Any) -> list[str]:
     allowed = {field.name for field in schema.fields}
     extras: list[str] = []
-    for field in CONNECTION_ARG_NAMES:
+    for field in _get_connection_arg_names():
         if field in allowed or field == "name":
             continue
         value = getattr(args, field, None)

sqlit/commands.py

@@ -20,7 +20,7 @@
     load_connections,
     save_connections,
 )
-from .db.providers import get_connection_schema, has_advanced_auth, is_file_based
+from .db.providers import get_adapter_class, get_connection_schema, has_advanced_auth, is_file_based
 from .services import ConnectionSession, QueryResult, QueryService
 from .services.credentials import (
     ALLOW_PLAINTEXT_CREDENTIALS_SETTING,
@@ -101,12 +101,16 @@ def cmd_connection_list(args: Any) -> int:
     for conn in connections:
         db_type_label = labels.get(conn.get_db_type(), conn.db_type)
         if is_file_based(conn.db_type):
-            conn_info = conn.file_path[:38] + ".." if len(conn.file_path) > 40 else conn.file_path
+            file_path = str(conn.get_option("file_path", ""))
+            conn_info = file_path[:38] + ".." if len(file_path) > 40 else file_path
             auth_label = "N/A"
         elif has_advanced_auth(conn.db_type):
             conn_info = f"{conn.server}@{conn.database}" if conn.database else conn.server
             conn_info = conn_info[:38] + ".." if len(conn_info) > 40 else conn_info
-            auth_label = AUTH_TYPE_LABELS.get(conn.get_auth_type(), conn.auth_type)
+            auth_value = str(conn.get_option("auth_type", ""))
+            adapter_class = get_adapter_class(conn.db_type)
+            auth_type = adapter_class().get_auth_type(conn)
+            auth_label = AUTH_TYPE_LABELS.get(auth_type, auth_value) if auth_type else auth_value
         else:
             # Server-based databases with simple auth
             conn_info = f"{conn.server}@{conn.database}" if conn.database else conn.server
@@ -226,8 +230,8 @@ def cmd_connection_edit(args: Any) -> int:
     if args.auth_type:
         try:
             auth_type = AuthType(args.auth_type)
-            conn.auth_type = auth_type.value
-            conn.trusted_connection = auth_type == AuthType.WINDOWS
+            conn.set_option("auth_type", auth_type.value)
+            conn.set_option("trusted_connection", auth_type == AuthType.WINDOWS)
         except ValueError:
             valid_types = ", ".join(t.value for t in AuthType)
             print(f"Error: Invalid auth type '{args.auth_type}'. Valid types: {valid_types}")
@@ -239,7 +243,7 @@ def cmd_connection_edit(args: Any) -> int:
 
     file_path = getattr(args, "file_path", None)
     if file_path is not None:
-        conn.file_path = file_path
+        conn.set_option("file_path", file_path)
 
     if (conn.password or conn.ssh_password) and not is_keyring_usable():
         if not _maybe_prompt_plaintext_credentials():
@@ -371,8 +375,9 @@ def cmd_query(
         print(f"Error: Connection '{args.connection}' not found.")
         return 1
 
-    if args.database and config.db_type == "mssql":
-        config = replace(config, database=args.database)
+    if args.database:
+        adapter_class = get_adapter_class(config.db_type)
+        config = adapter_class().apply_database_override(config, args.database)
 
     config = _prompt_for_password(config)
 

sqlit/config.py

@@ -10,9 +10,9 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 from enum import Enum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Mapping
 
 # Only import what's needed to create the DatabaseType enum
 from .db.providers import get_supported_db_types as _get_supported_db_types
@@ -160,12 +160,6 @@ class ConnectionConfig:
     database: str = ""
     username: str = ""
     password: str | None = None
-    # SQL Server specific fields
-    auth_type: str = "sql"
-    driver: str = field(default_factory=_get_default_driver)
-    trusted_connection: bool = False
-    # SQLite specific fields
-    file_path: str = ""
     # SSH tunnel fields
     ssh_enabled: bool = False
     ssh_host: str = ""
@@ -174,35 +168,55 @@ class ConnectionConfig:
     ssh_auth_type: str = "key"  # "key" or "password"
     ssh_password: str | None = None
     ssh_key_path: str = ""
-    # Supabase specific fields
-    supabase_region: str = ""
-    supabase_project_id: str = ""
-    # Oracle specific fields
-    oracle_role: str = "normal"  # "normal", "sysdba", "sysoper"
     # Source tracking (e.g., "docker" for auto-detected containers)
     source: str | None = None
     # Original connection URL if created from URL
     connection_url: str | None = None
     # Extra options from URL query parameters (e.g., sslmode=require)
     extra_options: dict[str, str] = field(default_factory=dict)
-
-    def __post_init__(self) -> None:
-        """Handle backwards compatibility with old configs."""
-        # Old configs without db_type are SQL Server
-        if not hasattr(self, "db_type") or not self.db_type:
-            self.db_type = "mssql"
-
-        # Apply default port for server-based DBs if missing (lazy import)
-        if not getattr(self, "port", None):
-            from .db.providers import get_default_port
-            default_port = get_default_port(self.db_type)
-            if default_port:
-                self.port = default_port
-
-        # Handle old SQL Server auth compatibility
-        if self.db_type == "mssql":
-            if self.auth_type == "windows" and not self.trusted_connection and self.username:
-                self.auth_type = "sql"
+    # Provider-specific options (auth_type, driver, file_path, etc.)
+    options: dict[str, Any] = field(default_factory=dict)
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> ConnectionConfig:
+        """Create a ConnectionConfig from a dict, with legacy key support."""
+        payload = dict(data)
+
+        if "host" in payload and "server" not in payload:
+            payload["server"] = payload.pop("host")
+
+        db_type = payload.get("db_type")
+        if not isinstance(db_type, str) or not db_type:
+            payload["db_type"] = "mssql"
+
+        raw_options = payload.pop("options", None)
+        options: dict[str, Any] = {}
+        if isinstance(raw_options, dict):
+            options.update(raw_options)
+
+        base_fields = {f.name for f in fields(cls)}
+        for key in list(payload.keys()):
+            if key in base_fields:
+                continue
+            if key not in options:
+                options[key] = payload.pop(key)
+            else:
+                payload.pop(key)
+
+        payload["options"] = options
+        return cls(**payload)
+
+    def get_option(self, name: str, default: Any | None = None) -> Any:
+        return self.options.get(name, default)
+
+    def set_option(self, name: str, value: Any) -> None:
+        self.options[name] = value
+
+    def get_field_value(self, name: str, default: Any = "") -> Any:
+        if name in self.__dataclass_fields__:
+            value = getattr(self, name)
+            return value if value is not None else default
+        return self.options.get(name, default)
 
     def get_db_type(self) -> DatabaseType:
         """Get the DatabaseType enum value."""
@@ -211,71 +225,6 @@ def get_db_type(self) -> DatabaseType:
         except ValueError:
             return DatabaseType.MSSQL  # type: ignore[attr-defined, no-any-return]
 
-    def get_auth_type(self) -> AuthType:
-        """Get the AuthType enum value."""
-        try:
-            return AuthType(self.auth_type)
-        except ValueError:
-            return AuthType.SQL_SERVER
-
-    def get_connection_string(self) -> str:
-        """Build the connection string for SQL Server.
-
-        .. deprecated::
-            This method is deprecated. Connection string building is now
-            handled internally by SQLServerAdapter._build_connection_string().
-            Use SQLServerAdapter.connect() directly instead.
-        """
-        import warnings
-
-        warnings.warn(
-            "ConnectionConfig.get_connection_string() is deprecated. "
-            "Connection string building is now handled internally by SQLServerAdapter.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-
-        if self.db_type != "mssql":
-            raise ValueError("get_connection_string() is only for SQL Server connections")
-
-        server_with_port = self.server
-        if self.port and self.port != "1433":
-            server_with_port = f"{self.server},{self.port}"
-
-        base = (
-            f"DRIVER={{{self.driver}}};"
-            f"SERVER={server_with_port};"
-            f"DATABASE={self.database or 'master'};"
-            f"TrustServerCertificate=yes;"
-        )
-
-        auth = self.get_auth_type()
-
-        if auth == AuthType.WINDOWS:
-            return base + "Trusted_Connection=yes;"
-        elif auth == AuthType.SQL_SERVER:
-            return base + f"UID={self.username};PWD={self.password};"
-        elif auth == AuthType.AD_PASSWORD:
-            return base + f"Authentication=ActiveDirectoryPassword;" f"UID={self.username};PWD={self.password};"
-        elif auth == AuthType.AD_INTERACTIVE:
-            return base + f"Authentication=ActiveDirectoryInteractive;" f"UID={self.username};"
-        elif auth == AuthType.AD_INTEGRATED:
-            return base + "Authentication=ActiveDirectoryIntegrated;"
-
-        return base + "Trusted_Connection=yes;"
-
-    def get_display_info(self) -> str:
-        """Get a display string for the connection."""
-        from .db.providers import is_file_based
-        if is_file_based(self.db_type):
-            return self.file_path or self.name
-
-        if self.db_type == "supabase":
-            return f"{self.name} ({self.supabase_region})"
-
-        db_part = f"@{self.database}" if self.database else ""
-        return f"{self.name}{db_part}"
-
     def get_source_emoji(self) -> str:
         """Get emoji indicator for connection source (e.g., '🐳 ' for docker)."""
         return get_source_emoji(self.source)