Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGES.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@ Unreleased
- include identity column info (#297), and
- avoid parse error when reflecting ENUMs (#303).
(CRDB 26.3+ required for full compatibility.)
- `get_table_names()` now returns base tables only; views were also listed
and are reported by `get_view_names()`. `has_table()` still returns True for
views, as in SQLAlchemy 2.0 (#310).

# Version 2.0.4
April 23, 2026
Expand Down
19 changes: 13 additions & 6 deletions sqlalchemy_cockroachdb/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,18 +127,25 @@ def get_table_names(self, conn, schema=None, **kw):
# v1.1 or earlier.
return [row.Table for row in conn.execute(text("SHOW TABLES"))]

# v2.0+ have a good information schema. Use it.
# v2.0+ have a good information schema. Use it. Only base tables are
# returned; views are reported by get_view_names().
return self._information_schema_relations(conn, schema, base_tables_only=True)

def _information_schema_relations(self, conn, schema, base_tables_only):
query = "SELECT table_name FROM information_schema.tables WHERE table_schema=:schema"
if base_tables_only:
query += " AND table_type = 'BASE TABLE'"
return [
row.table_name
for row in conn.execute(
text("SELECT table_name FROM information_schema.tables WHERE table_schema=:schema"),
{"schema": schema or self.default_schema_name},
)
for row in conn.execute(text(query), {"schema": schema or self.default_schema_name})
]

def has_table(self, conn, table, schema=None, info_cache=None):
# Upstream implementation needs pg_table_is_visible().
return any(t == table for t in self.get_table_names(conn, schema=schema))
# As in SQLAlchemy 2.0, has_table() is also true for views.
if not self._is_v2plus:
return any(t == table for t in self.get_table_names(conn, schema=schema))
return table in self._information_schema_relations(conn, schema, base_tables_only=False)

def get_multi_columns(self, connection, schema, filter_names, scope, kind, **kw):
_include_hidden = kw.get("include_hidden", False)
Expand Down
28 changes: 28 additions & 0 deletions test/test_introspection.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
UniqueConstraint,
CheckConstraint,
text,
inspect,
)
from sqlalchemy.types import Integer, String, Boolean
import sqlalchemy.types as sqltypes
Expand Down Expand Up @@ -166,3 +167,30 @@ def test_varchar(self):
]
for t in types:
self._test(t, sqltypes.VARCHAR)


class TableNamesTest(fixtures.TestBase):
__requires__ = ("sync_driver",)

def setup_method(self):
with testing.db.begin() as conn:
conn.execute(text("CREATE TABLE names_base (id INT PRIMARY KEY)"))
conn.execute(text("CREATE VIEW names_view AS SELECT id FROM names_base"))

def teardown_method(self, method):
with testing.db.begin() as conn:
conn.execute(text("DROP VIEW IF EXISTS names_view"))
conn.execute(text("DROP TABLE IF EXISTS names_base"))

def test_get_table_names_excludes_views(self):
insp = inspect(testing.db)
table_names = insp.get_table_names()
assert "names_base" in table_names
assert "names_view" not in table_names
assert "names_view" in insp.get_view_names()

def test_has_table_includes_views(self):
insp = inspect(testing.db)
assert insp.has_table("names_base")
assert insp.has_table("names_view")
assert not insp.has_table("names_absent")