Account for `schema` in `table()` `fullname` attribute.

Fixes: #7061
Change-Id: I715da89591c03d40d77734473bd42b34b9c4e1dc

diff --git a/doc/build/changelog/unreleased_14/7061.rst b/doc/build/changelog/unreleased_14/7061.rst
new file mode 100644 (file)
index 0000000..a7a184b
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7061.rst
@@ -0,0 +1,5 @@
+.. change::
+    :tags: bug, sql
+    :tickets: 7061
+
+    Account for ``schema`` in :func:`_sql.table` ``fullname`` attribute.

diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 970c7a0c567b636fff24345346065dbd5988c4b4..8f8e6b2a7e15526d191a3909fb161087929921d4 100644 (file)
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2622,7 +2622,7 @@ class TableClause(roles.DMLTableRole, Immutable, FromClause):
         """
 
         super(TableClause, self).__init__()
-        self.name = self.fullname = name
+        self.name = name
         self._columns = DedupeColumnCollection()
         self.primary_key = ColumnSet()
         self.foreign_keys = set()
@@ -2632,6 +2632,10 @@ class TableClause(roles.DMLTableRole, Immutable, FromClause):
         schema = kw.pop("schema", None)
         if schema is not None:
             self.schema = schema
+        if self.schema is not None:
+            self.fullname = "%s.%s" % (self.schema, self.name)
+        else:
+            self.fullname = self.name
         if kw:
             raise exc.ArgumentError("Unsupported argument(s): %s" % list(kw))
 

diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index b76873490462be9928b49466dc22cf49ba59677d..2157b5c718a7561ca18e56bb727b4e9155e4d818 100644 (file)
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1469,6 +1469,14 @@ class SelectableTest(
         assert s3._whereclause.left.table is not s1
         assert s3._whereclause.left.table in froms
 
+    def test_table_schema(self):
+        t = table("foo")
+        eq_(t.name, "foo")
+        eq_(t.fullname, "foo")
+        t = table("foo", schema="bar")
+        eq_(t.name, "foo")
+        eq_(t.fullname, "bar.foo")
+
 
 class RefreshForNewColTest(fixtures.TestBase):
     def test_join_uninit(self):