feat: implement ALTER TABLE (ADD/DROP/RENAME COLUMN) in SQLAlchemy dialect

Author: yeongseon

src/sqlalchemy_excel/ddl.py

@@ -2,7 +2,7 @@
 
 CREATE TABLE → creates a worksheet and writes column metadata.
 DROP TABLE   → deletes the worksheet and removes metadata.
-ALTER TABLE  → rejected (not supported).
+ALTER TABLE  → supports ADD/DROP/RENAME COLUMN.
 """
 
 from __future__ import annotations
@@ -16,6 +16,26 @@
 class ExcelDDLCompiler(compiler.DDLCompiler):
     """Compiles DDL statements for excel-dbapi."""
 
+    def _format_alter_table_name(self, operation: Any) -> str:
+        schema = getattr(operation, "schema", None)
+        if schema is not None:
+            raise exc.CompileError("Excel dialect does not support schemas")
+
+        table_name = getattr(operation, "table_name", None)
+        if table_name is None:
+            table = getattr(operation, "table", None)
+            table_name = getattr(table, "name", None)
+
+        if not isinstance(table_name, str) or not table_name:
+            raise exc.CompileError("ALTER TABLE operation requires a table name")
+
+        return self.preparer.quote(table_name)
+
+    def _format_alter_column_name(self, column_name: Any) -> str:
+        if not isinstance(column_name, str) or not column_name:
+            raise exc.CompileError("ALTER TABLE operation requires a column name")
+        return self.preparer.quote(column_name)
+
     def visit_create_table(self, create: Any, **kw: Any) -> str:
         """Compile CREATE TABLE into SQL that excel-dbapi's parser accepts.
 
@@ -38,6 +58,46 @@ def visit_drop_table(self, drop: Any, **kw: Any) -> str:
         table_name = self.preparer.format_table(table)
         return f"DROP TABLE {table_name}"
 
+    def visit_add_column(self, create: Any, **kw: Any) -> str:
+        table_name = self._format_alter_table_name(create)
+        column = getattr(create, "column", None)
+        if column is None:
+            raise exc.CompileError("ALTER TABLE ADD COLUMN requires a column")
+
+        col_name = self.preparer.format_column(column)
+        col_type = self.dialect.type_compiler.process(column.type)
+        return f"ALTER TABLE {table_name} ADD COLUMN {col_name} {col_type}"
+
+    def visit_drop_column(self, drop: Any, **kw: Any) -> str:
+        table_name = self._format_alter_table_name(drop)
+
+        column_name = getattr(drop, "column_name", None)
+        if column_name is None:
+            column = getattr(drop, "column", None)
+            column_name = getattr(column, "name", column)
+
+        column_name = self._format_alter_column_name(column_name)
+        return f"ALTER TABLE {table_name} DROP COLUMN {column_name}"
+
+    def visit_rename_column(self, rename: Any, **kw: Any) -> str:
+        table_name = self._format_alter_table_name(rename)
+
+        old_column_name = getattr(rename, "column_name", None)
+        if old_column_name is None:
+            old_column_name = getattr(rename, "old_column_name", None)
+        if old_column_name is None:
+            old_column_name = getattr(rename, "old_name", None)
+
+        new_column_name = getattr(rename, "new_column_name", None)
+        if new_column_name is None:
+            new_column_name = getattr(rename, "name", None)
+        if new_column_name is None:
+            new_column_name = getattr(rename, "new_name", None)
+
+        old_name = self._format_alter_column_name(old_column_name)
+        new_name = self._format_alter_column_name(new_column_name)
+        return f"ALTER TABLE {table_name} RENAME COLUMN {old_name} TO {new_name}"
+
     def visit_create_index(
         self,
         create: Any,

src/sqlalchemy_excel/dialect.py

@@ -91,7 +91,7 @@ class ExcelDialect(  # type: ignore[misc]  # pyright: ignore[reportIncompatibleM
     default_paramstyle: str = "qmark"
 
     # ── Feature flags ──────────────────────────────────────
-    supports_alter: bool = False
+    supports_alter: bool = True
     supports_sequences: bool = False
     supports_schemas: bool = False
     supports_views: bool = False
@@ -173,6 +173,7 @@ def do_execute(
         """Execute a statement, normalizing whitespace for excel-dbapi."""
         normalized = re.sub(r"\s+", " ", statement).strip()
         cursor.execute(normalized, parameters)
+        self._sync_alter_table_metadata(cursor, normalized)
 
     def do_execute_no_params(
         self,
@@ -183,6 +184,66 @@ def do_execute_no_params(
         """Execute a statement with no parameters."""
         normalized = re.sub(r"\s+", " ", statement).strip()
         cursor.execute(normalized, None)
+        self._sync_alter_table_metadata(cursor, normalized)
+
+    def _sync_alter_table_metadata(self, cursor: Any, statement: str) -> None:
+        if not statement.upper().startswith("ALTER TABLE "):
+            return
+
+        import excel_dbapi
+
+        tokens = statement.split()
+        if len(tokens) < 6:
+            return
+
+        table_name = tokens[2].strip('"')
+        operation = tokens[3].upper()
+
+        raw_conn = cursor.connection
+        current_meta = excel_dbapi.read_table_metadata(raw_conn, table_name) or []
+
+        type_map = {col["name"]: col["type_name"] for col in current_meta}
+        nullable_map = {col["name"]: col.get("nullable", True) for col in current_meta}
+        pk_map = {col["name"]: col.get("primary_key", False) for col in current_meta}
+
+        if operation == "ADD" and len(tokens) == 7 and tokens[4].upper() == "COLUMN":
+            added_type = tokens[6].upper()
+            if added_type == "FLOAT":
+                added_type = "REAL"
+            type_map[tokens[5].strip('"')] = added_type
+
+        if operation == "DROP" and len(tokens) == 6 and tokens[4].upper() == "COLUMN":
+            removed = tokens[5].strip('"')
+            type_map.pop(removed, None)
+            nullable_map.pop(removed, None)
+            pk_map.pop(removed, None)
+
+        if (
+            operation == "RENAME"
+            and len(tokens) == 8
+            and tokens[4].upper() == "COLUMN"
+            and tokens[6].upper() == "TO"
+        ):
+            old_name = tokens[5].strip('"')
+            new_name = tokens[7].strip('"')
+            if old_name in type_map:
+                type_map[new_name] = type_map.pop(old_name)
+            if old_name in nullable_map:
+                nullable_map[new_name] = nullable_map.pop(old_name)
+            if old_name in pk_map:
+                pk_map[new_name] = pk_map.pop(old_name)
+
+        live_columns = excel_dbapi.get_columns(raw_conn, table_name)
+        columns = [
+            {
+                "name": col["name"],
+                "type_name": type_map.get(col["name"], col.get("type", "TEXT")),
+                "nullable": nullable_map.get(col["name"], True),
+                "primary_key": pk_map.get(col["name"], False),
+            }
+            for col in live_columns
+        ]
+        excel_dbapi.write_table_metadata(raw_conn, table_name, columns)
 
     def do_ping(self, dbapi_connection: Any) -> bool:
         """Ping the connection by verifying it's not closed."""

tests/test_alter_table.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import Column, Integer, MetaData, String, Table, inspect
+from sqlalchemy.sql.ddl import ExecutableDDLElement
+
+
+class AddColumn(ExecutableDDLElement):
+    __visit_name__ = "add_column"
+
+    def __init__(self, table_name: str, column: Column[Any]) -> None:
+        self.table_name = table_name
+        self.column = column
+        self.schema = None
+
+
+class DropColumn(ExecutableDDLElement):
+    __visit_name__ = "drop_column"
+
+    def __init__(self, table_name: str, column_name: str) -> None:
+        self.table_name = table_name
+        self.column_name = column_name
+        self.schema = None
+
+
+class RenameColumn(ExecutableDDLElement):
+    __visit_name__ = "rename_column"
+
+    def __init__(
+        self,
+        table_name: str,
+        old_column_name: str,
+        new_column_name: str,
+    ) -> None:
+        self.table_name = table_name
+        self.column_name = old_column_name
+        self.new_column_name = new_column_name
+        self.schema = None
+
+
+def _create_users_table(metadata: MetaData) -> Table:
+    return Table(
+        "users",
+        metadata,
+        Column("id", Integer, primary_key=True),
+        Column("name", String),
+        Column("age", Integer),
+    )
+
+
+def test_alter_table_add_column(engine) -> None:
+    metadata = MetaData()
+    _create_users_table(metadata)
+    metadata.create_all(engine)
+
+    ddl = AddColumn("users", Column("email", String))
+    compiled = str(ddl.compile(dialect=engine.dialect)).strip()
+    assert compiled == "ALTER TABLE users ADD COLUMN email TEXT"
+
+    with engine.begin() as conn:
+        conn.execute(ddl)
+
+    columns = [col["name"] for col in inspect(engine).get_columns("users")]
+    assert columns == ["id", "name", "age", "email"]
+
+
+def test_alter_table_drop_column(engine) -> None:
+    metadata = MetaData()
+    _create_users_table(metadata)
+    metadata.create_all(engine)
+
+    ddl = DropColumn("users", "age")
+    compiled = str(ddl.compile(dialect=engine.dialect)).strip()
+    assert compiled == "ALTER TABLE users DROP COLUMN age"
+
+    with engine.begin() as conn:
+        conn.execute(ddl)
+
+    columns = [col["name"] for col in inspect(engine).get_columns("users")]
+    assert columns == ["id", "name"]
+
+
+def test_alter_table_rename_column(engine) -> None:
+    metadata = MetaData()
+    _create_users_table(metadata)
+    metadata.create_all(engine)
+
+    ddl = RenameColumn("users", "name", "full_name")
+    compiled = str(ddl.compile(dialect=engine.dialect)).strip()
+    assert compiled == "ALTER TABLE users RENAME COLUMN name TO full_name"
+
+    with engine.begin() as conn:
+        conn.execute(ddl)
+
+    columns = [col["name"] for col in inspect(engine).get_columns("users")]
+    assert columns == ["id", "full_name", "age"]

tests/test_dialect.py

@@ -39,8 +39,8 @@ def test_driver(self, engine):
     def test_paramstyle(self, engine):
         assert engine.dialect.default_paramstyle == "qmark"
 
-    def test_no_alter(self, engine):
-        assert engine.dialect.supports_alter is False
+    def test_supports_alter(self, engine):
+        assert engine.dialect.supports_alter is True
 
     def test_no_sequences(self, engine):
         assert engine.dialect.supports_sequences is False