fix: close oracle round 13 gaps

Author: yeongseon

docs/DEVELOPMENT.md

@@ -120,7 +120,7 @@ sqlalchemy-excel/
 │   ├── types.py              # ExcelTypeCompiler (type mappings)
 │   ├── reflection.py         # ExcelInspectionMixin (schema inspection)
 │   └── py.typed              # PEP 561 marker file
-├── tests/                    # 18 test modules (400 passed, 2 xfailed)
+├── tests/                    # 18 test modules (402 passed, 2 xfailed)
 │   ├── conftest.py           # Shared fixtures
 │   ├── test_alter_table.py
 │   ├── test_dialect.py

src/sqlalchemy_excel/compiler.py

@@ -979,6 +979,30 @@ def _on_conflict_target(self, clause: Any, **kw: Any) -> str:
 
         return target_text
 
+    @staticmethod
+    def _rewrite_upsert_same_table_column(value: Any, insert_statement: Any) -> Any:
+        if not isinstance(value, elements.ColumnElement):
+            return value
+
+        value_table = getattr(value, "table", None)
+        insert_table = getattr(insert_statement, "table", None)
+        if (
+            value_table is None
+            or insert_table is None
+            or value_table is not insert_table
+        ):
+            return value
+
+        excluded_columns = getattr(insert_statement, "excluded", None)
+        value_key = getattr(value, "key", None)
+        if excluded_columns is None or value_key is None:
+            return value
+
+        if value_key in excluded_columns:
+            return excluded_columns[value_key]
+
+        return value
+
     def visit_on_conflict_do_nothing(self, on_conflict: Any, **kw: Any) -> str:
         target_text = self._on_conflict_target(on_conflict, **kw)
 
@@ -1010,6 +1034,8 @@ def visit_on_conflict_do_update(self, on_conflict: Any, **kw: Any) -> str:
             else:
                 continue
 
+            value = self._rewrite_upsert_same_table_column(value, insert_statement)
+
             if coercions._is_literal(value):
                 value = elements.BindParameter(None, value, type_=c.type)
 

src/sqlalchemy_excel/dialect.py

@@ -709,6 +709,9 @@ def has_table(
         """Check if a worksheet (table) exists."""
         import excel_dbapi
 
+        if schema is not None:
+            return False
+
         raw_conn = connection.connection.dbapi_connection
         return cast("bool", excel_dbapi.has_table(raw_conn, table_name))
 

src/sqlalchemy_excel/reflection.py

@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+import logging
+from contextlib import suppress
 from typing import Any, cast
 
 from sqlalchemy import types as sa_types
@@ -28,6 +30,8 @@
     "TIMESTAMP": sa_types.DateTime,
 }
 
+_LOG = logging.getLogger(__name__)
+
 
 def _sa_type_from_name(type_name: str) -> sa_types.TypeEngine[Any]:
     """Convert an excel-dbapi type name to a SQLAlchemy type instance."""
@@ -51,6 +55,9 @@ def get_table_names(
         """Return all worksheet names, excluding the metadata sheet."""
         import excel_dbapi
 
+        if schema is not None:
+            return []
+
         raw_conn = connection.connection.dbapi_connection
         return cast("list[str]", excel_dbapi.list_tables(raw_conn, include_meta=False))
 
@@ -77,6 +84,9 @@ def get_columns(
         """
         import excel_dbapi
 
+        if schema is not None:
+            raise NoSuchTableError(table_name)
+
         raw_conn = self._raw_connection(connection)
         table_exists = cast("bool", excel_dbapi.has_table(raw_conn, table_name))
         if not table_exists:
@@ -123,6 +133,9 @@ def get_pk_constraint(
         """Return primary key constraint info from the metadata sheet."""
         import excel_dbapi
 
+        if schema is not None:
+            raise NoSuchTableError(table_name)
+
         raw_conn = self._raw_connection(connection)
         table_exists = cast("bool", excel_dbapi.has_table(raw_conn, table_name))
         if not table_exists:
@@ -147,6 +160,8 @@ def get_foreign_keys(
         **kw: Any,
     ) -> list[dict[str, Any]]:
         """Excel does not support foreign keys."""
+        if schema is not None:
+            raise NoSuchTableError(table_name)
         self._assert_table_exists(connection, table_name)
         return []
 
@@ -158,6 +173,8 @@ def get_indexes(
         **kw: Any,
     ) -> list[dict[str, Any]]:
         """Excel does not support indexes."""
+        if schema is not None:
+            raise NoSuchTableError(table_name)
         self._assert_table_exists(connection, table_name)
         return []
 
@@ -169,6 +186,8 @@ def get_unique_constraints(
         **kw: Any,
     ) -> list[dict[str, Any]]:
         """Excel does not support unique constraints."""
+        if schema is not None:
+            raise NoSuchTableError(table_name)
         self._assert_table_exists(connection, table_name)
         return []
 
@@ -180,6 +199,8 @@ def get_check_constraints(
         **kw: Any,
     ) -> list[dict[str, Any]]:
         """Excel does not support check constraints."""
+        if schema is not None:
+            raise NoSuchTableError(table_name)
         self._assert_table_exists(connection, table_name)
         return []
 
@@ -191,6 +212,8 @@ def get_table_comment(
         **kw: Any,
     ) -> dict[str, Any]:
         """Excel does not support table comments."""
+        if schema is not None:
+            raise NoSuchTableError(table_name)
         self._assert_table_exists(connection, table_name)
         return {"text": None}
 
@@ -208,9 +231,33 @@ def _metadata_header_matches(
         header_names: list[str] | None,
     ) -> bool:
         if header_names is None:
-            return True
+            return False
         return [col["name"] for col in meta] == header_names
 
+    @staticmethod
+    def _cursor_header_names(raw_conn: Any, table_name: str) -> list[str] | None:
+        try:
+            cursor = raw_conn.cursor()
+        except Exception:
+            return None
+
+        try:
+            cursor.execute(f"SELECT * FROM {table_name} LIMIT 0")
+            description = getattr(cursor, "description", None)
+            if not description:
+                return None
+            header_names = [
+                str(column[0])
+                for column in description
+                if column is not None and column[0] is not None
+            ]
+            return header_names or None
+        except Exception:
+            return None
+        finally:
+            with suppress(Exception):
+                cursor.close()
+
     def _validated_metadata(
         self,
         raw_conn: Any,
@@ -223,6 +270,15 @@ def _validated_metadata(
         if meta is None:
             return None
 
+        if header_names is None:
+            header_names = self._cursor_header_names(raw_conn, table_name)
+            if header_names is None:
+                _LOG.warning(
+                    "Could not validate metadata headers for table '%s'; using stored metadata",
+                    table_name,
+                )
+                return meta
+
         if not self._metadata_header_matches(meta, header_names):
             excel_dbapi.remove_table_metadata(raw_conn, table_name)
             return None

tests/test_reflection.py

@@ -58,6 +58,10 @@ def test_metadata_sheet_excluded(self, populated_engine):
         tables = insp.get_table_names()
         assert "__excel_meta__" not in tables
 
+    def test_nonexistent_schema_returns_no_tables(self, populated_engine):
+        insp = inspect(populated_engine)
+        assert insp.get_table_names(schema="nonexistent") == []
+
 
 class TestHasTable:
     """Test has_table."""
@@ -70,6 +74,10 @@ def test_nonexistent_table(self, populated_engine):
         insp = inspect(populated_engine)
         assert not insp.has_table("nonexistent")
 
+    def test_schema_argument_returns_false(self, populated_engine):
+        insp = inspect(populated_engine)
+        assert not insp.has_table("users", schema="nonexistent")
+
 
 class TestGetColumns:
     """Test get_columns — from metadata sheet."""

tests/test_reflection_full.py

@@ -107,6 +107,60 @@ def test_reflection_rebuilds_columns_when_metadata_headers_are_stale(tmp_path) -
     engine.dispose()
 
 
+def test_reflection_validates_metadata_with_cursor_when_headers_unavailable(
+    tmp_path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    engine = _engine_for(tmp_path)
+
+    with engine.begin() as conn:
+        conn.execute(text("CREATE TABLE users (id INTEGER, name TEXT)"))
+
+    with engine.connect() as conn:
+        import excel_dbapi
+
+        raw_conn = conn.connection.dbapi_connection
+        excel_dbapi.write_table_metadata(
+            raw_conn,
+            "users",
+            [
+                {
+                    "name": "old_id",
+                    "type_name": "INTEGER",
+                    "nullable": True,
+                    "primary_key": False,
+                },
+                {
+                    "name": "old_name",
+                    "type_name": "TEXT",
+                    "nullable": True,
+                    "primary_key": False,
+                },
+            ],
+        )
+
+    from sqlalchemy_excel.reflection import ExcelInspectionMixin
+
+    monkeypatch.setattr(
+        ExcelInspectionMixin,
+        "_worksheet_header_names",
+        staticmethod(lambda _raw_conn, _table_name: None),
+    )
+
+    inspector = inspect(engine)
+    columns = inspector.get_columns("users")
+    assert [column["name"] for column in columns] == ["id", "name"]
+
+    with engine.connect() as conn:
+        import excel_dbapi
+
+        raw_conn = conn.connection.dbapi_connection
+        metadata = excel_dbapi.read_table_metadata(raw_conn, "users")
+        assert metadata is not None
+        assert [column["name"] for column in metadata] == ["id", "name"]
+
+    engine.dispose()
+
+
 def test_table_scoped_reflection_methods_raise_for_missing_table(tmp_path) -> None:
     engine = _engine_for(tmp_path)
     inspector = inspect(engine)
@@ -123,3 +177,20 @@ def test_table_scoped_reflection_methods_raise_for_missing_table(tmp_path) -> No
         inspector.get_table_comment("missing")
 
     engine.dispose()
+
+
+def test_table_scoped_reflection_methods_raise_for_non_default_schema(tmp_path) -> None:
+    engine = _engine_for(tmp_path)
+
+    with engine.begin() as conn:
+        conn.execute(text("CREATE TABLE users (id INTEGER, name TEXT)"))
+
+    inspector = inspect(engine)
+    with pytest.raises(exc.NoSuchTableError):
+        inspector.get_columns("users", schema="nonexistent")
+    with pytest.raises(exc.NoSuchTableError):
+        inspector.get_pk_constraint("users", schema="nonexistent")
+    with pytest.raises(exc.NoSuchTableError):
+        inspector.get_foreign_keys("users", schema="nonexistent")
+
+    engine.dispose()

tests/test_upsert.py

@@ -160,15 +160,34 @@ def test_expression_set_value(self, populated_engine, items_table):
             assert rows == [(1, "Alice", 31, "A")]
 
     def test_set_with_table_columns(self, populated_engine, items_table):
-        upsert_stmt = insert(items_table).on_conflict_do_update(
-            index_elements=["id"],
-            set_=items_table.c,
+        with populated_engine.begin() as conn:
+            conn.execute(
+                insert(items_table).values(id=1, name="Alice", age=30, code="A")
+            )
+
+        upsert_stmt = (
+            insert(items_table)
+            .values(id=1, name="Renamed", age=41, code="B")
+            .on_conflict_do_update(
+                index_elements=["id"],
+                set_=items_table.c,
+            )
         )
         compiled = " ".join(
             str(upsert_stmt.compile(dialect=populated_engine.dialect)).split()
         )
         assert "items." not in compiled
-        assert " SET id = id, name = name, age = age, code = code" in compiled
+        assert (
+            " SET id = excluded.id, name = excluded.name, age = excluded.age,"
+            " code = excluded.code"
+        ) in compiled
+
+        with populated_engine.begin() as conn:
+            conn.execute(upsert_stmt)
+
+        with populated_engine.connect() as conn:
+            rows = conn.execute(select(items_table).order_by(items_table.c.id)).all()
+            assert rows == [(1, "Renamed", 41, "B")]
 
     def test_multi_column_conflict_target(
         self, populated_engine, composite_items_table